long_articles_agentic_supply/src/handlers/demand_enqueue.py

"""需求入队 XXL-JOB Handler —— ODPS 拉取 → 聚合 → 写入 MySQL

XXL-JOB Admin 配置:
    执行器: long-articles-agentic-src
    处理器: demandEnqueue
    参数:   dt=yyyyMMdd (可选，不传则默认前天)
    超时:   建议 ≥300s（ODPS + PG + MySQL 多段查询）
"""

import logging
from datetime import datetime, timedelta

from src.config.odps import OdpsConfig
from src.infra.xxl_jobs import xxl_job
from src.domains.demand_enqueue_task import FetchDemands, EnqueueDemands
from src.handlers._utils import get_db, parse_dt

logger = logging.getLogger(__name__)


def _default_dt() -> str:
    return (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")


@xxl_job("demandEnqueue")
async def demand_enqueue(param: str) -> dict:
    dt = parse_dt(param, _default_dt())
    logger.info("demandEnqueue started, dt=%s", dt)

    # 1. ODPS 拉取 → PG 查搜索词
    try:
        pg = get_db().pg_manager("pg")
        fetcher = FetchDemands(OdpsConfig(), pg)
        items = await fetcher.deal(dt)
    except Exception:
        logger.exception("ODPS 拉取失败, dt=%s", dt)
        return {"code": 500, "msg": "ODPS 拉取异常"}

    if not items:
        logger.info("无队列项生成, dt=%s", dt)
        return {"code": 200, "msg": "no items"}

    # 2. MySQL 入队
    try:
        mysql = get_db().mysql_manager("long_articles")
        enqueuer = EnqueueDemands(mysql)
        n = await enqueuer.deal(items)
    except Exception:
        logger.exception("MySQL 入队失败, dt=%s", dt)
        return {"code": 500, "msg": "入队异常"}

    accounts = len(set(it["account"] for it in items))
    logger.info(
        "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
        n,
        accounts,
        dt,
    )
    return {"code": 200, "msg": f"enqueued={n}, accounts={accounts}"}