"""
执行引擎 — auto_put_ad_mini

职责：
  1. 加载护栏验证后的决策
  2. 按自治级别分类（Tier 1 自动 / Tier 2 审批 / Tier 3 阻断）
  3. 调用腾讯广告 API 执行操作
  4. 记录审计日志（JSONL）
  5. 执行后效果检查

依赖：
  - tools/ad_api.py 的底层 HTTP 函数
  - tools/guardrails.py 的 AdjustmentHistory
"""

import asyncio
import json
import logging
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd

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

_MINI_DIR = Path(__file__).resolve().parent.parent
_TOOLS_DIR = Path(__file__).resolve().parent
if str(_MINI_DIR) not in sys.path:
    sys.path.insert(0, str(_MINI_DIR))
if str(_TOOLS_DIR) not in sys.path:
    sys.path.insert(0, str(_TOOLS_DIR))

from config import (
    EXECUTION_ENABLED,
    EXECUTION_LOG_DIR,
    API_QPS_LIMIT,
    API_MAX_RETRIES,
    TIER1_MAX_CHANGE_PCT,
    TIER3_MIN_DAILY_SPEND,
    FEEDBACK_CHECK_HOURS,
    DRY_RUN_MODE,
    IM_ENABLED,
)

logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════
# QPS 令牌桶限流
# ═══════════════════════════════════════════


class TokenBucket:
    """简单令牌桶限流器。"""

    def __init__(self, rate: float = API_QPS_LIMIT):
        self._rate = rate
        self._tokens = rate
        self._last_refill = time.monotonic()

    async def acquire(self):
        """获取一个令牌，不够时等待。"""
        while True:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self._rate, self._tokens + elapsed * self._rate)
            self._last_refill = now

            if self._tokens >= 1:
                self._tokens -= 1
                return
            else:
                wait = (1 - self._tokens) / self._rate
                await asyncio.sleep(wait)


# ═══════════════════════════════════════════
# 腾讯广告执行器
# ═══════════════════════════════════════════


class TencentAdExecutor:
    """
    腾讯广告 API 执行器。

    复用 ad_api.py 的底层 HTTP 函数，添加：
    - QPS 令牌桶限流
    - 指数退避重试
    - 操作前快照
    """

    def __init__(self):
        self._bucket = TokenBucket(rate=API_QPS_LIMIT)

    async def get_ad_state(self, ad_id: int, account_id: int) -> Optional[Dict]:
        """获取广告当前状态快照。"""
        try:
            from ad_api import _get, _check
            await self._bucket.acquire()
            resp = _get("/adgroups/get", {
                "account_id": account_id,
                "filtering": {"adgroup_id_list": [ad_id]},
                "page": 1,
                "page_size": 1,
            })
            data = _check(resp, "get_ad_state")
            items = data.get("list", [])
            return items[0] if items else None
        except Exception as e:
            logger.warning("获取广告 %s 状态失败: %s", ad_id, e)
            return None

    async def update_bid(self, ad_id: int, account_id: int, bid_amount_fen: int) -> Dict:
        """更新广告出价（分）。"""
        # 白名单安全检查（最后防线）
        from config import WHITELIST_ENABLED, WHITELIST_ACCOUNTS
        if WHITELIST_ENABLED and account_id not in WHITELIST_ACCOUNTS:
            logger.error(
                f"⚠️ 白名单安全阻断：账户 {account_id} 不在白名单内，拒绝执行 update_bid。"
                f"白名单: {WHITELIST_ACCOUNTS}"
            )
            return {"code": -1, "message": f"账户 {account_id} 不在白名单内"}

        # 预先获取并验证 access_token
        from ad_api import _post, _check, _get_access_token
        try:
            token = _get_access_token(account_id)
            if not token or len(token) < 10:
                return {"code": -1, "message": f"获取 access_token 失败：token={token[:20]}"}
            logger.info(f"[update_bid] 已获取 access_token (账户={account_id}): {token[:10]}...")
        except Exception as e:
            logger.error(f"[update_bid] 获取 access_token 异常: {e}")
            return {"code": -1, "message": f"获取 access_token 异常: {e}"}

        for attempt in range(API_MAX_RETRIES):
            try:
                await self._bucket.acquire()
                body = {
                    "account_id": account_id,
                    "adgroup_id": ad_id,
                    "bid_amount": bid_amount_fen,
                }
                resp = _post("/adgroups/update", body)
                _check(resp, "update_bid")
                return {"code": 0, "message": "success"}
            except Exception as e:
                if attempt < API_MAX_RETRIES - 1:
                    wait = 2 ** attempt
                    logger.warning("update_bid 重试 %d/%d (等待%ds): %s", attempt + 1, API_MAX_RETRIES, wait, e)
                    await asyncio.sleep(wait)
                else:
                    return {"code": -1, "message": str(e)}

    async def pause_creative(self, creative_id: int, account_id: int) -> Dict:
        """暂停单条动态创意 (v3.0: /dynamic_creatives/update)。"""
        from config import WHITELIST_ENABLED, WHITELIST_ACCOUNTS
        if WHITELIST_ENABLED and account_id not in WHITELIST_ACCOUNTS:
            logger.error(
                f"⚠️ 白名单安全阻断:账户 {account_id} 不在白名单内,拒绝执行 pause_creative。"
                f"白名单: {WHITELIST_ACCOUNTS}"
            )
            return {"code": -1, "message": f"账户 {account_id} 不在白名单内"}

        from ad_api import _post, _check, _get_access_token
        try:
            token = _get_access_token(account_id)
            if not token or len(token) < 10:
                return {"code": -1, "message": f"获取 access_token 失败:token={token[:20]}"}
        except Exception as e:
            logger.error(f"[pause_creative] 获取 access_token 异常: {e}")
            return {"code": -1, "message": f"获取 access_token 异常: {e}"}

        for attempt in range(API_MAX_RETRIES):
            try:
                await self._bucket.acquire()
                body = {
                    "account_id": account_id,
                    "dynamic_creative_id": creative_id,
                    "configured_status": "AD_STATUS_SUSPEND",
                }
                resp = _post("/dynamic_creatives/update", body)
                _check(resp, "pause_creative")
                return {"code": 0, "message": "success"}
            except Exception as e:
                if attempt < API_MAX_RETRIES - 1:
                    wait = 2 ** attempt
                    logger.warning("pause_creative 重试 %d/%d: %s", attempt + 1, API_MAX_RETRIES, e)
                    await asyncio.sleep(wait)
                else:
                    return {"code": -1, "message": str(e)}

    async def pause_ad(self, ad_id: int, account_id: int) -> Dict:
        """暂停广告。"""
        # 白名单安全检查（最后防线）
        from config import WHITELIST_ENABLED, WHITELIST_ACCOUNTS
        if WHITELIST_ENABLED and account_id not in WHITELIST_ACCOUNTS:
            logger.error(
                f"⚠️ 白名单安全阻断：账户 {account_id} 不在白名单内，拒绝执行 pause_ad。"
                f"白名单: {WHITELIST_ACCOUNTS}"
            )
            return {"code": -1, "message": f"账户 {account_id} 不在白名单内"}

        # 预先获取并验证 access_token
        from ad_api import _post, _check, _get_access_token
        try:
            token = _get_access_token(account_id)
            if not token or len(token) < 10:
                return {"code": -1, "message": f"获取 access_token 失败：token={token[:20]}"}
            logger.info(f"[pause_ad] 已获取 access_token (账户={account_id}): {token[:10]}...")
        except Exception as e:
            logger.error(f"[pause_ad] 获取 access_token 异常: {e}")
            return {"code": -1, "message": f"获取 access_token 异常: {e}"}

        for attempt in range(API_MAX_RETRIES):
            try:
                await self._bucket.acquire()
                body = {
                    "account_id": account_id,
                    "adgroup_id": ad_id,
                    "configured_status": "AD_STATUS_SUSPEND",
                }
                resp = _post("/adgroups/update", body)
                _check(resp, "pause_ad")
                return {"code": 0, "message": "success"}
            except Exception as e:
                if attempt < API_MAX_RETRIES - 1:
                    wait = 2 ** attempt
                    logger.warning("pause_ad 重试 %d/%d: %s", attempt + 1, API_MAX_RETRIES, e)
                    await asyncio.sleep(wait)
                else:
                    return {"code": -1, "message": str(e)}


# ═══════════════════════════════════════════
# 审计日志
# ═══════════════════════════════════════════


class AuditLogger:
    """JSONL 审计日志。"""

    def __init__(self, log_dir: Path = EXECUTION_LOG_DIR):
        self._log_dir = log_dir
        self._log_dir.mkdir(parents=True, exist_ok=True)
        today = datetime.now().strftime("%Y%m%d")
        self._path = self._log_dir / f"exec_{today}.jsonl"

    def log(self, entry: Dict):
        """追加一条审计记录（自动补时间戳，JSONL 格式写入）。"""
        entry["ts"] = datetime.now().isoformat()
        with open(self._path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    @property
    def path(self) -> Path:
        """当前审计日志文件路径。"""
        return self._path


# ═══════════════════════════════════════════
# 创意级 pause 解析与执行
# ═══════════════════════════════════════════


def _parse_creative_ids(cell) -> List[str]:
    """从 CSV 单元格解析 pause_creative_ids。

    DataFrame 内是 Python list,写入 CSV 后变 "['c1','c2']" 字符串,
    再读回时变字符串。统一解析为字符串列表。
    """
    if cell is None:
        return []
    if isinstance(cell, list):
        return [str(c) for c in cell]
    s = str(cell).strip()
    if not s or s.lower() == "nan" or s == "[]":
        return []
    try:
        import ast
        v = ast.literal_eval(s)
        if isinstance(v, (list, tuple)):
            return [str(c) for c in v]
        return [str(v)]
    except Exception:
        # 兜底:逗号分隔
        return [c.strip().strip("'\"") for c in s.strip("[]").split(",") if c.strip()]


async def _execute_pause(
    executor: "TencentAdExecutor",
    row: pd.Series,
    audit: "AuditLogger",
    history: "AdjustmentHistory",
) -> Dict[str, Any]:
    """根据 pause_scope 路由到广告级或创意级 pause。

    返回: {code, message, exec_status, scope, creative_results}
    创意级 pause 严格按 ad_id 串行(腾讯 v3.0 同 adgroup 创意更新串行约束)。
    """
    ad_id = int(row["ad_id"])
    account_id = int(row.get("account_id", 0) or 0)
    scope = str(row.get("pause_scope", "ad") or "ad").strip()

    if scope != "creatives":
        # 广告级 pause(原路径)
        result = await executor.pause_ad(ad_id, account_id)
        return {
            "code": result.get("code", -1),
            "message": result.get("message", ""),
            "exec_status": "success" if result.get("code") == 0 else "failed",
            "scope": "ad",
            "creative_results": [],
        }

    # 创意级 pause:同 ad_id 下逐个串行
    creative_ids = _parse_creative_ids(row.get("pause_creative_ids"))
    if not creative_ids:
        logger.warning(
            f"广告 {ad_id} pause_scope=creatives 但 pause_creative_ids 为空,降级为广告级 pause"
        )
        result = await executor.pause_ad(ad_id, account_id)
        return {
            "code": result.get("code", -1),
            "message": result.get("message", ""),
            "exec_status": "success" if result.get("code") == 0 else "failed",
            "scope": "ad_fallback",
            "creative_results": [],
        }

    creative_results = []
    success_count = 0
    fail_count = 0

    # 持久化:创意级 pause 历史(用于后续速率限制)
    try:
        from guardrails import CreativePauseHistory
        creative_history = CreativePauseHistory()
    except Exception as e:
        logger.warning(f"创意级 pause 历史加载失败: {e}")
        creative_history = None

    for cid in creative_ids:
        try:
            cid_int = int(cid)
        except (ValueError, TypeError):
            logger.warning(f"广告 {ad_id} 创意 ID 解析失败: {cid}")
            creative_results.append({
                "creative_id": cid, "code": -1, "message": "creative_id 非整数",
            })
            fail_count += 1
            continue

        r = await executor.pause_creative(cid_int, account_id)
        rcode = r.get("code", -1)
        creative_results.append({
            "creative_id": cid_int,
            "code": rcode,
            "message": r.get("message", ""),
        })
        if rcode == 0:
            success_count += 1
            if creative_history is not None:
                try:
                    creative_history.record_pause(ad_id, cid_int, action="pause")
                except Exception as e:
                    logger.warning(f"记录创意 pause 历史失败 ({ad_id}:{cid_int}): {e}")
        else:
            fail_count += 1

    if success_count == len(creative_ids):
        exec_status = "success"
    elif success_count > 0:
        exec_status = "partial"
    else:
        exec_status = "failed"

    return {
        "code": 0 if exec_status == "success" else -1,
        "message": f"创意级 pause: 成功 {success_count}/{len(creative_ids)}, 失败 {fail_count}",
        "exec_status": exec_status,
        "scope": "creatives",
        "creative_results": creative_results,
    }


# ═══════════════════════════════════════════
# 自治级别分类
# ═══════════════════════════════════════════


def _classify_tier(row: pd.Series) -> int:
    """
    自治级别分类。

    Tier 0 (无操作): hold, observe, creative_adjust, scale_up（不调用API，仅建议）
    Tier 1 (自动执行): bid 调幅 ≤ 5%
    Tier 2 (需审批):   pause 或 bid 调幅 > 5%
    Tier 3 (高价值阻断): 日消耗 > 1500 元的高价值广告
    """
    action = row.get("final_action", row.get("action", "hold"))

    # Tier 0: 无需操作的动作（仅建议，不执行）
    if action in ("hold", "observe", "creative_adjust", "scale_up"):
        return 0  # 无操作（observe=观察，creative_adjust/scale_up=需人工执行）

    cost_7d_avg = float(row.get("cost_7d_avg", 0) or 0)
    change_pct = row.get("recommended_change_pct", 0)
    if isinstance(change_pct, str):
        try:
            change_pct = float(change_pct)
        except ValueError:
            change_pct = 0

    # Tier 3: 高价值广告
    if cost_7d_avg >= TIER3_MIN_DAILY_SPEND:
        return 3

    # Tier 2: 暂停 或 大幅调价
    if action == "pause" or abs(change_pct) > TIER1_MAX_CHANGE_PCT:
        return 2

    # Tier 1: 小幅调价
    return 1


# ═══════════════════════════════════════════
# 工具：执行已验证的决策
# ═══════════════════════════════════════════


@tool(description="执行已验证的决策：分级自治 → API调用 → 审计日志（部分批准型场景请传 filter_ad_ids）")
async def execute_decisions(
    ctx: ToolContext = None,
    validated_csv: str = "",
    approval_mode: str = "tiered",
    filter_ad_ids: Optional[List[int]] = None,
) -> ToolResult:
    """
    执行已通过护栏验证的决策。

    Pipeline:
    1. 加载 validated_decisions CSV (guardrail_status != blocked)
    2. 按自治级别分类
    3. Tier 1 (小调, ≤5%): 直接执行 + 通知
    4. Tier 2 (大调/暂停): 标记待审批
    5. Tier 3 (高价值): 标记需人工审批
    6. 记录审计日志

    Args:
        validated_csv: 护栏验证后的 CSV 路径
        approval_mode: "auto"=全部自动执行, "tiered"=分级, "manual"=全部需审批
        filter_ad_ids: 仅执行该 ad_id 列表中的决策（部分批准型协议主路径）；为 None 时执行全部
    """
    try:
        import time as _time
        _started_ms = int(_time.monotonic() * 1000)
        logger.info(
            "event=execute_decisions stage=run result=started "
            "validated_csv=%s approval_mode=%s filter_ids=%s",
            validated_csv or "auto", approval_mode,
            len(filter_ad_ids) if filter_ad_ids else "all",
        )

        if not EXECUTION_ENABLED:
            return ToolResult(
                title="执行引擎未启用",
                output="EXECUTION_ENABLED=False，跳过执行。修改 config.py 启用。",
            )

        if DRY_RUN_MODE:
            return ToolResult(
                title="干运行模式",
                output="DRY_RUN_MODE=True，操作不会实际执行。修改 config.py 关闭干运行。",
            )

        # 查找验证 CSV
        if not validated_csv:
            reports_dir = _MINI_DIR / "outputs" / "reports"
            candidates = sorted(reports_dir.glob("validated_decisions_*.csv"), reverse=True)
            if not candidates:
                return ToolResult(title="execute_decisions", output="未找到验证后的决策 CSV")
            validated_csv = str(candidates[0])

        df = pd.read_csv(validated_csv)
        if df.empty:
            return ToolResult(title="execute_decisions", output="决策数据为空")

        # 只执行 approved / modified（非 blocked）
        df_exec = df[df["guardrail_status"].isin(["approved", "modified"])].copy()
        df_exec = df_exec[df_exec["final_action"] != "hold"]

        # ===== 部分批准过滤（filter_ad_ids 优先）=====
        # 部分批准型协议：只执行 approved_ids 子集，其余视为默认拒绝
        if filter_ad_ids is not None:
            try:
                allow_set = {int(x) for x in filter_ad_ids}
            except (TypeError, ValueError) as _e:
                return ToolResult(
                    title="filter_ad_ids 格式错误",
                    output=f"必须是 ad_id 整数列表；收到: {filter_ad_ids}（{_e}）",
                )
            before = len(df_exec)
            df_exec = df_exec[df_exec["ad_id"].astype(int).isin(allow_set)].copy()
            logger.info(
                "部分批准过滤：filter_ad_ids=%d 个，决策从 %d 条筛到 %d 条",
                len(allow_set), before, len(df_exec),
            )
            if df_exec.empty:
                return ToolResult(
                    title="过滤后无可执行决策",
                    output=f"filter_ad_ids={sorted(allow_set)} 与决策 CSV 中的 ad_id 无交集，"
                           f"或这些 ad_id 都被护栏拦截 / 都是 hold",
                )

        if df_exec.empty:
            return ToolResult(
                title="无需执行的操作",
                output="所有决策要么是 hold，要么被护栏拦截",
            )

        # ===== 白名单过滤（仅在执行阶段生效）=====
        # 分析阶段覆盖所有账户，执行阶段仅操作白名单账户
        from config import WHITELIST_ENABLED, WHITELIST_ACCOUNTS
        if WHITELIST_ENABLED:
            if "account_id" in df_exec.columns:
                non_whitelist = df_exec[~df_exec["account_id"].isin(WHITELIST_ACCOUNTS)]
                if not non_whitelist.empty:
                    logger.info(
                        f"白名单过滤：跳过 {len(non_whitelist)} 个非白名单账户的决策，"
                        f"仅执行白名单账户。跳过账户: {non_whitelist['account_id'].unique().tolist()[:10]}"
                    )
                    df_exec = df_exec[df_exec["account_id"].isin(WHITELIST_ACCOUNTS)].copy()
                    if df_exec.empty:
                        return ToolResult(
                            title="白名单过滤后无可执行决策",
                            output=f"所有决策均属于非白名单账户，无需执行API操作。"
                                   f"白名单账户: {WHITELIST_ACCOUNTS[:5]}"
                        )
            else:
                logger.warning("白名单检查：数据中缺少 account_id 列，跳过检查")

        # 分级
        df_exec["tier"] = df_exec.apply(_classify_tier, axis=1)

        executor = TencentAdExecutor()
        audit = AuditLogger()

        # 从 guardrails 导入 AdjustmentHistory 记录操作
        from guardrails import AdjustmentHistory
        history = AdjustmentHistory()

        executed = 0
        failed = 0
        pending_approval = 0
        approved_executed = 0
        rejected_count = 0
        timeout_count = 0
        tier_summary = {1: 0, 2: 0, 3: 0}

        # ─── 分离 Tier 1 和 Tier 2/3 ───
        df_tier1 = df_exec[df_exec["tier"] == 1]
        df_tier2_3 = df_exec[df_exec["tier"] >= 2]

        # ⚠️ 硬开关：TIER1_MAX_CHANGE_PCT <= 0 → 完全禁用自动执行通道
        # 即使分类器误将某条归为 Tier 1，也强制转入审批通道，避免未审批就执行
        if TIER1_MAX_CHANGE_PCT <= 0 and not df_tier1.empty:
            logger.warning(
                "TIER1_MAX_CHANGE_PCT=%s ≤ 0，自动执行通道已禁用；强制将 %d 条 Tier 1 决策转入审批",
                TIER1_MAX_CHANGE_PCT, len(df_tier1),
            )
            df_tier1_forced = df_tier1.copy()
            df_tier1_forced["tier"] = 2  # 提升到 Tier 2 走审批
            df_tier2_3 = pd.concat([df_tier2_3, df_tier1_forced], ignore_index=True)
            df_tier1 = df_tier1.iloc[0:0]  # 清空 Tier 1

        for t in [1, 2, 3]:
            tier_summary[t] = int((df_exec["tier"] == t).sum() if not df_exec.empty else 0)

        # ═══ Phase 1: 自动执行 Tier 1 ═══
        # 注意：当 TIER1_MAX_CHANGE_PCT <= 0 时 df_tier1 已被清空，此循环不会执行
        for _, row in df_tier1.iterrows():
            action = row.get("final_action", row.get("action"))
            ad_id = int(row["ad_id"])
            account_id = int(row.get("account_id", 0) or 0)

            pre_state = await executor.get_ad_state(ad_id, account_id)
            pause_extra: Dict[str, Any] = {}

            if action == "pause":
                pause_result = await _execute_pause(executor, row, audit, history)
                result = {"code": pause_result["code"], "message": pause_result["message"]}
                pause_extra = {
                    "pause_scope": pause_result["scope"],
                    "creative_results": pause_result["creative_results"],
                }
                exec_status_override = pause_result["exec_status"]
            elif action in ("bid_up", "bid_down"):
                final_bid = row.get("final_bid", row.get("recommended_bid"))
                if final_bid is None or final_bid == "" or pd.isna(final_bid):
                    audit.log({
                        "ad_id": ad_id,
                        "action": action,
                        "tier": 1,
                        "execution_status": "skipped",
                        "reason": "无出价数据（NaN/空）",
                    })
                    continue
                bid_fen = int(float(final_bid) * 100)
                result = await executor.update_bid(ad_id, account_id, bid_fen)
                exec_status_override = None
            else:
                continue

            api_code = result.get("code", -1)
            if exec_status_override is not None:
                exec_status = exec_status_override  # success / partial / failed
            else:
                exec_status = "success" if api_code == 0 else "failed"

            if exec_status in ("success", "partial"):
                executed += 1
                change_pct = row.get("recommended_change_pct", 0)
                if isinstance(change_pct, str):
                    try:
                        change_pct = float(change_pct)
                    except ValueError:
                        change_pct = 0
                history.record_adjustment(str(ad_id), action, change_pct)
            else:
                failed += 1

            logger.info(
                "event=execute_decisions stage=auto_exec ad_id=%d account_id=%d "
                "action=%s tier=1 result=%s api_code=%d",
                ad_id, account_id, action, exec_status, api_code,
            )

            post_state = await executor.get_ad_state(ad_id, account_id) if exec_status in ("success", "partial") else None

            audit_entry = {
                "ad_id": ad_id,
                "account_id": account_id,
                "action": action,
                "tier": 1,
                "pre_state": {
                    "bid_amount": pre_state.get("bid_amount") if pre_state else None,
                    "status": pre_state.get("configured_status") if pre_state else None,
                } if pre_state else None,
                "post_state": {
                    "bid_amount": post_state.get("bid_amount") if post_state else None,
                    "status": post_state.get("configured_status") if post_state else None,
                } if post_state else None,
                "api_code": api_code,
                "api_message": result.get("message", ""),
                "execution_status": exec_status,
                "source": row.get("source", ""),
            }
            if pause_extra:
                audit_entry.update(pause_extra)
            audit.log(audit_entry)

        # ═══ Phase 2: Tier 2/3 — 审批 + 执行 ═══
        if not df_tier2_3.empty:
            # 决定审批状态
            if filter_ad_ids is not None:
                # 部分批准型：filter_ad_ids 已是上游审批结果，直接视为"已批准"，**严禁再发审批表**
                logger.info(
                    "Tier 2/3 共 %d 个操作，filter_ad_ids 已上游批准，跳过 Phase 2 二次审批，直接执行",
                    len(df_tier2_3),
                )
                approval_status = "approved"
                approved_ids = list({int(x) for x in filter_ad_ids})
                rejected_ids = []
            elif IM_ENABLED:
                # 阻塞式审批：调用 send_approval_request(wait_for_reply=True)
                logger.info("Tier 2/3 共 %d 个操作，发送 IM 审批并等待...", len(df_tier2_3))

                from im_approval import send_approval_request
                approval_result = await send_approval_request(
                    ctx=ctx,
                    validated_csv=validated_csv,
                    wait_for_reply=True,
                )

                approval_status = (
                    approval_result.metadata.get("status", "timeout")
                    if approval_result.metadata
                    else "timeout"
                )
                approved_ids = (
                    approval_result.metadata.get("approved_ids", [])
                    if approval_result.metadata
                    else []
                )
                rejected_ids = (
                    approval_result.metadata.get("rejected_ids", [])
                    if approval_result.metadata
                    else []
                )
            else:
                # 兜底：IM 关 + 无 filter_ad_ids，全部标记 timeout
                approval_status = "timeout"
                approved_ids = []
                rejected_ids = []

            # 根据审批状态执行
            if approval_status == "timeout":
                # 超时：所有 Tier 2/3 标记为 timeout
                timeout_count = len(df_tier2_3)
                for _, row in df_tier2_3.iterrows():
                    audit.log({
                        "ad_id": int(row["ad_id"]),
                        "account_id": int(row.get("account_id", 0) or 0),
                        "action": row.get("final_action", row.get("action")),
                        "tier": int(row.get("tier", 2)),
                        "execution_status": "timeout",
                        "source": row.get("source", ""),
                    })
            else:
                # 执行已批准的广告
                approved_set = set(int(x) for x in approved_ids)
                rejected_set = set(int(x) for x in rejected_ids)

                for _, row in df_tier2_3.iterrows():
                    ad_id = int(row["ad_id"])
                    account_id = int(row.get("account_id", 0) or 0)
                    action = row.get("final_action", row.get("action"))
                    tier = int(row.get("tier", 2))

                    if ad_id in rejected_set:
                        rejected_count += 1
                        audit.log({
                            "ad_id": ad_id,
                            "account_id": account_id,
                            "action": action,
                            "tier": tier,
                            "execution_status": "rejected",
                            "source": row.get("source", ""),
                        })
                        continue

                    if ad_id not in approved_set:
                        # 既不在 approved 也不在 rejected（部分审批场景遗漏）
                        pending_approval += 1
                        audit.log({
                            "ad_id": ad_id,
                            "account_id": account_id,
                            "action": action,
                            "tier": tier,
                            "execution_status": "pending_approval",
                            "source": row.get("source", ""),
                        })
                        continue

                    # 已批准 → 执行
                    pre_state = await executor.get_ad_state(ad_id, account_id)
                    pause_extra: Dict[str, Any] = {}

                    if action == "pause":
                        pause_result = await _execute_pause(executor, row, audit, history)
                        result = {"code": pause_result["code"], "message": pause_result["message"]}
                        pause_extra = {
                            "pause_scope": pause_result["scope"],
                            "creative_results": pause_result["creative_results"],
                        }
                        exec_status_override = pause_result["exec_status"]
                    elif action in ("bid_up", "bid_down"):
                        final_bid = row.get("final_bid", row.get("recommended_bid"))
                        if final_bid is None or final_bid == "" or pd.isna(final_bid):
                            audit.log({
                                "ad_id": ad_id,
                                "action": action,
                                "tier": tier,
                                "execution_status": "skipped",
                                "reason": "无出价数据（NaN/空）",
                            })
                            continue
                        bid_fen = int(float(final_bid) * 100)
                        result = await executor.update_bid(ad_id, account_id, bid_fen)
                        exec_status_override = None
                    else:
                        continue

                    api_code = result.get("code", -1)
                    if exec_status_override is not None:
                        exec_status = exec_status_override
                    else:
                        exec_status = "success" if api_code == 0 else "failed"

                    if exec_status in ("success", "partial"):
                        approved_executed += 1
                        change_pct = row.get("recommended_change_pct", 0)
                        if isinstance(change_pct, str):
                            try:
                                change_pct = float(change_pct)
                            except ValueError:
                                change_pct = 0
                        history.record_adjustment(str(ad_id), action, change_pct)
                    else:
                        failed += 1

                    post_state = await executor.get_ad_state(ad_id, account_id) if exec_status in ("success", "partial") else None

                    audit_entry = {
                        "ad_id": ad_id,
                        "account_id": account_id,
                        "action": action,
                        "tier": tier,
                        "pre_state": {
                            "bid_amount": pre_state.get("bid_amount") if pre_state else None,
                            "status": pre_state.get("configured_status") if pre_state else None,
                        } if pre_state else None,
                        "post_state": {
                            "bid_amount": post_state.get("bid_amount") if post_state else None,
                            "status": post_state.get("configured_status") if post_state else None,
                        } if post_state else None,
                        "api_code": api_code,
                        "api_message": result.get("message", ""),
                        "execution_status": f"approved_{exec_status}",
                        "source": row.get("source", ""),
                    }
                    if pause_extra:
                        audit_entry.update(pause_extra)
                    audit.log(audit_entry)

                    logger.info(
                        "event=execute_decisions stage=approved_exec ad_id=%d account_id=%d "
                        "action=%s tier=%d result=%s api_code=%d",
                        ad_id, account_id, action, tier, exec_status, api_code,
                    )

        total_executed = executed + approved_executed

        output_lines = [
            f"执行完成，审计日志: {audit.path}",
            "",
            "执行结果:",
            f"  Tier 1 自动执行: {executed} 个成功 / {failed} 个失败",
        ]

        if not df_tier2_3.empty and (filter_ad_ids is not None or IM_ENABLED):
            output_lines.extend([
                f"  Tier 2/3 审批后执行: {approved_executed} 个成功",
                f"  审批拒绝: {rejected_count} 个",
                f"  审批超时: {timeout_count} 个",
            ])
        if pending_approval > 0:
            output_lines.append(f"  待审批（未执行）: {pending_approval} 个")

        output_lines.extend([
            "",
            "自治级别分布:",
            f"  Tier 1 (自动): {tier_summary.get(1, 0)} 个",
            f"  Tier 2 (审批): {tier_summary.get(2, 0)} 个",
            f"  Tier 3 (高价值): {tier_summary.get(3, 0)} 个",
        ])

        logger.info(
            "event=execute_decisions stage=run result=completed "
            "tier1_executed=%d tier1_failed=%d approved=%d rejected=%d pending=%d "
            "duration_ms=%d",
            executed, failed, approved_executed, rejected_count, pending_approval,
            int(_time.monotonic() * 1000) - _started_ms,
        )
        return ToolResult(
            title=f"执行完成（自动{executed}/审批通过{approved_executed}/拒绝{rejected_count}/超时{timeout_count}）",
            output="\n".join(output_lines),
            metadata={
                "audit_log": str(audit.path),
                "tier1_executed": executed,
                "tier1_failed": failed,
                "approved_executed": approved_executed,
                "rejected": rejected_count,
                "timeout": timeout_count,
                "pending_approval": pending_approval,
                "tier_summary": tier_summary,
            },
        )

    except Exception as e:
        logger.error(
            "event=execute_decisions stage=run result=failed duration_ms=%d error=%s",
            int(_time.monotonic() * 1000) - _started_ms if '_started_ms' in dir() else 0,
            str(e)[:200],
            exc_info=True,
        )
        return ToolResult(title="execute_decisions 失败", output=str(e))


# ═══════════════════════════════════════════
# 工具：执行后效果检查
# ═══════════════════════════════════════════


@tool(description="执行后效果检查：对比操作前后广告表现")
async def check_execution_feedback(
    ctx: ToolContext = None,
    execution_log_path: str = "",
    hours_after: int = FEEDBACK_CHECK_HOURS,
) -> ToolResult:
    """
    读取执行日志，通过 API 获取当前状态，对比操作前后。

    Args:
        execution_log_path: 执行日志路径（JSONL），默认最新
        hours_after: 操作后等待时间（小时），仅检查超过此时间的操作
    """
    try:
        import time as _time
        _started_ms = int(_time.monotonic() * 1000)
        logger.info(
            "event=check_execution_feedback stage=run result=started "
            "log_path=%s hours_after=%d",
            execution_log_path or "auto", hours_after,
        )

        # 查找最新执行日志
        if not execution_log_path:
            log_dir = EXECUTION_LOG_DIR
            if not log_dir.exists():
                return ToolResult(title="check_execution_feedback", output="无执行日志目录")
            candidates = sorted(log_dir.glob("exec_*.jsonl"), reverse=True)
            if not candidates:
                return ToolResult(title="check_execution_feedback", output="无执行日志")
            execution_log_path = str(candidates[0])

        # 读取日志
        entries = []
        with open(execution_log_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    entries.append(json.loads(line))

        if not entries:
            return ToolResult(title="check_execution_feedback", output="执行日志为空")

        # 过滤：只看成功执行的且超过等待时间的
        cutoff = datetime.now() - timedelta(hours=hours_after)
        check_entries = [
            e for e in entries
            if e.get("execution_status") == "success"
            and datetime.fromisoformat(e["ts"]) < cutoff
        ]

        if not check_entries:
            return ToolResult(
                title="check_execution_feedback",
                output=f"无需检查（没有超过{hours_after}小时前的成功操作）",
            )

        # 获取当前状态
        executor = TencentAdExecutor()
        results = []

        for entry in check_entries:
            ad_id = entry.get("ad_id")
            account_id = entry.get("account_id", 0)
            action = entry.get("action")
            pre_state = entry.get("pre_state", {})

            current_state = await executor.get_ad_state(ad_id, account_id)

            result = {
                "ad_id": ad_id,
                "action": action,
                "executed_at": entry.get("ts"),
                "pre_bid": pre_state.get("bid_amount") if pre_state else None,
                "pre_status": pre_state.get("status") if pre_state else None,
                "current_bid": current_state.get("bid_amount") if current_state else None,
                "current_status": current_state.get("configured_status") if current_state else None,
            }

            # 状态变化判断
            if action == "pause":
                result["effective"] = (
                    current_state.get("configured_status") == "AD_STATUS_SUSPEND"
                    if current_state else None
                )
            elif action in ("bid_up", "bid_down"):
                post_bid = entry.get("post_state", {})
                if post_bid:
                    expected_bid = post_bid.get("bid_amount")
                    actual_bid = current_state.get("bid_amount") if current_state else None
                    result["expected_bid"] = expected_bid
                    result["effective"] = (actual_bid == expected_bid) if actual_bid is not None else None

            results.append(result)

        # 统计
        effective = sum(1 for r in results if r.get("effective") is True)
        ineffective = sum(1 for r in results if r.get("effective") is False)
        unknown = sum(1 for r in results if r.get("effective") is None)

        output_lines = [
            f"效果检查完成（{len(results)} 个操作）",
            "",
            f"  有效: {effective} 个",
            f"  无效/被覆盖: {ineffective} 个",
            f"  未知: {unknown} 个",
        ]

        if ineffective > 0:
            output_lines.append("")
            output_lines.append("⚠️ 以下操作可能未生效或被覆盖:")
            for r in results:
                if r.get("effective") is False:
                    output_lines.append(
                        f"  - 广告{r['ad_id']}: {r['action']} "
                        f"(执行于 {r['executed_at']})"
                    )

        logger.info(
            "event=check_execution_feedback stage=run result=completed "
            "total=%d effective=%d ineffective=%d unknown=%d duration_ms=%d",
            len(results), effective, ineffective, unknown,
            int(_time.monotonic() * 1000) - _started_ms,
        )
        return ToolResult(
            title=f"效果检查（有效{effective}/无效{ineffective}）",
            output="\n".join(output_lines),
            metadata={
                "total": len(results),
                "effective": effective,
                "ineffective": ineffective,
                "unknown": unknown,
                "details": results,
            },
        )

    except Exception as e:
        logger.error(
            "event=check_execution_feedback stage=run result=failed error=%s",
            str(e)[:200], exc_info=True,
        )
        return ToolResult(title="check_execution_feedback 失败", output=str(e))
