Blame for src/web.py

"""
Web 界面：浏览器打开即可输入指令，报告在页面展示。
定时任务自动推送日报/周报到钉钉群。
"""
import logging
import os
import sys
from datetime import date, datetime, timedelta
from pathlib import Path

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

from flask import Flask, request, jsonify, render_template
from src.analyzer import run_analysis, run_monthly_analysis, run_daily_analysis, run_range_analysis, build_trend_table
from src.dingtalk import send_markdown
from src.formatter import (
    format_report, format_title,
    format_monthly_report, format_monthly_title,
    format_daily_report, format_daily_title,
    format_range_report, format_range_title,
)
from src.loader import load_config, load_weekly_data, get_available_weeks
from src.ai_analyzer import ai_parse_query, chat_with_chence, build_business_context
from src.memory import save_analysis, build_memory_context
from src.monitor import run_daily_monitor, format_monitor_dingtalk, compute_trend_data, compute_dashboard_data, get_filter_options
from src.budget import compute_budget_tracking

log = logging.getLogger("ad-agent")
app = Flask(__name__, template_folder=str(_root / "templates"))
CONFIG_PATH = _root / "config.yaml"
import math
import numpy as np

_API_TOKEN = os.environ.get("API_TOKEN", "").strip()


@app.before_request
def _check_auth():
    """When API_TOKEN is set, protect /api/* endpoints with Bearer token auth."""
    if not _API_TOKEN:
        return
    if not request.path.startswith("/api/"):
        return
    auth = request.headers.get("Authorization", "")
    if auth == f"Bearer {_API_TOKEN}":
        return
    return jsonify(ok=False, error="Unauthorized"), 401


_monitor_cache = {"data": None, "ts": None}
_last_quick_sync: datetime | None = None


def _clean_nan(obj):
    """递归清理：NaN/Inf→None，numpy 类型→Python 原生类型，确保 JSON 可序列化。"""
    if isinstance(obj, (np.integer,)):
        return int(obj)
    if isinstance(obj, (np.floating,)):
        v = float(obj)
        return None if (math.isnan(v) or math.isinf(v)) else v
    if isinstance(obj, np.ndarray):
        return _clean_nan(obj.tolist())
    if isinstance(obj, float) and (math.isnan(obj) or math.isinf(obj)):
        return None
    if isinstance(obj, dict):
        return {k: _clean_nan(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_clean_nan(v) for v in obj]
    return obj


@app.route("/")
def index():
    return render_template("index.html", api_token=_API_TOKEN)


@app.route("/api/datasource")
def datasource():
    config = load_config(CONFIG_PATH)
    gs = config.get("google_sheets", {})
    url = gs.get("spreadsheet_url", "") if gs.get("enabled") else ""
    try:
        from src.loader import load_daily_data
        df = load_daily_data(CONFIG_PATH)
        latest = df["date"].max()
        latest_str = latest.strftime("%Y-%m-%d") if latest is not None and str(latest) != "NaT" else None
    except Exception:
        latest_str = None
    return jsonify(ok=True, sheets_url=url, latest_date=latest_str)


@app.route("/api/memory")
def memory():
    from src.memory import load_recent
    records = load_recent(10)
    return jsonify(ok=True, records=records)


def _sync_platform(platform: str, fetch_fn, config_key: str):
    """通用同步逻辑：拉取 API 数据 → 写回 Google Sheets → 保存本地 CSV。"""
    from src.loader import write_api_data_to_sheets

    config = load_config(CONFIG_PATH)
    p_cfg = config.get(config_key, {})
    if not p_cfg.get("enabled"):
        return jsonify(ok=False, error=f"{platform} API 未启用（config.yaml → {config_key}.enabled）")

    days_back = p_cfg.get("days_back", 90)
    df = fetch_fn(days_back=days_back)

    if df.empty:
        return jsonify(ok=False, error="未拉取到数据")

    # 保存本地 CSV 备份
    save_path = _root / "data" / f"{config_key}_data.csv"
    save_path.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(save_path, index=False)

    # 写回 Google Sheets
    sheets_result = {"added": 0}
    sheets_error = None
    try:
        sheets_result = write_api_data_to_sheets(df, config_path=CONFIG_PATH)
    except Exception as e:
        sheets_error = str(e)
        log.warning(f"{platform} 数据写回 Sheets 失败（数据已存本地）: {e}")

    date_min = df["date"].min().strftime("%Y-%m-%d")
    date_max = df["date"].max().strftime("%Y-%m-%d")
    log.info(f"{platform} 数据已同步: {len(df)} 条, {date_min}~{date_max}, 写入 Sheets {sheets_result['added']} 行")

    _monitor_cache["data"] = None
    _monitor_cache["ts"] = None

    resp = dict(
        ok=True, rows=len(df),
        date_range=f"{date_min} ~ {date_max}",
        sheets_added=sheets_result["added"],
    )
    if sheets_error:
        resp["sheets_warning"] = sheets_error
    return jsonify(resp)


@app.route("/api/rebuild_sheets", methods=["POST"])
def rebuild_sheets():
    """一次性操作：清空分平台表旧数据 → 全量重新同步（纯数字格式）。"""
    import importlib
    from src.loader import load_config, _get_spreadsheet, write_api_data_to_sheets

    try:
        config = load_config(CONFIG_PATH)
        sh = _get_spreadsheet(config)
        gs_cfg = config.get("google_sheets", {})
        platform_sheets = gs_cfg.get("platform_sheets", ["Meta", "Google", "Criteo", "Bing"])

        cleared = []
        for name in platform_sheets:
            try:
                ws = sh.worksheet(name)
                rows = ws.row_count
                if rows > 1:
                    ws.delete_rows(2, rows)
                    cleared.append(name)
                    log.info(f"[rebuild] 已清空 Sheet [{name}] ({rows - 1} 行)")
            except Exception as e:
                log.warning(f"[rebuild] 清空 {name} 失败: {e}")

        platform_defs = [
            ("Meta", "meta_api", "src.meta_api", "fetch_meta_data"),
            ("Google", "google_ads_api", "src.google_api", "fetch_google_data"),
            ("Criteo", "criteo_api", "src.criteo_api", "fetch_criteo_data"),
            ("Bing", "bing_api", "src.bing_api", "fetch_bing_data"),
        ]
        synced = {}
        for name, cfg_key, mod_path, fn_name in platform_defs:
            p_cfg = config.get(cfg_key, {})
            if not p_cfg.get("enabled"):
                continue
            try:
                mod = importlib.import_module(mod_path)
                fetch_fn = getattr(mod, fn_name)
                days = p_cfg.get("days_back", 90)
                df = fetch_fn(days_back=days)
                if df.empty:
                    synced[name] = {"rows": 0, "status": "no_data"}
                    continue
                res = write_api_data_to_sheets(df, config_path=CONFIG_PATH)
                synced[name] = {"rows": len(df), "sheets_added": res.get("added", 0), "status": "ok"}
                log.info(f"[rebuild] {name}: {len(df)} 条, Sheets +{res.get('added', 0)}")
            except Exception as e:
                synced[name] = {"status": "error", "error": str(e)}
                log.warning(f"[rebuild] {name} 同步失败: {e}")

        _monitor_cache["data"] = None
        _monitor_cache["ts"] = None

        return jsonify(ok=True, cleared=cleared, synced=synced)
    except Exception as e:
        log.exception(f"rebuild_sheets 出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/sync_meta", methods=["POST"])
def sync_meta():
    """从 Meta API 拉取数据 → 写回 Google Sheets → 保存本地 CSV。"""
    try:
        from src.meta_api import fetch_meta_data
        return _sync_platform("Meta", fetch_meta_data, "meta_api")
    except Exception as e:
        log.exception(f"Meta 同步出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/sync_google", methods=["POST"])
def sync_google():
    """从 Google Ads API 拉取数据 → 写回 Google Sheets → 保存本地 CSV。"""
    try:
        from src.google_api import fetch_google_data
        return _sync_platform("Google", fetch_google_data, "google_ads_api")
    except Exception as e:
        log.exception(f"Google Ads 同步出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/sync_criteo", methods=["POST"])
def sync_criteo():
    """从 Criteo API 拉取数据 → 写回 Google Sheets → 保存本地 CSV。"""
    try:
        from src.criteo_api import fetch_criteo_data
        return _sync_platform("Criteo", fetch_criteo_data, "criteo_api")
    except Exception as e:
        log.exception(f"Criteo 同步出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/sync_bing", methods=["POST"])
def sync_bing():
    """从 Bing Ads API 拉取数据 → 写回 Google Sheets → 保存本地 CSV。"""
    try:
        from src.bing_api import fetch_bing_data
        return _sync_platform("Bing", fetch_bing_data, "bing_api")
    except Exception as e:
        log.exception(f"Bing Ads 同步出错: {e}")
        return jsonify(ok=False, error=str(e))


def _quick_sync_stale_platforms():
    """每天至多执行一次：拉取最近 7 天数据并写回 Sheets。"""
    global _last_quick_sync
    import importlib
    from src.loader import write_api_data_to_sheets

    now = datetime.now()
    if _last_quick_sync and _last_quick_sync.date() == now.date():
        return []

    config = load_config(CONFIG_PATH)

    platform_defs = [
        ("Meta", "meta_api", "src.meta_api", "fetch_meta_data"),
        ("Google", "google_ads_api", "src.google_api", "fetch_google_data"),
        ("Criteo", "criteo_api", "src.criteo_api", "fetch_criteo_data"),
        ("Bing", "bing_api", "src.bing_api", "fetch_bing_data"),
    ]
    synced = []

    for name, cfg_key, mod_path, fn_name in platform_defs:
        p_cfg = config.get(cfg_key, {})
        if not p_cfg.get("enabled"):
            continue

        try:
            mod = importlib.import_module(mod_path)
            fetch_fn = getattr(mod, fn_name)
            df_new = fetch_fn(days_back=7)
            if df_new.empty:
                log.info(f"[快速同步] {name}: 无新数据")
                continue

            try:
                res = write_api_data_to_sheets(df_new, config_path=CONFIG_PATH)
                log.info(f"[快速同步] {name}: {len(df_new)} 条, Sheets +{res.get('added', 0)} 行")
            except Exception as e:
                log.warning(f"[快速同步] {name}: 写 Sheets 失败: {e}")

            synced.append(name)
        except Exception as e:
            log.warning(f"[快速同步] {name}: 同步失败 - {e}")

    _last_quick_sync = now
    if synced:
        _monitor_cache["data"] = None
        _monitor_cache["ts"] = None

    return synced


@app.route("/api/monitor")
def monitor():
    try:
        date_str = request.args.get("date")
        target_date = date.fromisoformat(date_str) if date_str else None
        country_param = request.args.get("country")

        has_filter = bool(date_str or (country_param and country_param.lower() != "all"))

        if not has_filter:
            now = datetime.now()
            cached = _monitor_cache["data"]
            cached_ts = _monitor_cache["ts"]
            if cached and cached_ts and cached_ts.date() == now.date():
                age_min = int((now - cached_ts).total_seconds() // 60)
                return jsonify(ok=True, cached=True, cache_age_min=age_min, **cached)

        result = run_daily_monitor(CONFIG_PATH, target_date=target_date, country=country_param)
        result = _clean_nan(result)

        if not has_filter:
            _monitor_cache["data"] = result
            _monitor_cache["ts"] = datetime.now()

        return jsonify(ok=True, cached=False, **result)
    except Exception as e:
        log.exception(f"监控出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/monitor/trend")
def monitor_trend():
    try:
        start_str = request.args.get("start")
        end_str = request.args.get("end")
        if start_str:
            start = date.fromisoformat(start_str)
        else:
            start = date.today() - timedelta(days=14)
        if end_str:
            end = date.fromisoformat(end_str)
        else:
            end = date.today() - timedelta(days=1)
        country = request.args.get("country")
        result = compute_trend_data(CONFIG_PATH, start, end, country=country)
        result = _clean_nan(result)
        return jsonify(ok=True, **result)
    except Exception as e:
        log.exception(f"趋势数据出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/dashboard")
def api_dashboard():
    """数据看板：区间 + 多维筛选。"""
    try:
        start_str = request.args.get("start")
        end_str = request.args.get("end")
        if start_str:
            start = date.fromisoformat(start_str)
        else:
            start = date.today() - timedelta(days=14)
        if end_str:
            end = date.fromisoformat(end_str)
        else:
            end = date.today() - timedelta(days=1)

        filters = {}
        for dim in ("country", "platform", "ad_objective", "product_line", "product_model", "project_type"):
            val = request.args.get(dim)
            if val:
                filters[dim] = val

        result = compute_dashboard_data(CONFIG_PATH, start, end, filters=filters or None)
        result = _clean_nan(result)
        return jsonify(ok=True, **result)
    except Exception as e:
        log.exception(f"Dashboard 出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/dashboard/filters")
def api_dashboard_filters():
    """返回各筛选维度的可选值。"""
    try:
        options = get_filter_options(CONFIG_PATH)
        return jsonify(ok=True, **options)
    except Exception as e:
        log.exception(f"Dashboard filters 出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/budget")
def api_budget():
    try:
        month = request.args.get("month")
        result = compute_budget_tracking(CONFIG_PATH, target_month=month)
        result = _clean_nan(result)
        if "error" in result:
            config = load_config(CONFIG_PATH)
            gs = config.get("google_sheets", {})
            if gs.get("enabled") and gs.get("spreadsheet_url"):
                result["sheets_url"] = gs["spreadsheet_url"]
            return jsonify(ok=False, **result)
        return jsonify(ok=True, **result)
    except Exception as e:
        log.exception(f"预算追踪出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/analyze", methods=["POST"])
def analyze():
    data = request.get_json(force=True)
    query = (data.get("query") or "").strip()
    if not query:
        return jsonify(ok=False, error="请输入指令")

    biz_note = (data.get("biz_note") or "").strip()
    platform_filter = (data.get("platform") or "").strip() or None

    params, parse_err = ai_parse_query(query)
    if params is None:
        return jsonify(ok=False, error=parse_err or "无法理解你的查询")

    try:
        mode = params["mode"]
        mem_ctx = build_memory_context(n=5)
        config = load_config(CONFIG_PATH)

        if mode == "range":
            s = datetime.strptime(params["start"], "%Y-%m-%d").date()
            e = datetime.strptime(params["end"], "%Y-%m-%d").date()
            trend = build_trend_table(config_path=CONFIG_PATH, mode="day", current_date=e, n=8)
            biz_ctx = build_business_context(config, f"{s} ~ {e}", biz_note)
            result = run_range_analysis(config_path=CONFIG_PATH, start_date=s, end_date=e,
                                        trend_table=trend, memory_context=mem_ctx,
                                        business_context=biz_ctx, platform=platform_filter)
            if result.get("error"):
                return jsonify(ok=False, error=result["error"])
            title = format_range_title(result)
            period_label = result.get("period_label", title)
            cp = result.get("prev_label", "")
            yp = result.get("yoy_label", "")
            days = result.get("days_count", "")
            cmp_parts = []
            if days:
                cmp_parts.append(f"共 {days} 天")
            if cp:
                cmp_parts.append(f"环比 vs {cp}")
            if yp:
                cmp_parts.append(f"同比 vs {yp}")
            compare_label = " | ".join(cmp_parts)
        elif mode == "day":
            target_date = datetime.strptime(params["date"], "%Y-%m-%d").date()
            trend = build_trend_table(config_path=CONFIG_PATH, mode="day", current_date=target_date, n=8)
            biz_ctx = build_business_context(config, params["date"], biz_note)
            result = run_daily_analysis(config_path=CONFIG_PATH, target_date=target_date,
                                        trend_table=trend, memory_context=mem_ctx,
                                        business_context=biz_ctx, platform=platform_filter)
            if result.get("error"):
                return jsonify(ok=False, error=result["error"])
            title = format_daily_title(result)
            period_label = result.get("period_label", title)
            cp = result.get("prev_label", "")
            yp = result.get("yoy_label", "")
            compare_label = " | ".join(filter(None, [f"环比 vs {cp}" if cp else "", f"同比 vs {yp}" if yp else ""]))
        elif mode == "month":
            trend = build_trend_table(config_path=CONFIG_PATH, mode="month",
                                      current_year=params["year"], current_period=params["month"], n=6)
            biz_ctx = build_business_context(config, f"{params['year']}年{params['month']}月", biz_note)
            result = run_monthly_analysis(
                config_path=CONFIG_PATH,
                target_year=params["year"],
                target_month=params["month"],
                trend_table=trend, memory_context=mem_ctx,
                business_context=biz_ctx,
                platform=platform_filter,
            )
            if result.get("error"):
                return jsonify(ok=False, error=result["error"])
            title = format_monthly_title(result)
            period_label = result.get("period_label", title)
            cp = result.get("prev_label", "")
            yp = result.get("yoy_label", "")
            compare_label = " | ".join(filter(None, [f"环比 vs {cp}" if cp else "", f"同比 vs {yp}" if yp else ""]))
        else:
            year, week = None, None
            if mode == "week":
                year, week = params["year"], params["week"]
            elif mode == "prev":
                df = load_weekly_data(CONFIG_PATH)
                weeks = get_available_weeks(df)
                if len(weeks) >= 2:
                    year, week = weeks[-2]
            trend = build_trend_table(config_path=CONFIG_PATH, mode="week",
                                      current_year=year, current_period=week, n=8)
            wk_label = f"{year} W{week}" if year and week else ""
            biz_ctx = build_business_context(config, wk_label, biz_note)
            result = run_analysis(config_path=CONFIG_PATH, target_year=year, target_week=week,
                                  trend_table=trend, memory_context=mem_ctx,
                                  business_context=biz_ctx, platform=platform_filter)
            title = format_title(result)
            cp_ = result.get("current_period", {})
            period_label = f"{cp_.get('year', '')} W{cp_.get('week', '')}"
            pp = result.get("previous_period")
            yp = result.get("yoy_period")
            parts = []
            if pp:
                parts.append(f"环比 vs {pp['year']} W{pp['week']}")
            if yp:
                parts.append(f"同比 vs {yp['year']} W{yp['week']}")
            compare_label = " | ".join(parts)

        expert = result.get("expert")
        metrics_snap = (result.get("current") or {}).get("total")
        save_analysis(period_label, expert, metrics_snap)

        report_context = _build_chat_context(title, result, expert)

        report_data = _clean_nan({
            "title": title,
            "compare_label": compare_label,
            "current": result.get("current"),
            "wow_total": result.get("wow_total"),
            "yoy_total": result.get("yoy_total"),
            "wow_by_platform": result.get("wow_by_platform"),
            "yoy_by_platform": result.get("yoy_by_platform"),
            "wow_by_campaign": result.get("wow_by_campaign"),
            "yoy_by_campaign": result.get("yoy_by_campaign"),
            "wow_by_country": result.get("wow_by_country"),
            "yoy_by_country": result.get("yoy_by_country"),
            "expert": expert,
        })

        return jsonify(ok=True, title=title, report=report_data, report_context=report_context)

    except Exception as e:
        log.exception(f"分析出错: {e}")
        return jsonify(ok=False, error=str(e))


def _build_chat_context(title: str, result: dict, expert: dict | None) -> str:
    """构建报告摘要字符串，供多轮对话时作为上下文。"""
    parts = [f"报告: {title}"]
    total = (result.get("current") or {}).get("total", {})
    if total:
        nums = []
        for k in ["Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR", "GA4.Sessions"]:
            v = total.get(k)
            if v is not None and v == v:
                if k in ("CTR", "CVR"):
                    nums.append(f"{k}: {v*100:.2f}%")
                elif abs(v) >= 1000:
                    nums.append(f"{k}: {v:,.0f}")
                else:
                    nums.append(f"{k}: {v:.2f}")
        parts.append("整体: " + ", ".join(nums))
    wow = result.get("wow_total") or {}
    if wow:
        chgs = [f"{k}: {v:+.1f}%" for k, v in wow.items() if v is not None and v == v][:6]
        if chgs:
            parts.append("环比: " + ", ".join(chgs))
    for dim_key, dim_label in [("by_platform", "平台"), ("by_campaign_type", "漏斗"), ("by_country", "市场")]:
        dim = (result.get("current") or {}).get(dim_key, {})
        for name, metrics in dim.items():
            roas = metrics.get("ROAS")
            cost = metrics.get("Cost")
            if roas is not None:
                parts.append(f"{dim_label}-{name}: ROAS {roas:.2f}, Cost ${cost:,.0f}" if cost else f"{dim_label}-{name}: ROAS {roas:.2f}")
    if expert:
        if expert.get("verdict"):
            parts.append(f"判定: {expert['verdict']}")
        for f in (expert.get("key_findings") or []):
            parts.append(f"发现: {f}")
    return "\n".join(parts)


# ─────────────────────────────────────────────
# P3 投放分析 API
# ─────────────────────────────────────────────

@app.route("/api/p3/analysis")
def p3_analysis():
    """P3 四层投放分析。"""
    try:
        from src.p3_analyzer import run_p3_analysis
        from src.p3_benchmark import load_benchmarks, load_platform_benchmarks, apply_benchmarks, run_diagnostics
        from src.p3_report import format_p3_report

        period_type = request.args.get("period_type", "week")
        year = request.args.get("year", type=int)
        period = request.args.get("period", type=int)
        start_str = request.args.get("start")
        end_str = request.args.get("end")
        market_filter = request.args.get("market")

        start_date = date.fromisoformat(start_str) if start_str else None
        end_date = date.fromisoformat(end_str) if end_str else None

        result = run_p3_analysis(
            config_path=CONFIG_PATH,
            period_type=period_type,
            target_year=year,
            target_period=period,
            start_date=start_date,
            end_date=end_date,
            market=market_filter,
        )

        if result.get("error"):
            return jsonify(ok=False, error=result["error"])

        p3_cfg = result.pop("p3_config", {})
        benchmarks = load_benchmarks(p3_cfg)
        plat_benchmarks = load_platform_benchmarks(p3_cfg)
        result["L1"] = apply_benchmarks(result["L1"], benchmarks, "L1")
        result["L2"] = apply_benchmarks(result["L2"], benchmarks, "L2")
        result["L3"] = apply_benchmarks(result["L3"], benchmarks, "L3", platform_benchmarks=plat_benchmarks)
        result["L4"] = apply_benchmarks(result["L4"], benchmarks, "L4")
        diagnostics = run_diagnostics(result, p3_cfg)
        result["diagnostics"] = diagnostics

        report_md = format_p3_report(result, diagnostics)
        result["report_markdown"] = report_md

        return jsonify(ok=True, **_clean_nan(result))
    except Exception as e:
        log.exception(f"P3 分析出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/p3/config")
def p3_config():
    """当前 P3 配置数据（从 Google Sheets 动态读取）。"""
    try:
        from src.p3_config import load_all_p3_config
        cfg = load_all_p3_config(CONFIG_PATH)
        return jsonify(ok=True, **_clean_nan(cfg))
    except Exception as e:
        log.exception(f"P3 配置读取出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/p3/push", methods=["POST"])
def p3_push():
    """手动推送 P3 周报到钉钉。"""
    try:
        from src.p3_analyzer import run_p3_analysis
        from src.p3_benchmark import load_benchmarks, load_platform_benchmarks, apply_benchmarks, run_diagnostics
        from src.p3_report import format_p3_report, format_p3_title

        result = run_p3_analysis(config_path=CONFIG_PATH, period_type="week")
        if result.get("error"):
            return jsonify(ok=False, error=result["error"])

        p3_cfg = result.pop("p3_config", {})
        benchmarks = load_benchmarks(p3_cfg)
        plat_benchmarks = load_platform_benchmarks(p3_cfg)
        result["L1"] = apply_benchmarks(result["L1"], benchmarks, "L1")
        result["L2"] = apply_benchmarks(result["L2"], benchmarks, "L2")
        result["L3"] = apply_benchmarks(result["L3"], benchmarks, "L3", platform_benchmarks=plat_benchmarks)
        result["L4"] = apply_benchmarks(result["L4"], benchmarks, "L4")
        diagnostics = run_diagnostics(result, p3_cfg)

        title = format_p3_title(result)
        body = format_p3_report(result, diagnostics)
        ok_push, err_push = send_markdown(title=title, text=body)
        if ok_push:
            return jsonify(ok=True, message=f"P3 周报已推送: {title}")
        return jsonify(ok=False, error=f"推送失败: {err_push}")
    except Exception as e:
        log.exception(f"P3 推送出错: {e}")
        return jsonify(ok=False, error=str(e))


@app.route("/api/chat", methods=["POST"])
def chat():
    data = request.get_json(force=True)
    messages = data.get("messages", [])
    report_context = data.get("report_context", "")
    biz_note = (data.get("biz_note") or "").strip()
    if not messages:
        return jsonify(ok=False, error="无对话内容")
    try:
        biz_ctx = None
        if biz_note:
            biz_ctx = f"【用户补充的业务备注】\n{biz_note}"
        reply = chat_with_chence(messages, report_context, business_context=biz_ctx)
        return jsonify(ok=True, reply=reply)
    except Exception as e:
        log.exception(f"对话出错: {e}")
        return jsonify(ok=False, error=str(e))


# ─────────────────────────────────────────────
# 定时推送：自动同步 + 日报 + 周报 → 钉钉
# ─────────────────────────────────────────────

def _auto_sync():
    """定时自动同步所有 API 数据源。"""
    from src.loader import write_api_data_to_sheets
    config = load_config(CONFIG_PATH)
    platforms = [
        ("Meta", "meta_api", "src.meta_api", "fetch_meta_data"),
        ("Google", "google_ads_api", "src.google_api", "fetch_google_data"),
        ("Criteo", "criteo_api", "src.criteo_api", "fetch_criteo_data"),
        ("Bing", "bing_api", "src.bing_api", "fetch_bing_data"),
    ]
    for name, cfg_key, mod_path, fn_name in platforms:
        p_cfg = config.get(cfg_key, {})
        if not p_cfg.get("enabled"):
            continue
        try:
            import importlib
            mod = importlib.import_module(mod_path)
            fetch_fn = getattr(mod, fn_name)
            df = fetch_fn(days_back=p_cfg.get("days_back", 90))
            if df.empty:
                log.info(f"[自动同步] {name}: 无数据")
                continue
            save_path = _root / "data" / f"{cfg_key}_data.csv"
            save_path.parent.mkdir(parents=True, exist_ok=True)
            df.to_csv(save_path, index=False)
            try:
                res = write_api_data_to_sheets(df, config_path=CONFIG_PATH)
                log.info(f"[自动同步] {name}: {len(df)} 条, Sheets +{res.get('added', 0)} 行")
            except Exception as e:
                log.warning(f"[自动同步] {name}: 写 Sheets 失败: {e}")
        except Exception as e:
            log.warning(f"[自动同步] {name}: 同步失败: {e}")

    _monitor_cache["data"] = None
    _monitor_cache["ts"] = None
    log.info("[自动同步] 完成")


def _push_daily_report():
    """每天 9:30 执行：推送监控告警 + 昨日日报到钉钉。"""
    try:
        import pytz
        yesterday = (datetime.now(pytz.timezone("Asia/Shanghai")) - timedelta(days=1)).date()
        log.info(f"[定时] 生成日报: {yesterday}")

        # 1) 推送监控告警
        try:
            monitor_result = run_daily_monitor(CONFIG_PATH, target_date=yesterday)
            mon_title, mon_body = format_monitor_dingtalk(monitor_result)
            ok_m, err_m = send_markdown(title=mon_title, text=mon_body)
            if ok_m:
                log.info(f"[定时] 监控已推送: {mon_title}")
            else:
                log.warning(f"[定时] 监控推送失败: {err_m}")
        except Exception as e:
            log.warning(f"[定时] 监控推送异常: {e}")

        # 2) 推送日报
        result = run_daily_analysis(config_path=CONFIG_PATH, target_date=yesterday)
        if result.get("error"):
            log.warning(f"[定时] 日报跳过: {result['error']}")
            return
        title = format_daily_title(result)
        body = format_daily_report(result)
        ok, err = send_markdown(title=title, text=body)
        if ok:
            log.info(f"[定时] 日报已推送: {title}")
        else:
            log.warning(f"[定时] 日报推送失败: {err}")
    except Exception as e:
        log.exception(f"[定时] 日报异常: {e}")


def _push_weekly_report():
    """每周一 9:30 执行：推送上周周报到钉钉。"""
    try:
        log.info("[定时] 生成周报")
        df = load_weekly_data(CONFIG_PATH)
        weeks = get_available_weeks(df)
        if len(weeks) < 2:
            log.warning("[定时] 周报跳过: 数据不足两周")
            return
        year, week = weeks[-2]
        result = run_analysis(config_path=CONFIG_PATH, target_year=year, target_week=week)
        title = format_title(result)
        body = format_report(result)
        ok, err = send_markdown(title=title, text=body)
        if ok:
            log.info(f"[定时] 周报已推送: {title}")
        else:
            log.warning(f"[定时] 周报推送失败: {err}")
    except Exception as e:
        log.exception(f"[定时] 周报异常: {e}")


def _init_scheduler():
    """启动后台定时任务。"""
    try:
        from apscheduler.schedulers.background import BackgroundScheduler
        from apscheduler.triggers.cron import CronTrigger
        import pytz
    except ImportError:
        log.warning("apscheduler 或 pytz 未安装，定时任务未启动")
        return

    tz = pytz.timezone("Asia/Shanghai")
    config = load_config(CONFIG_PATH)
    schedule_cfg = config.get("schedule", {})

    if schedule_cfg.get("disabled"):
        log.info("[定时] 定时任务已禁用 (schedule.disabled: true)")
        return

    sync_cron = schedule_cfg.get("sync_cron", "0 9 * * *").strip().split()
    daily_cron = schedule_cfg.get("daily_cron", "30 9 * * *").strip().split()
    weekly_cron = schedule_cfg.get("weekly_cron", "30 9 * * 1").strip().split()

    scheduler = BackgroundScheduler()

    if len(sync_cron) == 5:
        mi, hr, d, mo, dow = sync_cron
        scheduler.add_job(
            _auto_sync,
            CronTrigger(minute=mi, hour=hr, day=d, month=mo, day_of_week=dow, timezone=tz),
            id="auto_sync",
        )
        log.info(f"[定时] 自动同步已注册: {' '.join(sync_cron)} (Asia/Shanghai)")

    if len(daily_cron) == 5:
        mi, hr, d, mo, dow = daily_cron
        scheduler.add_job(
            _push_daily_report,
            CronTrigger(minute=mi, hour=hr, day=d, month=mo, day_of_week=dow, timezone=tz),
            id="daily_report",
        )
        log.info(f"[定时] 日报任务已注册: {' '.join(daily_cron)} (Asia/Shanghai)")

    if len(weekly_cron) == 5:
        mi, hr, d, mo, dow = weekly_cron
        scheduler.add_job(
            _push_weekly_report,
            CronTrigger(minute=mi, hour=hr, day=d, month=mo, day_of_week=dow, timezone=tz),
            id="weekly_report",
        )
        log.info(f"[定时] 周报任务已注册: {' '.join(weekly_cron)} (Asia/Shanghai)")

    scheduler.start()


def start_web(host: str = "0.0.0.0", port: int = None):
    port = port or int(os.environ.get("PORT", 5000))
    _init_scheduler()
    if os.environ.get("CLOUD") != "1":
        import webbrowser
        webbrowser.open(f"http://127.0.0.1:{port}")
    app.run(host=host, port=port, debug=False)


if __name__ == "__main__":
    start_web()