zhang_crawl_bio/worker/search_engine/camoufox_broswer.py

from camoufox import Camoufox
from camoufox.server import launch_server
from camoufox.async_api import AsyncCamoufox
from playwright.async_api import Browser,Page
import asyncio
import signal
import os
import datetime
from typing import Optional, Dict
import logging
from pydantic import BaseModel
from typing import Optional, Dict
from config.settings import OUTPUT_DIR, WORK_DIR

class BrowserConfig(BaseModel):
    """浏览器配置参数模型"""
    headless: bool = False
    geoip: bool = True
    proxy: Optional[Dict] = {'server': 'http://localhost:1881'}
    init_url: str = "https://www.browserscan.net"
    screenshot_dir: str = OUTPUT_DIR / "screenshots" 

class PageOperations:
    """封装页面交互操作"""
    def __init__(self, page: Page, config: BrowserConfig):
        self.page = page
        self.config = config

    async def search_element(self, selector: str, timeout: float = 30.0):
        """等待并返回指定元素"""
        return await self.page.wait_for_selector(selector, timeout=timeout)

    async def click_element(self, selector: str):
        """点击指定选择器的元素"""
        element = await self.search_element(selector)
        await element.click()

    async def take_screenshot(self, filename: str):
        """带时间戳的截图保存"""
        os.makedirs(self.config.screenshot_dir, exist_ok=True)
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        path = os.path.join(self.config.screenshot_dir, f"{timestamp}_{filename}")
        await self.page.screenshot(path=path, full_page=True)
        return path

    async def fill_input(self, selector: str, text: str):
        """在指定输入框填充文本"""
        element = await self.search_element(selector)
        await element.fill(text)

    async def press_enter(self):
        """执行键盘回车操作"""
        await self.page.keyboard.press("Enter")

class CamoufoxManager:
    """管理浏览器实例的生命周期"""
    def __init__(self, config: BrowserConfig = BrowserConfig()):
        self.config = config
        self.browser:Browser = None
        self.page:Page = None
        self.page_ops: PageOperations = None

    async def __aenter__(self):
        """异步上下文管理器入口"""
        self.browser = await AsyncCamoufox(
            headless=self.config.headless,
            geoip=self.config.geoip,
            proxy=self.config.proxy
        ).__aenter__()
        # 自动创建新页面并导航到初始URL
        self.page = await self.browser.new_page()
        await self.page.goto(self.config.init_url)
        self.page_ops = PageOperations(self.page, self.config)
        logging.info(f"Browser initialized with page: {self.page.url}")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        """异步上下文管理器退出"""
        await self.browser.__aexit__(exc_type, exc, tb)

async def aio_main(config: BrowserConfig = BrowserConfig(init_url="https://www.google.com")):
    """新版主运行函数（测试自动页面创建）"""
    async with CamoufoxManager(config) as manager:
        logging.info(f"当前页面URL: {manager.page.url}")
        
        # 保持页面存活直到手动关闭
        try:
            while not manager.page.is_closed():
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            logging.info("接收到终止信号，关闭浏览器...")
        finally:
            await manager.page.close()
            logging.info("测试完成")


async def task():
    pass
def main():
    asyncio.run(aio_main())

if __name__ == "__main__":
    main()