用户输入任务
    → LLM 处理与推理
    → DOM 提取 + 视觉分析
    → 动作规划
    → Playwright 执行
    → 状态更新
    → 反馈回路（循环直到任务完成）

from browser_use import Agent, ChatBrowserUse
from dotenv import load_dotenv
import asyncio

load_dotenv()

async def main():
    llm = ChatBrowserUse()
    agent = Agent(
        task="Find the number 1 post on Show HN",
        llm=llm
    )
    await agent.run()

if __name__ == "__main__":
    asyncio.run(main())

from browser_use import Browser

browser = Browser(
    headless=False,                    # 是否无头模式
    window_size={'width': 1920, 'height': 1080},
    highlight_elements=True,           # 高亮交互元素（辅助 AI 视觉识别）
    paint_order_filtering=True,        # 过滤被遮挡的元素，优化 DOM 树
    minimum_wait_page_load_time=0.25,  # 页面加载最小等待时间
    wait_for_network_idle_page_load_time=0.5,
    wait_between_actions=0.5,          # 动作间等待时间
)

agent = Agent(
    task="...",
    llm=llm,
    use_vision="auto",        # "auto" | True | False
    vision_detail_level="auto" # "low" | "high" | "auto"
)

while not done and steps < max_steps:
    # 1. 获取当前页面状态
    state = extract_dom() + take_screenshot()
  
    # 2. 组装上下文，调用 LLM
    response = llm.call(system_prompt + state + task + history)
  
    # 3. 解析 LLM 返回的动作
    actions = parse_actions(response)
  
    # 4. 执行动作（最多 max_actions_per_step 个）
    for action in actions:
        result = playwright.execute(action)
        if page_changed:
            break  # 页面变化后停止执行剩余动作
  
    # 5. 更新历史记录
    history.append(state, actions, result)
  
    # 6. 检查是否完成
    if response.is_done:
        done = True

from browser_use import Tools, Agent, ActionResult, BrowserSession

tools = Tools()

@tools.action(description='Save extracted data to database')
async def save_to_db(data: dict, browser_session: BrowserSession) -> ActionResult:
    # 你的数据库写入逻辑
    db.insert(data)
    return ActionResult(extracted_content=f"Saved {len(data)} records")

@tools.action(description='Get 2FA code from authenticator')
async def get_2fa_code() -> ActionResult:
    code = authenticator.get_current_code()
    return ActionResult(extracted_content=f"2FA code: {code}")

agent = Agent(
    task="Login to the dashboard and export monthly report",
    llm=llm,
    tools=tools
)

history = await agent.run()

# 基本信息
history.urls()                    # 访问过的 URL 列表
history.action_names()            # 执行过的动作名称
history.extracted_content()       # 提取的内容列表
history.errors()                  # 错误列表

# 分析方法
history.final_result()            # 最终结果
history.is_done()                 # 是否完成
history.is_successful()           # 是否成功
history.model_thoughts()          # Agent 的推理过程
history.number_of_steps()         # 总步数
history.total_duration_seconds()  # 总耗时

# 结构化输出
history.structured_output         # 当使用 output_model_schema 时

+--------------------------------------------------+
|  Layer 3: 编排层 (Orchestration)                   |
|  工作流逻辑、Agent 协调、路由                        |
+--------------------------------------------------+
|  Layer 2: 运行时环境 (Runtime Environment)          |
|  工具、记忆、护栏、I/O 处理                          |
+--------------------------------------------------+
|  Layer 1: 模型接口 (Model Interface)               |
|  API 调用、Prompt 组装、响应解析                     |
+--------------------------------------------------+

请求
  |
  v
[尝试 1: 主模型]
  |-- 成功 --> 验证 --> 返回
  |-- 失败 --> [尝试 2: 重新组织 Prompt]
                  |-- 成功 --> 验证 --> 返回
                  |-- 失败 --> [尝试 3: 降级模型]
                                  |-- 成功 --> 验证 --> 返回
                                  |-- 失败 --> [升级到人工处理]

AGENTS.md (约 100 行，作为地图)
  ├── docs/design/        # 设计文档（含验证状态）
  ├── docs/architecture/  # 架构文档（域和包分层的顶层地图）
  ├── docs/quality/       # 质量文档（每个领域和层的评分）
  ├── docs/plans/         # 执行计划（进度和决策日志）
  └── docs/principles/    # 核心理念（Agent 优先的操作原则）

Types → Config → Repo → Service → Runtime → UI

import asyncio
import logging
from browser_use import Agent, Browser, Tools, ActionResult, ChatBrowserUse
from pydantic import BaseModel

# ============================================
# Layer 1: 模型接口 - 配置与降级策略
# ============================================

# 主模型
primary_llm = ChatBrowserUse()

# 降级模型（当主模型失败时）
from browser_use import ChatOpenAI
fallback_llm = ChatOpenAI(model="gpt-4.1-mini")

# 页面提取专用小模型（降低成本）
extraction_llm = ChatOpenAI(model="gpt-4.1-mini")

# ============================================
# Layer 2: 运行时环境 - 工具、护栏、验证
# ============================================

# 结构化输出 Schema（输出护栏）
class ResearchResult(BaseModel):
    title: str
    url: str
    summary: str
    confidence: float

# 自定义工具
tools = Tools(exclude_actions=['evaluate'])  # 移除 JS 执行（安全护栏）

@tools.action(
    description='Save research results to database',
    allowed_domains=['*.example.com']  # 域名限制护栏
)
async def save_results(results: list[dict]) -> ActionResult:
    # 输出验证
    validated = [ResearchResult(**r) for r in results]
    # 持久化逻辑
    return ActionResult(
        extracted_content=f"Saved {len(validated)} results",
        long_term_memory="Results have been saved, no need to save again"
    )

@tools.action(description='Ask human for confirmation before proceeding')
async def human_confirmation(question: str) -> ActionResult:
    answer = input(f"[Agent asks]: {question} (y/n) > ")
    return ActionResult(extracted_content=f"Human responded: {answer}")

# 浏览器配置（执行护栏）
browser = Browser(
    headless=True,
    window_size={'width': 1920, 'height': 1080},
    allowed_domains=['*.google.com', '*.github.com'],  # 域名白名单
    prohibited_domains=['*.gambling-site.net'],          # 域名黑名单
    highlight_elements=True,
    paint_order_filtering=True,
    record_video_dir='./recordings',  # 可观测性：录制操作视频
    minimum_wait_page_load_time=0.5,
    wait_for_network_idle_page_load_time=1.0,
)

# ============================================
# Layer 3: 编排层 - 重试、降级、反馈回路
# ============================================

async def run_with_harness(task: str, max_retries: int = 3):
    """带完整 Harness 的 Agent 执行函数"""
  
    for attempt in range(max_retries):
        try:
            # 选择模型（首次用主模型，重试用降级模型）
            llm = primary_llm if attempt == 0 else fallback_llm
          
            agent = Agent(
                task=task,
                llm=llm,
                browser=browser,
                tools=tools,
                page_extraction_llm=extraction_llm,
                use_vision="auto",
                max_actions_per_step=3,
                max_failures=3,
                use_thinking=True,
                output_model_schema=ResearchResult,
                save_conversation_path=f'./logs/conversation_{attempt}.json',
                generate_gif=True,
            )


            history = await agent.run(max_steps=50)
          
            # ========== 执行后验证（输出护栏） ==========
            if history.is_done() and history.is_successful():
                result = history.structured_output
              
                # 业务逻辑验证
                if result and result.confidence >= 0.7:
                    logging.info(f"Task completed successfully on attempt {attempt + 1}")
                    logging.info(f"Steps: {history.number_of_steps()}, "
                                 f"Duration: {history.total_duration_seconds():.1f}s")
                    return result
                else:
                    logging.warning(f"Low confidence result ({result.confidence}), retrying...")
                    continue
          
            # 任务未完成但未报错——可能是 max_steps 耗尽
            if not history.is_done():
                logging.warning(f"Agent did not finish within max_steps on attempt {attempt + 1}")
                # 分析失败原因，调整下一轮策略
                thoughts = history.model_thoughts()
                logging.debug(f"Last thoughts: {thoughts[-3:] if thoughts else 'None'}")
                continue
              
        except Exception as e:
            logging.error(f"Attempt {attempt + 1} failed with error: {e}")
            if attempt == max_retries - 1:
                # 最终降级：人工升级
                logging.critical("All retries exhausted. Escalating to human operator.")
                raise
            await asyncio.sleep(2 ** attempt)  # 指数退避
  
    return None


# ============================================
# 入口
# ============================================
async def main():
    result = await run_with_harness(
        task="Go to GitHub trending repositories, find the top 3 Python projects "
             "from this week, extract their name, URL, and description."
    )
    if result:
        print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

系统 Prompt（角色定义 + 工具描述）
  + 任务描述
  + 当前页面 DOM（可能非常大）
  + 截图（如果启用视觉）
  + 历史记录（之前的动作和结果）
  + 自定义指令

agent = Agent(
    task="...",
    llm=llm,
  
    # 上下文压缩：用小模型提取页面关键内容，而非传递完整 DOM
    page_extraction_llm=ChatOpenAI(model="gpt-4.1-mini"),
  
    # 历史窗口：仅保留最近 N 步的完整历史
    # 更早的历史会被压缩为摘要
    max_history_items=20,
  
    # DOM 过滤：paint_order_filtering 移除被遮挡的元素
    # 减少 DOM 噪声
    browser=Browser(paint_order_filtering=True),
  
    # 视觉细节级别：low 模式使用更少的 Token
    vision_detail_level="low",
)

@tools.action(description='Login to the system')
async def login(username: str, password: str) -> ActionResult:
    # ... 登录逻辑 ...
    return ActionResult(
        extracted_content="Login successful",
        # 这条信息会被持久化到 Agent 的记忆中
        # 在后续所有步骤中都可见
        long_term_memory="Already logged in as admin. No need to login again."
    )

agent = Agent(
    task="Continue the research from yesterday",
    llm=llm,
    save_conversation_path='./sessions/research_session.json',
)

┌─────────────────────────────────────────────────┐
│  第一层：输入护栏                                  │
│  - 任务描述的内容过滤（防止 Prompt 注入）           │
│  - 敏感信息脱敏（PII 检测与替换）                   │
│  - 任务复杂度评估（过于复杂的任务拒绝或拆分）        │
├─────────────────────────────────────────────────┤
│  第二层：执行护栏                                  │
│  - 域名白名单/黑名单                               │
│  - 工具权限控制（禁用危险工具如 evaluate）           │
│  - 最大步数限制（max_steps）                       │
│  - 最大失败次数（max_failures）                    │
│  - 单步最大动作数（max_actions_per_step）           │
│  - 页面加载超时                                    │
│  - 成本预算（Token 消耗上限）                       │
├─────────────────────────────────────────────────┤
│  第三层：输出护栏                                  │
│  - 结构化输出验证（output_model_schema）            │
│  - 业务逻辑检查（如置信度阈值）                     │
│  - 敏感数据过滤（防止 Agent 泄露密码等）             │
│  - 人工审批门（关键操作前请求确认）                  │
└─────────────────────────────────────────────────┘

from browser_use import Agent, Browser, Tools, ActionResult

# 执行护栏：浏览器级别
browser = Browser(
    allowed_domains=['*.target-site.com'],
    prohibited_domains=['*.social-media.com', '*.payment.com'],
    headless=True,
)

# 工具护栏：禁用危险操作
tools = Tools(
    exclude_actions=[
        'evaluate',      # 禁止执行任意 JavaScript
        'upload_file',   # 禁止上传文件
        'write_file',    # 禁止写入本地文件
    ]
)

# 输出护栏：结构化验证
from pydantic import BaseModel, validator

class SafeOutput(BaseModel):
    data: str
    source_url: str
  
    @validator('data')
    def no_sensitive_info(cls, v):
        import re
        # 检测并拒绝包含信用卡号的输出
        if re.search(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', v):
            raise ValueError("Output contains potential credit card number")
        return v

# 人工审批护栏
@tools.action(description='Submit form with important data')
async def submit_form(data: dict) -> ActionResult:
    print(f"\n⚠️  Agent wants to submit: {data}")
    approval = input("Approve? (yes/no) > ")
    if approval.lower() != 'yes':
        return ActionResult(extracted_content="Human rejected the submission. Try a different approach.")
    # ... 执行提交 ...
    return ActionResult(extracted_content="Form submitted successfully")

from browser_use import Agent, Browser

browser = Browser()

# Agent 1：负责搜索和收集信息
researcher = Agent(
    task="Search for the latest AI research papers on arxiv about browser agents",
    llm=llm,
    browser=browser,
)

# Agent 2：负责分析和总结
analyst = Agent(
    task="Analyze the papers found in the browser tabs and create a summary",
    llm=llm,
    browser=browser,  # 共享同一个浏览器
)

# 顺序执行
research_history = await researcher.run()
analysis_history = await analyst.run()

async def orchestrated_research(topic: str):
    browser = Browser()
  
    # 阶段 1：并行搜索（多个 Agent 搜索不同来源）
    agents = [
        Agent(task=f"Search {source} for: {topic}", llm=llm, browser=browser)
        for source in ["Google Scholar", "arXiv", "GitHub"]
    ]
  
    # 注意：Browser Use 的浏览器实例是单线程的
    # 并行需要多个浏览器实例
    results = []
    for agent in agents:
        history = await agent.run()
        results.append(history.extracted_content())
  
    # 阶段 2：综合分析
    synthesizer = Agent(
        task=f"Based on these findings, create a comprehensive summary:\n{results}",
        llm=llm,
        browser=browser,
    )
  
    return await synthesizer.run()

import os

# 方式一：通过 sensitive_data 参数注入（不会出现在日志中）
agent = Agent(
    task="Login to the admin dashboard at https://admin.example.com",
    llm=llm,
    sensitive_data={
        'ADMIN_USERNAME': os.getenv('ADMIN_USERNAME'),
        'ADMIN_PASSWORD': os.getenv('ADMIN_PASSWORD'),
        'TOTP_SECRET': os.getenv('TOTP_SECRET'),
    },
)

# 方式二：通过自定义工具注入 2FA 代码
import pyotp

@tools.action(description='Get current 2FA code')
async def get_2fa() -> ActionResult:
    totp = pyotp.TOTP(os.getenv('TOTP_SECRET'))
    return ActionResult(extracted_content=f"Current 2FA code: {totp.now()}")

from browser_use import Browser

# 方式一：连接到已运行的 Chrome 实例（通过 CDP）
# 先启动 Chrome：chrome --remote-debugging-port=9222
browser = Browser(
    cdp_url="ws://localhost:9222/devtools/browser/..."
)

# 方式二：使用持久化的用户数据目录（保留 Cookie 和登录状态）
browser = Browser(
    user_data_dir='./chrome_profile',
    keep_alive=True,  # 任务结束后不关闭浏览器
)

# 方式三：使用 Browser Use Cloud（自动处理验证码和反爬）
browser = Browser(use_cloud=True)

agent = Agent(
    task="...",
    llm=llm,
  
    # 1. 使用小模型提取页面内容（大幅降低主模型的输入 Token）
    page_extraction_llm=ChatOpenAI(model="gpt-4.1-mini"),
  
    # 2. 降低视觉细节级别
    vision_detail_level="low",
  
    # 3. 限制历史记忆窗口
    max_history_items=10,
  
    # 4. 启用 flash 模式（跳过评估和推理）
    flash_mode=True,  # 适合简单、重复性任务
  
    # 5. 增加单步动作数（减少 LLM 调用次数）
    max_actions_per_step=5,
)

browser = Browser(
    headless=True,                          # 无头模式更快
    minimum_wait_page_load_time=0.1,        # 减少等待时间
    wait_for_network_idle_page_load_time=0.3,
    wait_between_actions=0.2,               # 减少动作间隔
)

import logging

# 启用详细日志
logging.basicConfig(level=logging.DEBUG)

agent = Agent(
    task="...",
    llm=llm,
  
    # 保存完整对话历史（含 LLM 输入输出）
    save_conversation_path='./debug/conversation.json',
  
    # 生成操作 GIF 动画
    generate_gif=True,
  
    # 录制操作视频
    browser=Browser(record_video_dir='./debug/videos'),
)

history = await agent.run()

# 执行后分析
print(f"总步数: {history.number_of_steps()}")
print(f"总耗时: {history.total_duration_seconds():.1f}s")
print(f"访问的 URL: {history.urls()}")
print(f"执行的动作: {history.action_names()}")
print(f"错误: {history.errors()}")
print(f"Agent 思考过程: {history.model_thoughts()}")

# 自定义指标收集
class AgentMetrics:
    def __init__(self):
        self.runs = []
  
    async def track_run(self, task: str, agent: Agent):
        import time
        start = time.time()
        history = await agent.run()
        duration = time.time() - start
      
        metrics = {
            'task': task,
            'success': history.is_successful(),
            'steps': history.number_of_steps(),
            'duration_seconds': duration,
            'errors': len(history.errors()),
            'urls_visited': len(history.urls()),
            'actions_executed': len(history.action_names()),
            # Token 使用量需要从 LLM 回调中获取
        }
      
        self.runs.append(metrics)
      
        # 发送到监控系统
        # datadog.send_metrics(metrics)
        # prometheus.observe(metrics)
      
        return history

@tools.action(description='Solve CAPTCHA that the agent cannot handle')
async def solve_captcha() -> ActionResult:
    input("Please solve the CAPTCHA in the browser window, then press Enter...")
    return ActionResult(extracted_content="CAPTCHA solved by human")

browser = Browser(
    # 等待网络空闲，确保异步请求完成
    wait_for_network_idle_page_load_time=2.0,
    # 额外的页面加载等待
    minimum_wait_page_load_time=1.0,
)

# 或者在 Agent 层面使用 wait 动作
# Agent 可以主动调用 wait 工具等待特定条件

@tools.action(description='Verify current page matches expected state')
async def verify_state(expected_url_pattern: str, expected_content: str, 
                       browser_session) -> ActionResult:
    page = browser_session.current_page
    current_url = page.url
  
    if expected_url_pattern not in current_url:
        return ActionResult(
            extracted_content=f"WARNING: Expected URL containing '{expected_url_pattern}' "
                            f"but got '{current_url}'. Navigation may have gone wrong.",
            long_term_memory=f"State verification failed. Current URL: {current_url}"
        )
  
    content = await page.content()
    if expected_content.lower() not in content.lower():
        return ActionResult(
            extracted_content=f"WARNING: Expected content '{expected_content}' not found on page."
        )
  
    return ActionResult(extracted_content="State verification passed.")

预算控制层
├── 任务级预算：每个任务设置最大 Token/成本上限
├── 步骤级优化：flash_mode、减少视觉使用、压缩历史
├── 模型级策略：简单步骤用小模型，关键决策用大模型
└── 架构级设计：缓存常见页面的 DOM 结构，避免重复提取

# 未来可能的自进化模式

# 1. Agent 从失败中学习，自动生成新的工具
@tools.action(description='Create a new reusable tool from learned pattern')
async def create_tool(tool_name: str, tool_code: str, description: str) -> ActionResult:
    # Agent 识别到重复的操作模式后
    # 自动将其封装为可复用的工具
    # 保存到工具库中供后续任务使用
    pass

# 2. Agent 自动优化 Prompt
# 基于历史成功/失败记录，调整系统 Prompt 中的指导信息

# 3. Agent 自动生成测试用例
# 对关键业务流程生成回归测试，确保网站变化时及时发现

import asyncio
import json
import logging
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from browser_use import Agent, Browser, Tools, ActionResult, ChatBrowserUse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ============================================
# 数据模型（输出护栏）
# ============================================

class ProductPrice(BaseModel):
    product_name: str
    price: float = Field(ge=0)
    currency: str = Field(default="USD")
    in_stock: bool
    promotion: Optional[str] = None
    source_url: str
    extracted_at: str = Field(default_factory=lambda: datetime.now().isoformat())

class CompetitorReport(BaseModel):
    competitor: str
    products: list[ProductPrice]
    extraction_success: bool
    error_message: Optional[str] = None

# ============================================
# 工具定义
# ============================================

tools = Tools(
    exclude_actions=['evaluate', 'upload_file']  # 安全护栏
)

# 模拟数据库
price_history_db: list[dict] = []

@tools.action(description='Save product price to database and check for significant changes')
async def save_price(product_name: str, price: float, currency: str, 
                     in_stock: bool, source_url: str, 
                     promotion: str = None) -> ActionResult:
    record = ProductPrice(
        product_name=product_name,
        price=price,
        currency=currency,
        in_stock=in_stock,
        promotion=promotion,
        source_url=source_url,
    )
  
    # 检查价格变动
    previous = [r for r in price_history_db 
                if r['product_name'] == product_name and r['source_url'] == source_url]
  
    alert_msg = ""
    if previous:
        last_price = previous[-1]['price']
        change_pct = abs(price - last_price) / last_price * 100
        if change_pct >= 5:
            alert_msg = (f" ALERT: Price changed by {change_pct:.1f}% "
                        f"(was {last_price} {currency}, now {price} {currency})")
            logger.warning(f"Price alert for {product_name}: {alert_msg}")
            # 在生产环境中，这里会发送 Slack/Email 通知
  
    price_history_db.append(record.model_dump())
  
    return ActionResult(
        extracted_content=f"Saved: {product_name} @ {price} {currency}.{alert_msg}",
        long_term_memory=f"Price for {product_name} has been saved. Do not save it again."
    )

@tools.action(description='Report extraction failure for a competitor')
async def report_failure(competitor: str, error: str) -> ActionResult:
    logger.error(f"Extraction failed for {competitor}: {error}")
    return ActionResult(
        extracted_content=f"Failure reported for {competitor}: {error}"
    )

# ============================================
# Harness 编排层
# ============================================

class CompetitorMonitor:
    def __init__(self):
        self.llm = ChatBrowserUse()
        self.browser = Browser(
            headless=True,
            window_size={'width': 1920, 'height': 1080},
            highlight_elements=True,
            paint_order_filtering=True,
            minimum_wait_page_load_time=1.0,
            wait_for_network_idle_page_load_time=2.0,
        )

    async def monitor_competitor(self, competitor_name: str, url: str,
                                  products: list[str]) -> CompetitorReport:
        """监控单个竞品网站，带完整的重试和降级策略"""
      
        max_retries = 3
      
        for attempt in range(max_retries):
            try:
                product_list = ", ".join(products)
                task = (
                    f"Go to {url} and find the following products: {product_list}. "
                    f"For each product, extract the current price, currency, "
                    f"whether it's in stock, and any active promotions. "
                    f"Use the save_price tool to save each product's data. "
                    f"If a product cannot be found, skip it and continue with the next one."
                )
              
                agent = Agent(
                    task=task,
                    llm=self.llm,
                    browser=self.browser,
                    tools=tools,
                    use_vision="auto",
                    use_thinking=True,
                    max_actions_per_step=3,
                    max_failures=3,
                    save_conversation_path=(
                        f'./logs/{competitor_name}_{datetime.now().strftime("%Y%m%d_%H%M")}'
                        f'_attempt{attempt}.json'
                    ),
                    generate_gif=True,
                )
              
                history = await agent.run(max_steps=40)
              
                if history.is_done():
                    # 从数据库中提取本次运行保存的记录
                    saved_products = [
                        ProductPrice(**r) for r in price_history_db
                        if r['source_url'].startswith(url.split('/')[2])  # 匹配域名
                    ]
                  
                    logger.info(
                        f"[{competitor_name}] Completed: {len(saved_products)} products, "
                        f"{history.number_of_steps()} steps, "
                        f"{history.total_duration_seconds():.1f}s"
                    )
                  
                    return CompetitorReport(
                        competitor=competitor_name,
                        products=saved_products,
                        extraction_success=True,
                    )
              
                # Agent 未完成——可能是步数耗尽
                logger.warning(
                    f"[{competitor_name}] Agent did not finish on attempt {attempt + 1}. "
                    f"Last thoughts: {history.model_thoughts()[-1] if history.model_thoughts() else 'N/A'}"
                )
              
            except Exception as e:
                logger.error(f"[{competitor_name}] Attempt {attempt + 1} failed: {e}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # 指数退避：1s, 2s, 4s
      
        # 所有重试耗尽
        return CompetitorReport(
            competitor=competitor_name,
            products=[],
            extraction_success=False,
            error_message=f"All {max_retries} attempts failed",
        )
  
    async def run_daily_monitoring(self, competitors: list[dict]) -> list[CompetitorReport]:
        """每日监控入口：顺序执行所有竞品监控"""
      
        reports = []
      
        for comp in competitors:
            logger.info(f"Starting monitoring for: {comp['name']}")
            report = await self.monitor_competitor(
                competitor_name=comp['name'],
                url=comp['url'],
                products=comp['products'],
            )
            reports.append(report)
          
            # 竞品之间间隔，避免触发反爬
            await asyncio.sleep(5)
      
        # 生成汇总报告
        success_count = sum(1 for r in reports if r.extraction_success)
        total_products = sum(len(r.products) for r in reports)
      
        logger.info(
            f"\n{'='*50}\n"
            f"Daily Monitoring Summary\n"
            f"Competitors monitored: {len(reports)}\n"
            f"Successful: {success_count}/{len(reports)}\n"
            f"Total products tracked: {total_products}\n"
            f"{'='*50}"
        )
      
        return reports


# ============================================
# 入口
# ============================================

async def main():
    competitors = [
        {
            "name": "Competitor A",
            "url": "https://www.competitor-a.com/products",
            "products": ["Widget Pro", "Widget Lite", "Widget Enterprise"],
        },
        {
            "name": "Competitor B",
            "url": "https://www.competitor-b.com/pricing",
            "products": ["Basic Plan", "Pro Plan", "Enterprise Plan"],
        },
        {
            "name": "Competitor C",
            "url": "https://www.competitor-c.com/store",
            "products": ["Starter Kit", "Professional Kit"],
        },
    ]
  
    monitor = CompetitorMonitor()
    reports = await monitor.run_daily_monitoring(competitors)
  
    # 持久化报告
    with open(f'./reports/daily_{datetime.now().strftime("%Y%m%d")}.json', 'w') as f:
        json.dump([r.model_dump() for r in reports], f, indent=2, default=str)


if __name__ == "__main__":
    asyncio.run(main())

# 未来可能的声明式任务定义
task:
  name: "daily-price-check"
  schedule: "0 9 * * *"
  targets:
    - url: "https://competitor.com/pricing"
      extract:
        - field: "price"
          selector_hint: "main pricing table"
          type: float
        - field: "plan_name"
          type: string
  output:
    format: "json"
    destination: "s3://my-bucket/prices/"
  guardrails:
    max_cost_per_run: "$0.50"
    max_duration: "5m"
    allowed_domains: ["competitor.com"]