zhang_crawl_bio/worker/search_engine/camoufox_broswer.py

from pathlib import Path
import subprocess
import orjson
import contextlib
from camoufox import Camoufox
from camoufox.server import launch_server,launch_options,get_nodejs,to_camel_case_dict,LAUNCH_SCRIPT
from camoufox.async_api import AsyncCamoufox
from browserforge.fingerprints import Screen

from playwright.async_api import Browser, Page, BrowserContext, async_playwright
import asyncio
import signal
import os
import datetime
from typing import Optional, Dict, Type, Protocol, Union,ClassVar
import logging
from camoufox import DefaultAddons
from pydantic import BaseModel
from config.settings import OUTPUT_DIR, WORK_DIR
from mylib.logu import logger
# ------------------- Core Implementation -------------------
class BrowserConfig(BaseModel):
    """浏览器基础配置模型"""
    headless: bool = False
    geoip: bool = True
    proxy: Optional[Dict] = {'server': 'http://localhost:1881'}
    humanize: bool=True
    constrains: ClassVar[Screen] = Screen(max_width=1920, max_height=1200)
    # C:\Users\mg\AppData\Local\camoufox\camoufox\Cache\browser\features
    # addons: Optional[list]=[str(WORK_DIR / r"download\addon")]
    exclude_addons:Optional[list]=[DefaultAddons.UBO]


class BrowserCore():
    """浏览器核心功能实现（支持上下文管理器）"""
    def __init__(self, config: BrowserConfig):
        self.config = config
        self.camoufox: Optional[AsyncCamoufox] = None
        self.browser:  Union[Browser, BrowserContext] = None
        self.page: Page = None
        self.last_activity: datetime.datetime = None

    async def __aenter__(self):
        """异步上下文管理器入口"""
        await self.initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口"""
        await self.close()

    async def initialize(self):
        # 修改启动参数传递方式
        self.camoufox = AsyncCamoufox(
            headless=self.config.headless,
            geoip=self.config.geoip,
            proxy=self.config.proxy,
            screen=self.config.constrains,
        )
        self.browser = await self.camoufox.__aenter__()
        self.page = await self.browser.new_page()
        self.last_activity = datetime.datetime.now()
        logger.info(f"Browser session initialized | URL: {self.page.url}")

    async def close(self):
        """关闭浏览器实例"""
        if self.camoufox:
            await self.camoufox.__aexit__(None, None, None)
            logger.info("Browser session closed")

    async def goto(self, url: str):
        """导航到指定URL"""
        await self.page.goto(url)
        self.last_activity = datetime.datetime.now()

    async def take_screenshot(self, filename: str) -> str:
        """截图操作"""
        os.makedirs(self.screenshot_dir, exist_ok=True)
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        path = os.path.join(self.screenshot_dir, f"{timestamp}_{filename}")
        await self.page.screenshot(path=path, full_page=True)
        return path

    async def get_page_info(self) -> dict:
        pass
    
    def launch_server(self):
        file = OUTPUT_DIR / "ws.txt"
        logger.info(f"launch_server {file}")
        
        """
        Launch a Playwright server. Takes the same arguments as `Camoufox()`.
        Prints the websocket endpoint to the console.
        """
        config = launch_options(
            headless=self.config.headless,
            geoip=self.config.geoip,
            proxy=self.config.proxy,
            )
        nodejs = get_nodejs()

        data = orjson.dumps(to_camel_case_dict(config)).decode()
        
        # 打开一个文件用于写入标准输出和标准错误
        with open(file, "w") as output_file:
            process = subprocess.Popen(  # nosec
                [
                    nodejs,
                    str(LAUNCH_SCRIPT),
                ],
                cwd=Path(nodejs).parent / "package",
                stdin=subprocess.PIPE,
                stdout=output_file,
                stderr=subprocess.STDOUT,
                text=True,
            )
            if process.stdin:
                process.stdin.write(data)
                process.stdin.close()

            process.wait()
            raise RuntimeError("Server process terminated unexpectedly")

# ------------------- API Service -------------------
async def aio_main(config: BrowserConfig = BrowserConfig()):
    """API服务主循环"""
    async with BrowserCore(config) as core:
        try:
            logger.info(f"API服务已启动 | 初始页面: {core.page.url}")
            while True:
                await asyncio.sleep(5)
                
        except KeyboardInterrupt:
            logger.info("接收到终止信号，关闭浏览器...")
        except Exception as e:
            logger.error(f"API服务异常: {str(e)}")

def main():
    asyncio.run(aio_main())

if __name__ == "__main__":
    main()