""" 入口： --web 网页界面（推荐）：浏览器打开，点击即用 --chat 终端交互模式（默认仍会推钉钉，加 --no-ding 仅本机预览） --run-once 单次分析最新一周并推送（加 --no-ding 仅打印，不调钉钉） --dry-run / --preview 只生成报告输出到终端（不推送） (无参数) 常驻定时任务 """ import argparse import logging import sys from pathlib import Path _root = Path(__file__).resolve().parent.parent if str(_root) not in sys.path: sys.path.insert(0, str(_root)) from src.analyzer import run_analysis, run_monthly_analysis from src.dingtalk import send_markdown from src.formatter import ( format_report, format_title, format_monthly_report, format_monthly_title, ) from src.loader import load_config, load_weekly_data, get_available_weeks from src.parser import parse_query, describe_query, HELP_TEXT logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler(_root / "logs" / "agent.log", encoding="utf-8"), ], ) log = logging.getLogger("ad-agent") def _configure_stdout_utf8() -> None: """避免 Windows 控制台默认 GBK 打印周报里的 emoji 时崩溃。""" try: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") except Exception: pass def _do_analysis(params: dict, config_path: Path) -> tuple[str, str]: """根据解析参数执行分析，返回 (title, markdown_body)。""" mode = params["mode"] if mode == "month": result = run_monthly_analysis( config_path=config_path, target_year=params["year"], target_month=params["month"], ) if result.get("error"): return "数据不足", result["error"] return format_monthly_title(result), format_monthly_report(result) 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] result = run_analysis(config_path=config_path, target_year=year, target_week=week) return format_title(result), format_report(result) def run_once( config_path: Path | None = None, target_year: int | None = None, target_week: int | None = None, *, send_ding: bool = True, ) -> bool: config_path = config_path or _root / "config.yaml" log.info("开始周度分析...") try: result = run_analysis( config_path=config_path, target_year=target_year, target_week=target_week, ) except Exception as e: log.error(f"分析失败: {e}", exc_info=True) return False title = format_title(result) text = format_report(result) if not send_ding: print("\n" + "=" * 64) print(f"【终端预览 · 未推送钉钉】{title}") print("=" * 64 + "\n") print(text) print("\n" + "=" * 64) log.info("已输出到终端（预览模式，未调用钉钉）。") return True ok, err = send_markdown(title=title, text=text) if not ok: log.error(f"钉钉发送失败: {err}") print(text) else: log.info(f"已发送: {title}") return ok def chat_loop(config_path: Path, *, send_ding: bool = True) -> None: """本地交互模式：读取用户输入 → 分析 → 可选推送到钉钉群。""" print("=" * 50) print(" DM 广告数据分析 Agent（交互模式）") print("=" * 50) print() if send_ding: print("输入分析指令，报告将通过钉钉 Webhook 推送到群。") else: print("【调试模式】报告只打印在本终端，不会推送到钉钉。") print("支持的指令：") print(" 分析最新一周 / 分析上周") print(" 分析25年12月 / 分析2025年12月") print(" 分析W9 / 分析第9周 / 分析2025 W40") print(" exit / quit 退出") print() while True: try: user_input = input(">>> ").strip() except (EOFError, KeyboardInterrupt): print("\n再见！") break if not user_input: continue if user_input.lower() in ("exit", "quit", "q", "退出"): print("再见！") break if user_input in ("帮助", "help", "?", "？"): print(HELP_TEXT) continue params = parse_query(user_input) if params is None: print(f"无法识别「{user_input}」，请输入「帮助」查看支持的指令。") continue desc = describe_query(params) print(f"正在分析: {desc} ...") try: title, body = _do_analysis(params, config_path) except Exception as e: print(f"分析出错: {e}") continue print(f"报告已生成: {title}（{len(body)} 字符）") if not send_ding: print("\n--- 报告正文（未推送钉钉）---\n") print(body) print("\n--- 结束 ---\n") continue ok, err = send_markdown(title=title, text=body) if ok: print(f"已推送到钉钉群。") else: print(f"钉钉推送失败: {err}") print("--- 报告内容 ---") print(body) print("---") print() def main() -> None: _configure_stdout_utf8() parser = argparse.ArgumentParser(description="广告数据周度分析 Agent") parser.add_argument("--web", action="store_true", help="启动网页界面（推荐）") parser.add_argument("--chat", action="store_true", help="终端交互模式") parser.add_argument("--run-once", action="store_true", help="分析最新一周并推送钉钉（加 --no-ding 则只打印到终端）") parser.add_argument("--no-ding", action="store_true", help="不调用钉钉，仅终端输出（配合 --run-once 或 --chat）") parser.add_argument("--config", type=Path, default=None, help="config.yaml 路径") parser.add_argument("--week", type=str, default=None, help="指定分析周 (YYYY-W)") parser.add_argument("--month", type=str, default=None, help="指定分析月 (YYYY-M)") parser.add_argument("--dry-run", action="store_true", help="只输出报告到终端，不推送钉钉") parser.add_argument("--preview", action="store_true", help="同 --dry-run，便于记") args = parser.parse_args() config_path = args.config or _root / "config.yaml" # ── 网页界面 ── if args.web: from src.web import start_web start_web() return # ── 终端交互模式 ── if args.chat: chat_loop(config_path, send_ding=not args.no_ding) return # ── 解析时间参数 ── target_year, target_week, target_month = None, None, None if args.week: parts = args.week.split("-") target_year, target_week = int(parts[0]), int(parts[1]) if args.month: parts = args.month.split("-") target_year, target_month = int(parts[0]), int(parts[1]) # ── Dry-run / Preview ── if args.dry_run or args.preview: if target_month is not None: result = run_monthly_analysis( config_path=config_path, target_year=target_year, target_month=target_month, ) title = format_monthly_title(result) text = format_monthly_report(result) else: result = run_analysis( config_path=config_path, target_year=target_year, target_week=target_week, ) title = format_title(result) text = format_report(result) print(f"=== {title} ===") print(text) return # ── Run-once ── if args.run_once: run_once( config_path=config_path, target_year=target_year, target_week=target_week, send_ding=not args.no_ding, ) return # ── 定时任务 ── from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger config = load_config(config_path) cron_expr = (config.get("schedule", {}).get("weekly_cron") or "0 9 * * 1").strip() parts = cron_expr.split() if len(parts) != 5: minute, hour, day, month, dow = "0", "9", "*", "*", "1" else: minute, hour, day, month, dow = parts scheduler = BlockingScheduler() scheduler.add_job( run_once, CronTrigger(minute=minute, hour=hour, day=day, month=month, day_of_week=dow), args=[config_path], id="weekly_report", ) log.info(f"定时任务已启动，cron: {cron_expr}。按 Ctrl+C 退出。") scheduler.start() if __name__ == "__main__": main()