Agent/examples/auto_put_ad/tools/execute_agent.py

"""
执行 Agent — 读取确认后的调整方案并执行 API 操作

支持两种输入方式：
1. 直接传入 adjustment_plan 列表（从分析 Agent 输出获取）
2. 传入 Excel 文件路径（运营在 Excel 中确认/修改后）

主要功能：
- 出价调整（increase / decrease）
- 广告关停（close → AD_STATUS_SUSPEND）
- 冷启动保护（跳过 is_cold_start=True 的广告）
- 赔付门槛保护（跳过 conversions_count 3-5 的关停）
- Dry-run 模式（只验证不执行）
- 执行后自动触发监控检查
"""

import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from agent.tools import tool
from agent.tools.models import ToolContext, ToolResult

from examples.auto_put_ad.tools.ad_api import ad_batch_update_status, ad_update
from examples.auto_put_ad.tools.budget_calc import MIN_BID, MAX_BID
from examples.auto_put_ad.tools.monitor_tools import monitor_check_metrics

logger = logging.getLogger(__name__)


def _load_plan_from_excel(excel_path: str) -> List[Dict]:
    """从 Excel 文件加载调整方案"""
    import pandas as pd

    df = pd.read_excel(excel_path)

    # 映射中文列名到英文字段（兼容两种格式）
    column_map = {
        "广告ID": "ad_id",
        "广告名称": "ad_name",
        "账户ID": "account_id",
        "动作": "action",
        "当前出价(分)": "current_bid",
        "新出价(分)": "new_bid",
        "调整幅度": "adjustment_ratio",
        "ROI等级": "roi_level",
        "跑量等级": "volume_level",
        "效率分": "efficiency",
        "昨日消耗(元)": "cost",
        "打开数": "open_count",
        "转化数": "conversions_count",
        "冷启动": "is_cold_start",
        "冷启动原因": "cold_start_reason",
        "广告状态": "ad_status",
    }

    # 重命名已有的中文列
    rename_cols = {k: v for k, v in column_map.items() if k in df.columns}
    if rename_cols:
        df = df.rename(columns=rename_cols)

    # 转换数据类型
    for col in ["ad_id", "account_id", "current_bid", "new_bid", "open_count", "conversions_count"]:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0).astype(int)
    if "is_cold_start" in df.columns:
        df["is_cold_start"] = df["is_cold_start"].astype(bool)
    if "cost" in df.columns:
        df["cost"] = pd.to_numeric(df["cost"], errors="coerce").fillna(0.0)

    return df.to_dict("records")


def _validate_plan(plan: List[Dict]) -> tuple:
    """验证调整方案，返回 (validated_plan, warnings, errors)

    验证规则：
    1. 出价范围 [MIN_BID, MAX_BID]
    2. 单次调整幅度 <= 15%
    3. 冷启动广告跳过
    4. 赔付门槛保护
    """
    validated = []
    warnings = []
    errors = []
    skip_cold_start = 0
    skip_payback = 0

    for item in plan:
        ad_id = item.get("ad_id", "unknown")
        action = item.get("action", "")

        # 跳过非操作项
        if action not in ("increase", "decrease", "close"):
            continue

        # 冷启动保护
        if item.get("is_cold_start", False):
            skip_cold_start += 1
            warnings.append(f"  跳过 ad_id={ad_id}: {item.get('cold_start_reason', '冷启动保护')}")
            continue

        # 赔付门槛保护（仅针对 close）
        conversions = item.get("conversions_count", 0)
        if action == "close" and 3 <= conversions < 6:
            skip_payback += 1
            warnings.append(f"  跳过关停 ad_id={ad_id}: 转化数={conversions}，接近赔付门槛")
            continue

        # 出价范围检查
        new_bid = item.get("new_bid")
        if new_bid is not None and action in ("increase", "decrease"):
            if new_bid < MIN_BID:
                warnings.append(f"  ad_id={ad_id}: 出价 {new_bid} < {MIN_BID}，自动调整为 {MIN_BID}")
                item["new_bid"] = MIN_BID
            elif new_bid > MAX_BID:
                warnings.append(f"  ad_id={ad_id}: 出价 {new_bid} > {MAX_BID}，自动调整为 {MAX_BID}")
                item["new_bid"] = MAX_BID

        # 调整幅度检查：> 15% 自动 clamp，不放行
        current_bid = item.get("current_bid")
        if current_bid and new_bid and action in ("increase", "decrease"):
            ratio = abs(new_bid - current_bid) / current_bid
            if ratio > 0.15:
                # 自动 clamp 到 15% 幅度
                if action == "increase":
                    clamped_bid = int(current_bid * 1.15)
                else:
                    clamped_bid = int(current_bid * 0.85)
                clamped_bid = max(MIN_BID, min(MAX_BID, clamped_bid))
                warnings.append(
                    f"  ad_id={ad_id}: 调整幅度 {ratio:.0%} > 15%，"
                    f"已 clamp: {new_bid} -> {clamped_bid} (限制在 15% 内)"
                )
                item["new_bid"] = clamped_bid

        validated.append(item)

    if skip_cold_start:
        warnings.insert(0, f"冷启动保护跳过: {skip_cold_start} 个广告")
    if skip_payback:
        warnings.insert(0 if not skip_cold_start else 1, f"赔付门槛保护跳过: {skip_payback} 个广告")

    return validated, warnings, errors


@tool(description="加载并执行运营确认后的出价调整方案（支持 Excel 文件或直接传入方案列表）")
async def execute_adjustment_plan(
    account_id: int,
    excel_path: str = "",
    adjustment_plan: Optional[List[Dict]] = None,
    dry_run: bool = False,
    execute_close: bool = False,
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """加载确认后的方案并执行出价调整和广告关停。

    Args:
        account_id: 账户ID（传 0 则使用方案中每条记录的 account_id）
        excel_path: Excel 文件路径（与 adjustment_plan 二选一）
        adjustment_plan: 调整方案列表（与 excel_path 二选一）
        dry_run: True=只验证不执行，输出模拟报告
        execute_close: True=同时执行关停操作（默认不执行，需运营单独确认）
    """
    try:
        # === 阶段一：加载方案 ===
        source = ""
        if excel_path:
            path = Path(excel_path)
            if not path.exists():
                return ToolResult(title="执行失败", output=f"Excel 文件不存在: {excel_path}")
            plan = _load_plan_from_excel(excel_path)
            source = f"Excel: {path.name}"
        elif adjustment_plan:
            plan = adjustment_plan
            source = "上游传入"
        else:
            return ToolResult(title="执行失败", output="请提供 excel_path 或 adjustment_plan")

        if not plan:
            return ToolResult(title="执行失败", output="方案为空，无可执行项")

        # === 阶段二：验证 ===
        validated, warnings, errors = _validate_plan(plan)

        mode = "Dry-Run" if dry_run else "正式执行"
        report_lines = [
            "=" * 50,
            f"执行报告",
            "=" * 50,
            f"方案来源: {source}",
            f"执行模式: {mode}",
            f"方案总数: {len(plan)} 条",
            f"有效操作: {len(validated)} 条",
            "",
        ]

        if warnings:
            report_lines.append("验证警告:")
            report_lines.extend(warnings)
            report_lines.append("")

        if errors:
            report_lines.append("验证错误:")
            report_lines.extend(errors)
            report_lines.append("")

        # 分类统计
        increase_items = [v for v in validated if v["action"] == "increase"]
        decrease_items = [v for v in validated if v["action"] == "decrease"]
        close_items = [v for v in validated if v["action"] == "close"]
        bid_items = increase_items + decrease_items

        report_lines.append(f"出价调整: 提价 {len(increase_items)} 个 + 降价 {len(decrease_items)} 个 = {len(bid_items)} 个")
        report_lines.append(f"广告关停: {len(close_items)} 个 {'(本次执行)' if execute_close else '(待确认)'}")
        report_lines.append("")

        # === Dry-Run 模式：只输出不执行 ===
        if dry_run:
            if bid_items:
                report_lines.append("将要执行的出价调整:")
                for item in bid_items[:20]:
                    report_lines.append(
                        f"  ad_id={item['ad_id']} | {item['action']} | "
                        f"{item.get('current_bid', '?')} -> {item.get('new_bid', '?')} 分"
                    )
                if len(bid_items) > 20:
                    report_lines.append(f"  ... 还有 {len(bid_items) - 20} 个")

            if close_items:
                report_lines.append("\n将要关停的广告:")
                for item in close_items[:10]:
                    report_lines.append(
                        f"  ad_id={item['ad_id']} | 消耗:{item.get('cost', 0):.0f}元 | "
                        f"转化:{item.get('conversions_count', 0)}"
                    )

            report_lines.append("\n[Dry-Run] 以上操作未实际执行")
            return ToolResult(
                title=f"Dry-Run 报告（{len(validated)} 条有效操作）",
                output="\n".join(report_lines),
                metadata={"mode": "dry_run", "validated_count": len(validated)},
            )

        # === 阶段三：执行出价调整 ===
        bid_success = 0
        bid_failed = 0
        bid_errors = []

        if bid_items:
            report_lines.append("--- 执行出价调整 ---")
            for item in bid_items:
                try:
                    target_account = account_id if account_id else item.get("account_id", 0)
                    await ad_update(
                        account_id=target_account,
                        adgroup_id=item["ad_id"],
                        bid_amount=item["new_bid"],
                    )
                    bid_success += 1
                    logger.info(
                        "出价调整成功: ad_id=%s, %s -> %s (%s)",
                        item["ad_id"], item.get("current_bid"), item["new_bid"], item["action"]
                    )
                except Exception as e:
                    bid_failed += 1
                    err_msg = f"ad_id={item['ad_id']}: {str(e)}"
                    bid_errors.append(err_msg)
                    logger.error("出价调整失败: %s", err_msg)

            report_lines.append(f"提价(increase): {len(increase_items)} 个")
            report_lines.append(f"降价(decrease): {len(decrease_items)} 个")
            report_lines.append(f"成功: {bid_success} / 失败: {bid_failed}")
            report_lines.append("")

        # === 阶段四：执行广告关停 ===
        close_success = 0
        close_failed = 0
        close_errors = []

        if close_items and execute_close:
            report_lines.append("--- 执行广告关停 ---")

            # 按 account_id 分组，避免跨账户批量操作
            from collections import defaultdict
            close_by_account = defaultdict(list)
            for item in close_items:
                target_acct = account_id if account_id else item.get("account_id", 0)
                close_by_account[target_acct].append(item["ad_id"])

            for target_acct, ad_ids in close_by_account.items():
                try:
                    result = await ad_batch_update_status(
                        adgroup_ids=ad_ids,
                        configured_status="AD_STATUS_SUSPEND",
                        account_id=target_acct,
                    )
                    if "失败" not in result.title:
                        close_success += len(ad_ids)
                        logger.info("关停成功: 账户%s, %d 个广告", target_acct, len(ad_ids))
                    else:
                        close_failed += len(ad_ids)
                        close_errors.append(f"账户{target_acct}: {result.output}")
                        logger.error("关停失败: 账户%s, %s", target_acct, result.output)
                except Exception as e:
                    close_failed += len(ad_ids)
                    close_errors.append(f"账户{target_acct}: {str(e)}")
                    logger.error("关停异常: 账户%s, %s", target_acct, e)

            report_lines.append(f"关停: 成功 {close_success} / 失败 {close_failed}")
            report_lines.append("")
        elif close_items and not execute_close:
            report_lines.append("--- 广告关停（待确认）---")
            report_lines.append(f"以下 {len(close_items)} 个广告建议关停，请确认后使用 execute_close=True 执行：")
            for item in close_items[:10]:
                report_lines.append(
                    f"  ad_id={item['ad_id']} | 消耗:{item.get('cost', 0):.0f}元 | "
                    f"转化:{item.get('conversions_count', 0)} | 效率:{item.get('efficiency', 0)}"
                )
            if len(close_items) > 10:
                report_lines.append(f"  ... 还有 {len(close_items) - 10} 个")
            report_lines.append("")

        # === 阶段五：执行后监控 ===
        monitor_result = None
        if bid_success > 0 or close_success > 0:
            report_lines.append("--- 执行后监控 ---")
            try:
                target_account = account_id if account_id else 0
                monitor_result = await monitor_check_metrics(
                    account_id=target_account,
                    check_items=["cost_spike", "budget_overrun"],
                    time_window="1h",
                )
                report_lines.append(f"监控结果: {monitor_result.title}")
                if monitor_result.metadata and monitor_result.metadata.get("anomalies"):
                    for a in monitor_result.metadata["anomalies"]:
                        report_lines.append(f"  {a.get('type', '')}: {a.get('message', '')}")
                else:
                    report_lines.append("  无异常")
            except Exception as e:
                report_lines.append(f"监控检查失败: {str(e)}")
                logger.warning("执行后监控失败: %s", e)
            report_lines.append("")

        # === 失败详情 ===
        all_errors = bid_errors + close_errors
        if all_errors:
            report_lines.append("--- 失败详情 ---")
            for err in all_errors[:20]:
                report_lines.append(f"  {err}")
            if len(all_errors) > 20:
                report_lines.append(f"  ... 还有 {len(all_errors) - 20} 个错误")
            report_lines.append("")

        # === 汇总 ===
        report_lines.append("=" * 50)
        total_success = bid_success + close_success
        total_failed = bid_failed + close_failed
        report_lines.append(f"总计: 成功 {total_success} / 失败 {total_failed}")
        report_lines.append(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

        return ToolResult(
            title=f"执行完成（成功{total_success}/失败{total_failed}）",
            output="\n".join(report_lines),
            metadata={
                "mode": "execute",
                "bid_success": bid_success,
                "bid_failed": bid_failed,
                "close_success": close_success,
                "close_failed": close_failed,
                "close_pending": len(close_items) if not execute_close else 0,
                "errors": all_errors,
            },
        )

    except Exception as e:
        logger.error("execute_adjustment_plan 失败: %s", e, exc_info=True)
        return ToolResult(title="执行失败", output=str(e))