Spaces:
Runtime error
Runtime error
| """ | |
| 入口： | |
| --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() | |