hourly_bot.py

"""按平台节奏执行推送。

用法:
    # 直接运行（按配置决定当前小时需要发哪些平台）
    python hourly_bot.py

    # 只抓 X
    python hourly_bot.py --only x

    # 忽略节奏限制，当前小时强制检查全部平台
    python hourly_bot.py --force

Windows 定时任务设置:
    schtasks /create /tn "AI-Daily-Hourly" /tr "python D:\\code\\self-project\\feishu-feed-test\\hourly_bot.py" /sc hourly
"""
import argparse
import json
import sys
import time
from datetime import datetime, timedelta, timezone

from config import load_config
from delivery import (
    due_platform_keys,
    filter_recent_articles,
    get_delivery_config,
    load_delivery_state,
    mark_sent_articles,
    save_delivery_state,
)
from fetcher import fetch_hn, fetch_reddit, fetch_x
from logger import get_logger, init_logging
from sender import build_ai_daily_card, send_card
from summarizer import summarize

log = get_logger("hourly_bot")

PLATFORM_FETCHERS = {
    "x": fetch_x,
    "reddit": fetch_reddit,
    "hn": fetch_hn,
}


def run_platform(
    platform_key: str,
    now_utc: datetime,
    state: dict[str, dict[str, str]],
    dry_run: bool,
    webhook_url: str,
) -> None:
    dconfig = get_delivery_config(platform_key)
    fetch_fn = PLATFORM_FETCHERS[platform_key]
    cn_tz = timezone(timedelta(hours=8))
    plog = log.bind(platform=platform_key)

    plog.info(
        "Fetcher start: platform={}, fetch_hours={}, cadence_hours={}",
        dconfig.platform, dconfig.fetch_hours, dconfig.cadence_hours,
    )
    t0 = time.time()
    articles = fetch_fn(hours=dconfig.fetch_hours)
    plog.info("Fetched {} articles in {:.1f}s", len(articles), time.time() - t0)

    filtered_articles = filter_recent_articles(
        state=state,
        platform_key=platform_key,
        articles=articles,
        now=now_utc,
        dedupe_hours=dconfig.dedupe_hours,
    )
    plog.info(
        "After dedupe filter: {} articles (raw={})",
        len(filtered_articles), len(articles),
    )

    if len(filtered_articles) < dconfig.min_articles:
        plog.warning(
            "Too few articles ({} < {}), skipping",
            len(filtered_articles), dconfig.min_articles,
        )
        return

    plog.info("Summarizer start")
    t0 = time.time()
    categories = summarize(filtered_articles, platform=dconfig.platform)
    item_total = sum(len(c.get("items", [])) for c in categories)
    plog.info(
        "Generated {} categories, {} items in {:.1f}s",
        len(categories), item_total, time.time() - t0,
    )

    if item_total == 0:
        plog.warning("No items generated by summarizer, skipping")
        return

    plog.info("Card builder start")
    start_time_utc = now_utc - timedelta(hours=dconfig.fetch_hours)
    payload = build_ai_daily_card(
        categories,
        platform=dconfig.platform,
        start_time=start_time_utc.astimezone(cn_tz),
        end_time=now_utc.astimezone(cn_tz),
    )

    preview_path = f"tmp/hourly_bot_{platform_key}_preview.json"
    with open(preview_path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    plog.info("Preview saved to: {}", preview_path)

    if dry_run:
        plog.info("Dry run mode — card NOT sent")
        return

    if not webhook_url:
        plog.error("FEISHU_WEBHOOK_URL not configured")
        return

    plog.info("Sending card to Feishu")
    try:
        result = send_card(webhook_url, payload)
        plog.info("Card sent: {}", result)
        mark_sent_articles(state, platform_key, filtered_articles, now_utc)
    except Exception:
        plog.exception("Failed to send card")


def main():
    parser = argparse.ArgumentParser(description="AI 资讯按平台节奏推送")
    parser.add_argument(
        "--only",
        choices=["x", "reddit", "hn", "all"],
        default="all",
        help="只抓某个平台 (默认 all)",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="忽略平台节奏限制，当前小时强制运行选中的平台",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="只生成预览，不发 Feishu",
    )
    args = parser.parse_args()

    sys.stdout.reconfigure(encoding="utf-8")

    cfg = load_config()
    init_logging(level=cfg.log_level, log_file=cfg.log_file)

    utc_now = datetime.now(timezone.utc)
    cn_tz = timezone(timedelta(hours=8))
    now_cn = utc_now.astimezone(cn_tz)

    log.info("AI hourly bot started")

    if args.only == "all":
        selected_keys = list(PLATFORM_FETCHERS.keys())
        if not args.force:
            due_keys = set(due_platform_keys(now_cn))
            selected_keys = [key for key in selected_keys if key in due_keys]
            log.info(
                "Current CN hour: {}, due platforms: {}",
                now_cn.strftime("%Y-%m-%d %H:%M"),
                ", ".join(selected_keys) or "none",
            )
    else:
        selected_keys = [args.only]

    if not selected_keys:
        log.info("No platforms due in this run, exiting")
        return

    state = load_delivery_state()
    for platform_key in selected_keys:
        try:
            run_platform(
                platform_key=platform_key,
                now_utc=utc_now,
                state=state,
                dry_run=args.dry_run,
                webhook_url=cfg.feishu_webhook_url,
            )
        except Exception:
            log.exception("Platform {} run failed", platform_key)
    save_delivery_state(state)
    log.info("AI hourly bot finished")


if __name__ == "__main__":
    main()