Blame for src/main.py

"""
入口：
  --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()