import json import time from concurrent.futures import ThreadPoolExecutor from process.tools import get_ldplayer_list, close_ldplayer, start_ldplayer, get_element, get_host_ip, \ split_phone_number, rename_ldplayer, pull_simulator_backup from loguru import logger import uiautomator2 as u2 from models.tg_models import TelegramAccount import socket import os import sys import struct def send_file(file_path, server_ip='localhost', port=5001): """ 发送任意类型文件到服务端 :param file_path: 本地文件路径（支持任意后缀名） :param server_ip: 服务端IP地址，默认localhost :param port: 服务端端口，默认5555 :return: 传输成功返回True，失败返回False """ # 检查文件是否存在 if not os.path.exists(file_path): print(f"[Error] File not found: {file_path}") return False file_name = os.path.basename(file_path) client_sock = None try: # 创建Socket连接 client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_sock.settimeout(10) # 设置超时时间 client_sock.connect((server_ip, port)) # 发送文件名（支持任意后缀） file_name_encoded = file_name.encode() # 先发送文件名长度（4字节整数），再发送文件名 client_sock.sendall(struct.pack('I', len(file_name_encoded))) client_sock.sendall(file_name_encoded) # 分块发送文件内容 print(f"[Transfer] Sending: {file_name} -> {server_ip}") sent_bytes = 0 with open(file_path, 'rb') as f: while chunk := f.read(4096): # 4KB分块传输 client_sock.sendall(chunk) sent_bytes += len(chunk) print(f"[Success] Sent {sent_bytes} bytes") return True except socket.error as e: print(f"[Error] Network failure: {str(e)}") return False except Exception as e: print(f"[Error] Unexpected error: {str(e)}") return False finally: if client_sock: client_sock.close() def decorator(func): def wrapper(self, *args, **kwargs): result = None # 这里可以在函数执行前增加逻辑 # 启动模拟器 if not self.start_d(): return try: result = func(self, *args, **kwargs) except Exception as e: logger.error(f"电话号码：{self.tg_phone_info.index_id}：{e}！！！") # 这里可以在函数执行后增加逻辑 close_ldplayer(simulator_id=self.tg_phone_info.index_id) return result return wrapper class Ld: def __init__(self, tg_phone_info): self.tg_phone_info = tg_phone_info # for i in get_ldplayer_list(): # if self.tg_phone_info.ld_name == i["Name"]: # self.index_id = i["ID"] # break self.id = "emulator-" + str(5554 + (int(self.tg_phone_info.index_id) * 2)) def start_d(self): d_type = 0 # 连接模拟器 for i in range(3): if d_type: break start_ldplayer(simulator_id=self.tg_phone_info.index_id) for i in range(3): time.sleep(15) try: self.d = u2.connect(self.id) d_type = 1 break except Exception as e: pass else: logger.error(f"电话号码：{self.tg_phone_info.ld_name}：连接失败！！！") self.tg_phone_info.is_logged_in_telegram = -1 self.tg_phone_info.save() close_ldplayer(simulator_id=self.tg_phone_info.index_id, ) time.sleep(10) return d_type # 使用链接打开tg的检测 def tg_testing(self): logger.info(f"电话号码：{self.tg_phone_info.ld_name}：账号检测中》》》") # 判断需要tg打开链接 for i in range(8): type1, ele = get_element(d=self.d, type1="xpath", grammar='//*[@text="Continue"]', timeout=0.5) if type1: ele.click() type1, ele = get_element(d=self.d, type1="xpath", grammar='//*[@text="允许"]', timeout=3) if type1: ele.click() type1, ele = get_element(d=self.d, type1="xpath", grammar='//*[@text="允许"]', timeout=3) if type1: ele.click() type1, ele = get_element(d=self.d, type1="xpath", grammar='//*[@text="OK"]', timeout=0.5) if type1: ele.click() type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="Remind me later"]', timeout=0.5) if type1: ele.click() type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="Telegram"]', timeout=0.5) if type1: ele.click() time.sleep(5) type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="始终"]', timeout=0.5) if type1: ele.click() # 判断tg是否掉了 type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="Start Messaging"]', timeout=0.5) # 等待tg打开 if type1: self.tg_phone_info.is_logged_in_telegram = 0 self.tg_phone_info.save() logger.info(f"电话号码：{self.tg_phone_info.ld_name}：检测完成：未登录！！！") return True logger.warning(f"电话号码：{self.tg_phone_info.ld_name}：检测完成，已登录！！！") return False @decorator def action(self): type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="Telegram"]', timeout=5) # 等待tg打开 if type1: ele.click() else: self.tg_phone_info.is_logged_in_telegram = 0 self.tg_phone_info.save() return if self.tg_testing(): logger.error(f"雷电名称：{self.tg_phone_info.ld_name}，掉了！！！") return logger.info(f"雷电名称：{self.tg_phone_info.ld_name}，检测完成！！！") type1, ele = get_element(self.d, type1="xpath", grammar='//*[@content-desc="Open navigation menu"]', timeout=0.5) # 等待tg打开 if type1: ele.click() # for i in range(3): # type1, ele = get_element(self.d, type1="xpath", grammar='//*[@content-desc="Open navigation menu"]', # timeout=0.5) # 等待tg打开 # if type1: # ele.click() # # time.sleep(1) # # type1, ele = get_element(self.d, type1="xpath", grammar='//*[contains(@text, "+")]', # timeout=0.5) # 等待tg打开 # if type1: # break phones = self.d.xpath('//*[contains(@text, "+")]').get_text().replace("-", "").replace(' ', "").replace( '(', "").replace(')', "").replace("+", "") one_phone_number, telephone = split_phone_number(phone_number=phones) # 修改名字 rename_ldplayer(index=self.tg_phone_info.index_id, name=f"{one_phone_number}{telephone}") self.tg_phone_info.ld_name = f"{one_phone_number}{telephone}" self.tg_phone_info.one_phone_number = one_phone_number self.tg_phone_info.telephone = telephone self.tg_phone_info.is_logged_in_telegram = 1 self.tg_phone_info.phone_type = 1 self.tg_phone_info.save() # 模拟按下返回键 self.d.press("back") time.sleep(3) self.d.xpath('(//android.view.ViewGroup)[1]').click() time.sleep(3) type1, ele = get_element( self.d, type1="xpath", grammar='//*[@text="Your account is frozen\nTap to view details"]', timeout=5 ) # 等待tg打开 if type1: self.tg_phone_info.is_logged_in_telegram = 3 self.tg_phone_info.phone_type = 3 self.tg_phone_info.save() @decorator def bf(self): type1, ele = get_element(self.d, type1="xpath", grammar='//*[@text="Telegram"]', timeout=5) # 等待tg打开 if type1: ele.click() else: self.tg_phone_info.is_logged_in_telegram = 0 self.tg_phone_info.save() return if self.tg_testing(): logger.error(f"雷电名称：{self.tg_phone_info.ld_name}，掉了！！！") return logger.info(f"雷电名称：{self.tg_phone_info.ld_name}，检测完成！！！") # 通过adb shell命令获取 all_packages = self.d.shell('pm list packages').output # 格式化输出（去掉"package:"前缀） clean_packages = [pkg.split(':')[1] for pkg in all_packages.splitlines()] print("纯净包名列表:", clean_packages) package_name = None for i in clean_packages: if "telegram" in i: package_name = i break # 停止应用 self.d.app_stop(package_name) time.sleep(5) pull_simulator_backup( d=self.d, phone=self.tg_phone_info.ld_name, local_backup=path ) time.sleep(10) send_file(file_path=fr"{path}\{self.tg_phone_info.ld_name}.tar", server_ip="192.168.50.122") @decorator def get_sb(self, ): print(self.d.shell(f"adb -s {self.id} shell getprop ro.product.model")) # adb -s 127.0.0.1:5555 shell getprop ro.product.model def get_sb(tg_phone_info, directory, filename, ): # 打开并读取配置文件 with open(os.path.join(directory, filename), 'r', encoding='utf-8') as file: config_data = json.load(file) # 提取所需的键值 phone_manufacturer = config_data.get('propertySettings.phoneManufacturer', 'N/A') phone_model = config_data.get('propertySettings.phoneModel', 'N/A') # 组合结果 result = f"{phone_manufacturer} {phone_model}" tg_phone_info.manufacturer = result tg_phone_info.save() # 输出结果 print(f"文件名中的数字: {filename}, 设备信息: {result}") if __name__ == '__main__': directory = r"E:\leidian\LDPlayer9\vms\config" ld_ids = [] for i in get_ldplayer_list(): ld_ids.append(i["ID"]) # print(i["ID"]) # 同时运行 max_threads = 15 delay_between_start = 15 # 每次启动线程之间的延迟时间（秒） with ThreadPoolExecutor(max_workers=max_threads) as executor: for _, id in enumerate(ld_ids): tg_phone_info = TelegramAccount.get_or_none( TelegramAccount.index_id == id, TelegramAccount.server_id == get_host_ip(), ) if not tg_phone_info.ld_name: continue get_sb(tg_phone_info=tg_phone_info, directory=directory, filename=f"leidian{id}.config", ) # ld = Ld(tg_phone_info, ) # executor.submit(ld.action) # time.sleep(delay_between_start) # # for i in get_ldplayer_list(): # # if id == i["ID"] and i["Name"] == "NoneNone": # # ld = Ld(tg_phone_info, ) # # executor.submit(ld.action) # # time.sleep(delay_between_start) # # if tg_phone_info.is_logged_in_telegram == -1: # # logger.info(f"雷电名称：{id} 已经存在！！！") # # continue # # ld = Ld(tg_phone_info, ) # executor.submit(ld.action) # # time.sleep(delay_between_start) # # if tg_phone_info.is_logged_in_telegram == -1: # ld = Ld(tg_phone_info, ) # executor.submit(ld.action) # # time.sleep(delay_between_start) # from pathlib import Path # # path = r"E:\backups" # # # 指定目录 # directory = Path(path) # # # 获取目录下的所有文件 # start = 0 # file_list = [str(file) for file in directory.rglob('*') if file.is_file()] # for file in file_list: # if tg_phone_info.ld_name in file: # start = 1 # break # # if start: # send_file(file_path=fr"{path}\{tg_phone_info.ld_name}.tar", server_ip="192.168.50.122") # continue # # ld = Ld(tg_phone_info, ) # executor.submit(ld.bf)