Agent/examples/auto_put_ad_mini/execute_once.py

"""
一键执行智能引擎 - 临时脚本
"""
import asyncio
import os
import sys
import time
from pathlib import Path
from datetime import datetime, timedelta
import logging

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

# 代理配置（从环境变量读取，海外部署时通过 Docker 环境变量注入）
logger = logging.getLogger(__name__)
http_proxy = os.getenv("HTTP_PROXY")
https_proxy = os.getenv("HTTPS_PROXY")
if http_proxy or https_proxy:
    logger.info(f"使用代理：HTTP={http_proxy}, HTTPS={https_proxy}")

from dotenv import load_dotenv
load_dotenv()

from agent.core.runner import AgentRunner
from agent.trace import FileSystemTraceStore, Trace, Message
from agent.llm import create_openrouter_llm_call
from examples.auto_put_ad_mini.logging_setup import setup_logging, get_trace_id

from examples.auto_put_ad_mini.config import (
    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
)

# 导入自定义工具
from examples.auto_put_ad_mini.tools.data_query import fetch_creative_data, merge_creative_data
from examples.auto_put_ad_mini.tools.roi_calculator import calculate_roi_metrics
from examples.auto_put_ad_mini.tools.creative_roi_calculator import calculate_creative_roi
from examples.auto_put_ad_mini.tools.portfolio_metrics import calculate_portfolio_summary
from examples.auto_put_ad_mini.tools.ad_decision import get_ads_for_review, apply_decisions, query_ad_detail, modify_decisions
from examples.auto_put_ad_mini.tools.report_generator import generate_report
from examples.auto_put_ad_mini.tools.guardrails import validate_decisions
from examples.auto_put_ad_mini.tools.execution_engine import execute_decisions, check_execution_feedback
from examples.auto_put_ad_mini.tools.im_approval import send_approval_request, check_approval_status, send_feishu_text_message

# 尝试导入飞书文档工具（如果存在）
try:
    from examples.auto_put_ad_mini.tools.feishu_doc import import_to_feishu
except ImportError:
    pass  # 工具不存在，忽略


async def main():
    """广告智能调控主流程：数据拉取 → ROI计算 → 候选筛选 → AI推理 → 护栏验证 → 报告生成。"""
    base_dir = Path(__file__).parent

    setup_logging(level=LOG_LEVEL)
    pipeline_logger = logging.getLogger("auto_put_ad_mini.pipeline")
    trace_id = get_trace_id() or "-"
    started_ms = int(time.monotonic() * 1000)

    pipeline_logger.info(
        "event=pipeline stage=run result=started trace_id=%s date=%s",
        trace_id, target_date if 'target_date' in dir() else "auto",
    )

    # 加载 system prompt
    prompt_path = base_dir / "prompts" / "system.prompt"
    system_prompt = ""
    if prompt_path.exists():
        system_prompt = prompt_path.read_text(encoding="utf-8")

    # 加载 presets
    presets_path = base_dir / "presets.json"
    if presets_path.exists():
        from agent.core.presets import load_presets_from_json
        load_presets_from_json(str(presets_path))

    store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
    runner = AgentRunner(
        trace_store=store,
        llm_call=create_openrouter_llm_call(model=MAIN_CONFIG.model),
        skills_dir=SKILLS_DIR if Path(SKILLS_DIR).exists() else None,
        logger_name="agents.auto_put_ad_mini",
    )

    config = MAIN_CONFIG
    if system_prompt:
        config.system_prompt = system_prompt

    print("=" * 70)
    print("  广告智能调控助手 — 智能引擎执行")
    print("=" * 70)
    print()
    print("🚀 自动执行：分析广告")
    print()
    print("=" * 70)
    print("  流程：数据拉取 → ROI计算 → 人群包基线 → 候选筛选 → AI推理 → 保存决策 → 护栏验证 → 生成报告")
    print("=" * 70)
    print()

    # 自动取 T-1（昨天）作为数据截止日期，避免硬编码
    target_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    target_date_display = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    messages = [{"role": "user", "content": f"分析广告，执行完整的ROI计算和决策流程。请使用 {target_date_display}（end_date={target_date}）作为数据截止日期，因为当天数据尚未回流。"}]
    config.trace_id = None

    step_count = 0

    try:
        async for item in runner.run(messages=messages, config=config):
            if isinstance(item, Trace):
                if item.status == "completed":
                    print(f"\n✅ [Trace] 完成")
                elif item.status == "failed":
                    print(f"\n❌ [Trace] 失败")

            elif isinstance(item, Message):
                if item.role == "assistant" and item.content:
                    content = item.content
                    text = content.get("text", "") if isinstance(content, dict) else content
                    if text and text.strip():
                        print(f"\n💭 {text}\n")

                elif item.role == "tool" and item.content:
                    content = item.content
                    if isinstance(content, dict):
                        tool_name = content.get("tool_name", "unknown")
                        result = content.get("result", content.get("text", str(content)))

                        # 识别关键步骤
                        _step_label = None
                        if tool_name == "fetch_creative_data":
                            step_count += 1
                            _step_label = "数据拉取"
                        elif tool_name == "calculate_roi_metrics":
                            step_count += 1
                            _step_label = "ROI 计算"
                        elif tool_name == "calculate_portfolio_summary":
                            step_count += 1
                            _step_label = "人群包基线计算"
                        elif tool_name == "get_ads_for_review":
                            step_count += 1
                            _step_label = "候选筛选"
                        elif tool_name == "apply_decisions":
                            step_count += 1
                            _step_label = "保存智能引擎决策"
                        elif tool_name == "validate_decisions":
                            step_count += 1
                            _step_label = "安全护栏验证"
                        elif tool_name == "execute_decisions":
                            step_count += 1
                            _step_label = "分级执行"
                        elif tool_name == "send_approval_request":
                            step_count += 1
                            _step_label = "IM 审批请求"
                        elif tool_name == "generate_report":
                            step_count += 1
                            _step_label = "生成最终报告"
                        elif tool_name == "check_execution_feedback":
                            step_count += 1
                            _step_label = "执行效果检查"

                        if _step_label:
                            step_slug = _step_label.replace(" ", "_")
                            pipeline_logger.info(
                                "event=step stage=%s step=%d result=started trace_id=%s",
                                step_slug, step_count, trace_id,
                            )
                            print(f"\n{'='*70}")
                            print(f"📌 步骤 {step_count}: {_step_label}")
                            print(f"{'='*70}")

                        # 打印简化结果
                        if isinstance(result, str):
                            text = result
                        else:
                            text = str(result)
                        if len(text) > 500:
                            text = text[:500] + "..."
                        print(f"  {text}")

        print("\n" + "=" * 70)
        print("✅ 执行完成")
        print("=" * 70)
        print()
        print("📁 输出文件：")
        print(f"  - 智能引擎决策：examples/auto_put_ad_mini/outputs/reports/llm_decisions_*.csv")
        print(f"  - 最终报告（带格式）：examples/auto_put_ad_mini/outputs/reports/decision_*.xlsx")
        print()

        pipeline_logger.info(
            "event=pipeline stage=run result=completed duration_ms=%d steps=%d trace_id=%s",
            int(time.monotonic() * 1000) - started_ms, step_count, trace_id,
        )

    except Exception as e:
        print(f"\n❌ 执行失败: {e}")
        import traceback
        traceback.print_exc()
        pipeline_logger.error(
            "event=pipeline stage=run result=failed duration_ms=%d error=%s trace_id=%s",
            int(time.monotonic() * 1000) - started_ms, str(e)[:200], trace_id,
        )


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