ca_auto_table/spider/bit_browser.py

import time
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)
    # res = bit.bit_browser_get()
    # print(res)

# if __name__ == '__main__':
    # main2()

bit_browser = BitBrowser()