""" 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()