Spaces:
Runtime error
Runtime error
| """ | |
| 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() | |
| 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 | |
| def index(): | |
| return render_template("index.html", api_token=_API_TOKEN) | |
| 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) | |
| 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) | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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 | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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 | |
| # ───────────────────────────────────────────── | |
| 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)) | |
| 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)) | |
| 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)) | |
| 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() | |