yiekheng/cm_bot_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
cm_bot_v2/app/cm_bot.py
yiekheng 45303d00aa Refactor: externalize all hardcoded config to env vars, add multi-deployment support 
- Remove all hardcoded credentials and config from Python source code:
  - db.py: DB host/user/password/name/port → env vars with connection retry support
  - cm_bot_hal.py: prefix, agent_id, agent_password, security_pin → env vars
  - cm_bot.py: base_url → env var, fix register_user return values
  - cm_web_view.py: hardcoded '13c' prefix → configurable CM_PREFIX_PATTERN
  - cm_telegram.py: hardcoded 'Sky533535' pin → env var CM_SECURITY_PIN

- Parameterize docker-compose.yml for multi-deployment on same host:
  - Container names use ${CM_DEPLOY_NAME} prefix (e.g. rex-cm-*, siong-cm-*)
  - Network name uses ${CM_DEPLOY_NAME}-network
  - Web view port configurable via ${CM_WEB_HOST_PORT}
  - All service config passed as env vars (not baked into image)

- Add per-deployment env configs:
  - envs/rex/.env (port 8001, prefix 13c, DB rex_cm)
  - envs/siong/.env (port 8005, prefix 13sa, DB siong_cm)
  - .env.example as template for new deployments
  - Remove .env from .gitignore (local server, safe to commit)

- Improve telegram bot reliability:
  - Add retry logic for polling with exponential backoff
  - Add error handlers for Conflict, RetryAfter, NetworkError, TimedOut
  - Add /9 command to show chat ID
  - Add telegram_notifier.py for alert notifications
  - Fix error handling in /2 and /3 command handlers

- Fix db.py cursor cleanup (close cursor before connection in finally blocks)
- Fix docker-compose.override.yml environment syntax (list → mapping)
- Update README with multi-deployment instructions
- Add AGENTS.md

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-22 22:25:40 +08:00

480 lines
20 KiB
Python
Raw Blame History

import datetime
import requests, re
from bs4 import BeautifulSoup
import os
# with open('security_response.html', 'wb') as f:
# f.write(response.content)
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 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 soup.find('input', {'name' : "token"})['value']
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 soup.find('input', {'name' : "token"})['value']
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 = soup.find('input', {'id': "name"})['value']
token = soup.find('input', {'name': "token"})['value']
toUserId = soup.find('input', {'id': "toUserId"})['value']
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
)
# with open('transfer_credit.html', 'wb') as f:
# f.write(response.content)
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 soup.find('input', {'name' : "token"})['value']
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()
Reference in New Issue View Git Blame Copy Permalink