import requests, re from bs4 import BeautifulSoup from cm_bot import CM_BOT from db import DB import secrets, string class CM_BOT_HAL: def __init__(self): self.db = DB() self.prefix = '13c' self.agent_id = 'cm13a3' self.agent_password = 'Sky533535' self.security_pin = 'Sky533535' def get_random_password(self): length = 8 lower_case = string.ascii_lowercase upper_case = string.ascii_uppercase digits = string.digits all_characters = lower_case + upper_case + digits password_list = [ secrets.choice(lower_case), secrets.choice(upper_case), secrets.choice(digits), secrets.choice(digits), secrets.choice(digits) ] while len(password_list) < length: password_list.append(secrets.choice(all_characters)) secrets.SystemRandom().shuffle(password_list) return "".join(password_list) def get_user_data_from_acc(self): query = "SELECT username, password, link FROM acc WHERE status = %s ORDER BY username ASC LIMIT 1" query_params = [''] result = self.db.query(query, query_params) return result[0] if len(result) else None def get_max_username(self, prefix: str): query = "SELECT username FROM acc WHERE username LIKE %s ORDER BY username DESC LIMIT 1" query_params = [f'{prefix}%'] result = self.db.query(query, query_params) return result[0]['username'] def get_next_username(self, prefix: str): last_index = int(self.get_max_username(prefix).replace(prefix, '')) next_index = last_index + 2 if last_index % 10 == 3 else last_index + 1 return f'{prefix}{next_index}' def insert_user_to_table_acc(self, user): query = "INSERT INTO acc (username, password, link) VALUES (%s, %s, %s)" query_params = [user['username'], user['password'], user['link']] return self.db.execute(query, query_params) def insert_user_to_table_user(self, user): query = "INSERT INTO user (f_username, f_password, t_username, t_password) VALUES (%s, %s, %s, %s)" query_params = [user['f_username'], user['f_password'], user['t_username'], user['t_password']] return self.db.execute(query, query_params) def create_new_acc(self): cm_bot = CM_BOT() username = self.get_next_username(self.prefix) password = self.get_random_password() cm_bot.login( username = self.agent_id, password = self.agent_password ) cm_bot.register_user( user_id = username, user_password = password ) cm_bot.logout() cm_bot = CM_BOT() cm_bot.login( username = username, password = password ) link = cm_bot.get_register_link() cm_bot.logout() user = {'username': username, 'password': password, 'link': link} self.insert_user_to_table_acc(user) return user def update_user_status_to_wait(self, username: str): query = "UPDATE acc SET status = 'wait' WHERE username = %s" query_params = [username] return self.db.execute(query, query_params) def update_user_status_to_done(self, username: str): query = "UPDATE acc SET status = 'done' WHERE username = %s" query_params = [username] return self.db.execute(query, query_params) def get_user_api(self): user = self.get_user_data_from_acc() if user is None: user = self.create_new_acc() self.update_user_status_to_wait(user['username']) return user def is_whatsapp_url(self, text): url_pattern = re.compile( r'^(https?://)?' # Optional http:// or https:// r'((www\.)|([a-zA-Z0-9-]+\.))+' # www. or subdomain. # r'[a-zA-Z]{2,6}' # Top-level domain (e.g., com, org, net) r'(whatsapp.com)' r'(/[a-zA-Z0-9-._~:/?#\[\]@!$&\'()*+,;=]*)?$' # Optional path, query, fragment ) return bool(url_pattern.search(text)) def get_whatsapp_link_username(self, whatsapp_link: str): if self.is_whatsapp_url(whatsapp_link) == False: return None response = requests.get(whatsapp_link) soup = BeautifulSoup(response.content, 'html.parser') username = [x.replace('*', '').replace(' ', '') for x in soup.find('h3', {"class": "_9vd5 _9scr"}).text.split("/")] return username def get_user_pass_from_acc(self, username: str): query = "SELECT password FROM acc WHERE username = %s" query_params = [username] result = self.db.query(query, query_params) return result[0]['password'] def set_security_pin_api(self, whatsapp_link: str): t_username, f_username = self.get_whatsapp_link_username(whatsapp_link) password = self.get_user_pass_from_acc(f_username) cm_bot = CM_BOT() cm_bot.login( username = f_username, password = password ) if cm_bot.set_security_pin(self.security_pin) == False: cm_bot.logout() raise Exception(f'Agent acc: {f_username} already has security pin!') cm_bot.logout() result = self.update_user_status_to_done(f_username) if result == False: raise Exception('Failed to update user status to done') result = self.insert_user_to_table_user( { 'f_username': f_username, 'f_password': password, 't_username': t_username, 't_password': self.security_pin } ) if result == False: raise Exception('Failed to insert user to table user') def get_user_credit(self, f_username: str, f_password: str): cm_bot = CM_BOT() cm_bot.login( username = f_username, password = f_password ) return float(cm_bot.get_user_credit()) def transfer_credit_api(self, f_username: str, f_password: str, t_username: str, t_password: str): cm_bot = CM_BOT() cm_bot.login( username = f_username, password = f_password ) amount = cm_bot.get_user_credit() - 0.01 if cm_bot.transfer_credit(t_username, t_password, amount) == True: print(f'Successfully transfer amount: {amount} from {f_username} to {t_username}') else: print(f'Failed to transfer credit from {f_username} to {t_username}') if __name__ == '__main__': bot = CM_BOT_HAL() # bot.transfer_credit_api('13c4070', 'zU2QoL', '4753kit', 'Sky533535') # print(bot.get_next_username('13c')) # print(bot.get_random_password()) # bot.get_user_api() # bot.insert_user_to_table_acc({'username': 'test0001', 'password': 'test0001', 'link': 'test0001'}) # print(bot.get_whatsapp_link_username('https://chat.whatsapp.com/DZDWcicr6MTFrR4kBfnJXO')) # print(bot.get_user_pass_from_acc('13c4151'))