import datetime
import requests, re
from bs4 import BeautifulSoup
import os
class ScraperError(Exception):
"""A cm99.net response did not contain the field we expected.
The raw response is saved to logs/scraper-failures/ before this is
raised; the message identifies which method failed and what was
being looked for.
"""
class CM_BOT:
def __init__(self):
self.session = requests.Session()
self.base_url = self._get_required_env('CM_BOT_BASE_URL')
self.is_logged_in = False
self._setup_headers()
def _get_required_env(self, name: str) -> str:
value = os.getenv(name)
if value is None or value == "":
raise RuntimeError(f"Missing required environment variable: {name}")
return value
def _setup_headers(self):
"""Set up default headers for requests."""
self.login_headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.8',
'cache-control': 'max-age=0',
'content-type': 'application/x-www-form-urlencoded',
'origin': self.base_url,
'referer': f'{self.base_url}/cm/login',
'sec-ch-ua': '"Not;A=Brand";v="99", "Brave";v="139", "Chromium";v="139"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'same-origin',
'sec-fetch-user': '?1',
'sec-gpc': '1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36'
}
self.get_user_tree_headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.8',
'priority': 'u=0, i',
'sec-ch-ua': '"Not;A=Brand";v="99", "Brave";v="139", "Chromium";v="139"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'sec-gpc': '1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36'
}
self.get_register_form_headers = {
'accept': 'text/html, */*; q=0.01',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.8',
'content-length': '0',
'origin': self.base_url,
'priority': 'u=1, i',
'referer': f'{self.base_url}/cm/userMainAction',
'sec-ch-ua': '"Not;A=Brand";v="99", "Brave";v="139", "Chromium";v="139"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'sec-gpc': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
'x-requested-with': 'XMLHttpRequest'
}
self.get_user_credit_headers = {
'accept': 'text/html, */*; q=0.01',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.7',
'content-length': '0',
'origin': self.base_url,
'priority': 'u=1, i',
'referer': f'{self.base_url}/cm/mainMenu',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Brave";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'sec-gpc': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
'x-requested-with': 'XMLHttpRequest'
}
# self.change_pass_headers = {
# 'accept': '*/*',
# 'accept-encoding': 'identity',
# 'accept-language': 'en-GB,en;q=0.8',
# 'content-length': '889',
# 'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
# 'origin': self.base_url,
# 'priority': 'u=1, i',
# 'referer': f'{self.base_url}/cm/userMainAction',
# 'sec-ch-ua': '"Not;A=Brand";v="99", "Brave";v="139", "Chromium";v="139"',
# 'sec-ch-ua-mobile': '?0',
# 'sec-ch-ua-platform': '"macOS"',
# 'sec-fetch-dest': 'empty',
# 'sec-fetch-mode': 'cors',
# 'sec-fetch-site': 'same-origin',
# 'sec-gpc': '1',
# 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
# 'x-requested-with': 'XMLHttpRequest'
# }
self.register_form_headers = {
'accept': '*/*',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.8',
'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
'origin': self.base_url,
'priority': 'u=1, i',
'referer': f'{self.base_url}/cm/userMainAction',
'sec-ch-ua': '"Not;A=Brand";v="99", "Brave";v="139", "Chromium";v="139"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'sec-gpc': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
'x-requested-with': 'XMLHttpRequest'
}
self.set_security_pin_headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.9',
'cache-control': 'max-age=0',
'content-length': '103',
'content-type': 'application/x-www-form-urlencoded',
'origin': self.base_url,
'priority': 'u=0, i',
'referer': f'{self.base_url}/cm/setSecurityPin',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Brave";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'same-origin',
'sec-fetch-user': '?1',
'sec-gpc': '1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
}
self.transfer_search_headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.7',
'cache-control': 'max-age=0',
'content-length': '78',
'content-type': 'application/x-www-form-urlencoded',
'origin': self.base_url,
'priority': 'u=0, i',
'referer': f'{self.base_url}/cm/transfer',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Brave";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'same-origin',
'sec-fetch-user': '?1',
'sec-gpc': '1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
}
self.transfer_credit_headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-encoding': 'identity',
'accept-language': 'en-GB,en;q=0.7',
'cache-control': 'max-age=0',
'content-length': '152',
'content-type': 'application/x-www-form-urlencoded',
'origin': self.base_url,
'priority': 'u=0, i',
'referer': f'{self.base_url}/cm/searchTransferUser',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Brave";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'same-origin',
'sec-fetch-user': '?1',
'sec-gpc': '1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
}
def _dump_html(self, context: str, content) -> str:
"""Save a failing cm99.net response to logs/scraper-failures/.
Returns the path written to so callers can include it in error
messages.
"""
ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
out_dir = os.path.join("logs", "scraper-failures")
os.makedirs(out_dir, exist_ok=True)
path = os.path.join(out_dir, f"{context}-{ts}.html")
if isinstance(content, (bytes, bytearray)):
data = bytes(content)
else:
data = str(content).encode("utf-8", "replace")
with open(path, "wb") as f:
f.write(data)
print(f"[scraper-failure] dumped {context} response to {path}")
return path
def _find_input_value(self, soup, ident: str, *, context: str, raw, by: str = "name") -> str:
"""Extract 's value or raise ScraperError.
`by` selects between matching (default) and
. Saves the raw response to logs/scraper-failures/
before raising so the operator can postmortem.
"""
el = soup.find("input", {by: ident})
if el is None or "value" not in el.attrs:
path = self._dump_html(context, raw)
raise ScraperError(
f"{context}: input[{by}={ident!r}] missing or has no value attribute "
f"(response saved to {path})"
)
return el["value"]
def get_register_data(self, token: str, username: str, password: str):
return {
'struts.token.name': 'token',
'token': f'{token}',
'userIsNew': 'true',
'searchContent': '',
'user.username': f'{username}',
'user.companyId': '1',
'user.userRoleId': '4',
'user.password': f'{password}',
'user.confirmPassword': f'{password}',
'user.name': f'{username}',
'user.mobileNumber': '',
'user.email': '',
'user.remarks': '',
'user.accountStatus': 'A',
'user.currencyMYR': 'true',
'__checkbox_user.currencyMYR': 'true',
'user.allowBetHl': 'true',
'__checkbox_user.allowBetHl': 'true',
'user.commission3d4d': '5.00',
'user.commission5d6d': '5.00',
'user.commissionHL': '19.00',
'user.commissionHL6d': '19.00',
'user.commissionNL': '19.00',
'user.commissionNL6d': '19.00',
'checkAllPrizePackages': 'on',
'selectedPrizePackageList': ['6', '1', '5', '3'],
'__multiselect_selectedPrizePackageList': '',
'checkAllPrizePackages5D6D': 'on',
'selectedPrizePackage5D6DList': '2',
'__multiselect_selectedPrizePackage5D6DList': ''
}
def get_security_pin_data(self, token: str, security_pin: str):
return {
'struts.token.name': 'token',
'token': token,
'newPin': security_pin,
'confirmNewPin': security_pin
}
def get_transfer_search_data(self, token: str, username: str):
return {
'struts.token.name': 'token',
'token': token,
'username': username
}
def get_transfer_data(self, token: str, username: str, name: str, toUserId: str, amount: float, security_pin: str):
return {
'struts.token.name': 'token',
'token': token,
'username': username,
'name': name,
'toUserId': toUserId,
'amount': amount,
'securityPin': security_pin
}
# def get_change_pass_data(self, token: str, user_encrypted_id: str, username: str, password: str):
# return {
# 'struts.token.name': 'token',
# 'token': f'{token}',
# 'searchContent': '',
# 'userIsNew': 'false',
# 'user.encryptedId': f'{user_encrypted_id}',
# 'user.username': f'{username}',
# 'user.parentId': '31308',
# 'user.companyId': '1',
# 'user.userRoleId': '4',
# 'user.password': f'{password}',
# 'user.confirmPassword': f'{password}',
# 'user.name': f'{username}',
# 'user.mobileNumber': '',
# 'user.email': '',
# 'user.remarks': '',
# 'user.accountStatus': 'A',
# 'user.currencyMYR': 'true',
# '__checkbox_user.currencyMYR': 'true',
# 'user.allowBetHl': 'true',
# '__checkbox_user.allowBetHl': 'true',
# 'user.balance': '1',
# 'user.outstanding': '1',
# 'user.commission3d4d': '5.00',
# 'user.commission5d6d': '5.00',
# 'user.commissionHL': '19.00',
# 'user.commissionHL6d': '19.00',
# 'user.commissionNL': '19.00',
# 'user.commissionNL6d': '19.00',
# 'selectedPrizePackageList': ['6', '1', '5', '3'],
# '__multiselect_selectedPrizePackageList': '',
# 'selectedPrizePackage5D6DList': '2',
# '__multiselect_selectedPrizePackage5D6DList': ''
# }
def login(self, username: str, password: str):
try:
# print("Starting login process...")
login_page = self.session.get(f'{self.base_url}/cm/login')
# print(f"Login page status: {login_page.status_code}")
login_data = {
'j_username': username,
'j_password': password
}
login_response = self.session.post(
f'{self.base_url}/cm/j_security_check',
data=login_data,
headers=self.login_headers,
allow_redirects=True
)
if login_response.status_code == 200 and 'login' not in login_response.url.lower():
# print("Login successful!")
self.is_logged_in = True
return True
else:
print("Login failed!")
self.is_logged_in = False
return False
except requests.exceptions.RequestException as e:
print(f"Error during login: {e}")
self.is_logged_in = False
return False
def get_max_user_in_pattern(self, prefix_pattern: str = None):
if prefix_pattern is None:
prefix_pattern = self._get_required_env("CM_PREFIX_PATTERN")
response = self.session.get(f'{self.base_url}/cm/json/generateUserTree?id=0')
regex = f'\\[{prefix_pattern}\\d+]'
matches = re.findall(regex, response.text)
last_match_text = matches[-1]
number_part = last_match_text.replace(f"[{prefix_pattern}", "").replace("]", "")
last_match = int(number_part)
return last_match
def get_register_form_token(self):
try:
response = self.session.post(
f'{self.base_url}/cm/loadUserAccount',
headers=self.get_register_form_headers
)
soup = BeautifulSoup(response.content, 'html.parser')
return self._find_input_value(
soup, "token",
context="register_form_token",
raw=response.content,
)
except requests.exceptions.RequestException as e:
print(f"Error getting register form: {e}")
return None
def get_security_pin_form_token(self):
response = self.session.get(f'{self.base_url}/cm/setSecurityPin')
soup = BeautifulSoup(response.content, 'html.parser')
return self._find_input_value(
soup, "token",
context="security_pin_form_token",
raw=response.content,
)
def register_user(self, user_id, user_password):
try:
print("Creating user account...")
token = self.get_register_form_token()
user_data = self.get_register_data(token, user_id, user_password)
response = self.session.post(
f'{self.base_url}/cm/saveUserAccount',
data=user_data,
headers=self.register_form_headers
)
if re.search('User created successfully', response.text):
print(f"User account: {user_id} password: {user_password} creation completed!")
return True
else:
print(f"User account: {user_id} creation FAIL!")
return False
except requests.exceptions.RequestException as e:
print(f"Error creating user account: {e}")
return False
# def change_user_password(self, user_id: str, new_user_pass: str):
# try:
# print(f"Changing user: {user_id} password...")
# self.login(self.username, self.password)
# token = self.get_register_form_token()
# headers = self.change_pass_headers
# user_data = self.get_change_pass_data(token, )
# response = self.session.post(
# f'{self.base_url}/cm/saveUserAccount',
# data=user_data,
# headrs=headers
# )
# except requests.exceptions.RequestException as e:
# print(f"Error change user password: {e}")
# return None
def get_register_link(self):
response = self.session.get(f"{self.base_url}/cm/showQrCode")
soup = BeautifulSoup(response.content, 'html.parser')
soup = soup.find('form', {'id': 'qrCodeForm'})
return soup.find('a')['href']
def get_generate_username(self, max_username_index: int):
max_username_index += 1
if max_username_index % 10 == 4:
max_username_index += 1
return max_username_index
def set_security_pin(self, security_pin: str):
token = self.get_security_pin_form_token()
security_data = self.get_security_pin_data(token, security_pin)
response = self.session.post(
f'{self.base_url}/cm/saveSecurityPin',
data=security_data,
headers=self.set_security_pin_headers
)
if 'java.lang.NullPointerException' in response.text:
return False
return True
def transfer_credit(self, t_username: str, t_password: str, amount: float):
token = self.get_transfer_token()
transfer_search_data = self.get_transfer_search_data(token, t_username)
response = self.session.post(
f'{self.base_url}/cm/searchTransferUser',
data=transfer_search_data,
headers=self.transfer_search_headers
)
soup = BeautifulSoup(response.content, 'html.parser')
name = self._find_input_value(
soup, "name",
context="transfer_search_name",
raw=response.content,
by="id",
)
token = self._find_input_value(
soup, "token",
context="transfer_search_token",
raw=response.content,
)
toUserId = self._find_input_value(
soup, "toUserId",
context="transfer_search_toUserId",
raw=response.content,
by="id",
)
transfer_data = self.get_transfer_data(token, t_username, name, toUserId, amount, t_password)
response = self.session.post(
f'{self.base_url}/cm/saveTransfer',
data=transfer_data,
headers=self.transfer_credit_headers
)
return True if re.search(r'Successfully saved the record\.', response.text) else False
def get_user_credit(self):
response = self.session.post(
f'{self.base_url}/cm/userProfile',
headers=self.get_user_credit_headers
)
soup = BeautifulSoup(response.content, 'html.parser')
try:
return round(float(soup.find('table', {'class': 'generalContent'}).find(text=re.compile('Credit Available')).parent.parent.find_all('td')[2].text.replace(",","")), 2)
except:
print(f"Error getting credit.")
now = datetime.datetime.now().strftime('%Y%m%d_%H%M')
# with open(f'credit-{now}.html', 'wb') as f:
# f.write(response.content)
return 0
def get_transfer_token(self):
response = self.session.get(f'{self.base_url}/cm/transfer')
soup = BeautifulSoup(response.content, 'html.parser')
return self._find_input_value(
soup, "token",
context="transfer_token",
raw=response.content,
)
def logout(self):
"""Logout from the system."""
self.session.close()
self.is_logged_in = False
print("Logged out successfully.")
def main():
print("CM_BOT helper module. Use from service entrypoints instead of running direct debug actions.")
if __name__ == "__main__":
main()