import os from re import T import time import aiohttp import asyncio import requests from loguru import logger from functools import wraps def retry(max_retries: int = 3, delay: float = 1.0, backoff: float = 1.0): """ 通用重试装饰器 :param max_retries: 最大重试次数 :param delay: 每次重试的初始延迟（秒） :param backoff: 每次重试延迟的递增倍数 """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): retries = 0 current_delay = delay while retries < max_retries: try: return func(*args, **kwargs) except Exception as e: retries += 1 if retries >= max_retries: logger.warning(f"函数 {func.__name__} 在尝试了 {max_retries} 次后失败，错误信息: {e}") return None # 重试次数用尽后返回 None logger.warning(f"正在重试 {func.__name__} {retries + 1}/{max_retries} 因错误: {e}") time.sleep(current_delay) current_delay *= backoff return None # 三次重试仍未成功，返回 None return wrapper return decorator # 比特浏览器模块 class BitBrowser: def __init__(self): self.bit_host = "http://127.0.0.1" pass # 创建比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_create(self, remark: str = '指纹浏览器', ua: str = None, host: str = None, port: str = None, proxy_user: str = None, proxy_pwd: str = None, proxy_type: str = 'noproxy', urls: str = None, bit_port: str = "54345") -> str: """ 创建比特币浏览器 :param bit_port: 可选，默认54345 :param ua: 可选，默认随机 :param proxy_type: 代理类型 (可选) ['noproxy', 'http', 'https', 'socks5', 'ssh'] :param urls: 额外打开的url (可选) 多个用,分割 :param host: 代理IP地址 (可选) :param port: 代理IP端口 (可选) :param proxy_user: 代理账号 (可选) :param proxy_pwd: 代理密码 (可选) :param remark: 备注 (可选) :param bit_port: 可选，默认54345 :return: 返回浏览器ID """ url = f"{self.bit_host}:{bit_port}/browser/update" headers = {'Content-Type': 'application/json'} data = { 'name': f'{remark if len(remark) < 40 else remark[:40]}', # 窗口名称 'remark': f'{remark}', # 备注 'proxyMethod': 2, # 代理方式 2自定义 3 提取IP # 代理类型 ['noproxy', 'http', 'https', 'socks5', 'ssh'] 'proxyType': f'{proxy_type}', "browserFingerPrint": {"userAgent": ua} # 留空，随机指纹 } if host is not None: data['host'] = host if port is not None: data['port'] = port if proxy_user is not None: data['proxyUserName'] = proxy_user if proxy_pwd is not None: data['proxyPassword'] = proxy_pwd if urls is not None: data['url'] = urls # 额外打开的url 多个用,分割 res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) browser_pk = res['data']['id'] return browser_pk # 修改比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_update(self, pk: str, remark: str = None, proxyType: str = 'noproxy', host: str = None, port: str = None, proxy_user: str = None, proxy_pwd: str = None, urls: str = None, bit_port: str = "54345") -> bool: """ 修改比特币浏览器 传入某个参数则修改某个参数 :param proxyType: 代理类型 noproxy|http|https|socks5(默认noproxy) :param pk: # 浏览器ID :param remark: # 备注 :param host: # 代理主机 :param port: # 代理端口 :param proxy_user: # 代理账号 :param proxy_pwd: # 代理密码 :param urls: # 额外打开的url 多个用,分割 :param bit_port: # 可选，默认54345 :return: bool """ url = f"{self.bit_host}:{bit_port}/browser/update/partial" headers = {'Content-Type': 'application/json'} data = dict() data['ids'] = [pk] if remark is not None: data['remark'] = remark data['name'] = remark if urls is not None: data['url'] = urls if proxyType != 'noproxy': data['proxyType'] = proxyType if host is not None: data['host'] = host if port is not None: data['port'] = port if isinstance(port, int) else int(port) if proxy_user is not None: data['proxyUserName'] = proxy_user if proxy_pwd is not None: data['proxyPassword'] = proxy_pwd res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) return True # 打开比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_open(self, pk: str, bit_port: str = "54345") -> str: """ 打开比特币浏览器 :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: 返回浏览器地址 """ url = f"{self.bit_host}:{bit_port}/browser/open" data = {"id": f'{pk}'} headers = {'Content-Type': 'application/json'} res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) debugger_address = res['data']['http'] return debugger_address # 关闭比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_close(self, pk: str, bit_port: str = "54345"): """ 关闭比特币浏览器 - 执行后需要等待5s :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: 无返回值 """ url = f"{self.bit_host}:{bit_port}/browser/close" headers = {'Content-Type': 'application/json'} data = {'id': f'{pk}'} res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) # 等待3秒 time.sleep(3) bol = self.bit_browser_status(pk) if bol: raise Exception(f'浏览器ID {pk} 未正常关闭, 等待3秒后重试') return True # 删除比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_delete(self, pk: str, bit_port: str = "54345"): """ 删除比特币浏览器 :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: 无返回值 """ url = f"{self.bit_host}:{bit_port}/browser/delete" headers = {'Content-Type': 'application/json'} data = {'id': f'{pk}'} res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) return True # 获取所有比特币浏览器 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_get(self, page: int = 0, limit: int = 10, group_id: str | None = None, bit_port: str | None = "54345") -> dict: """ 获取所有比特币浏览器 :param page: 页码 :param limit: 每页数量 :param group_id: 组ID(可选) :param bit_port: 可选，默认54345 :return: {'success': True, 'data': {'page': 1, 'pageSize': 10, 'totalNum': 128, 'list': [{'id': '12a3126accc14c93bd34adcccfc3083c'},{'id':'edc5d61a56214e9f8a8bbf1a2e1b405d'}]}} """ url = f"{self.bit_host}:{bit_port}/browser/list" headers = {'Content-Type': 'application/json'} data = {'page': page, 'pageSize': limit} if group_id is not None: data['groupId'] = group_id res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) return res # 获取比特浏览器窗口详情 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_detail(self, pk: str, bit_port: str = "54345") -> dict: """ 获取比特浏览器窗口详情 :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: {'success': True, 'data': {'id': '12a3126accc14c93bd34adcccfc3083c', 'name': '12a3126accc14c93bd34adcccfc3083c', 'remark': '12a3126accc14c93bd34adcccfc3083c', ' """ url = f"{self.bit_host}:{bit_port}/browser/detail" headers = {'Content-Type': 'application/json'} data = {'id': f'{pk}'} res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) return res # 获取比特浏览器的进程id def bit_browser_pid(self, pk: str, bit_port: str = "54345") -> str: """ 获取比特浏览器的进程id :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: 返回进程id """ url = f"{self.bit_host}:{bit_port}/browser/pids/alive" headers = {'Content-Type': 'application/json'} data = { "ids": [pk] } res = requests.post(url, json=data, headers=headers).json() if not res.get('success'): raise Exception(res) return res['data'][pk] # 获取窗口状态 @retry(max_retries=3, delay=1.0, backoff=1.0) def bit_browser_status(self, pk: str, bit_port: str = "54345") -> dict: """ 获取比特浏览器窗口状态 :param pk: 浏览器ID :param bit_port: 可选，默认54345 :return: {'success': True, 'data': {'id': '12a3126accc14c93bd34adcccfc3083c', 'name': '12a3126accc14c93bd34adcccfc3083c', 'remark': '12a3126accc14c93bd34adcccfc3083c', ' """ url = f"{self.bit_host}:{bit_port}/browser/pids" headers = {'Content-Type': 'application/json'} data = {'ids': [pk]} res = requests.post(url, json=data, headers=headers).json() # print(f'res --> {res}') if not res.get('success'): raise Exception(res) if res.get('data').get(pk) is None: return False else: return True async def main(): bit = BitBrowser() # res = await bit._bit_browser_get() jc = 0 while 1: res = await bit._bit_browser_get( page=jc, limit=100, group_id='4028808b9a52223a019a581bbea1275c') li = res["data"]["list"] if len(li) == 0: break for i in li: id = i["id"] # 读取浏览器详情 res = await bit._bit_browser_detail(id) # print(f'id -->{id} --> {res}') data = res["data"] ua = data["browserFingerPrint"]["userAgent"] proxy_type = data.get("proxyType") host = data.get("host") port = data.get("port") proxy_account = data.get("proxyUserName") proxy_password = data.get("proxyPassword") print(f'id -->{id}') print(f'ua -->{ua}') print(f'proxy_type -->{proxy_type}') print(f'host -->{host}') print(f'port -->{port}') print(f'proxy_account -->{proxy_account}') print(f'proxy_password -->{proxy_password}') print(f'='*50) jc += 1 def main2(): bit = BitBrowser() browser_id = '5ba9eb974c7c45e2bb086585c75f70e8' # 关闭浏览器 res = bit.bit_browser_close(browser_id) print(res) # if __name__ == '__main__': # main2() bit_browser = BitBrowser()