init

src/handlers/demand_enqueue.py

@@ -4,6 +4,9 @@ XXL-JOB Admin 配置:
     执行器: long-articles-agentic-src
     处理器: demandEnqueue
     参数:   dt=yyyyMMdd (可选，不传则默认前天)
+
+设计: Admin 调度的 HTTP 请求秒级返回，实际入队逻辑通过后台异步任务执行，
+     避免 ODPS 查询耗时导致 Admin 侧 Read timed out。
 """
 
 import asyncio
@@ -20,41 +23,32 @@ logger = logging.getLogger(__name__)
 
 
 def _parse_dt(param: str) -> str:
-    """从 XXL-JOB 参数中提取 dt，默认前天"""
     if param and param.strip():
         for part in param.strip().split():
             if part.startswith("dt="):
                 return part[3:]
-        # 如果没有 dt= 前缀，直接把参数当日期字符串
         return param.strip()
     return (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")
 
 
-@xxl_job("demandEnqueue")
-async def demand_enqueue(param: str) -> dict:
-    """需求入队 handler
-
-    param: executorParams，格式 "dt=20260629" 或 "20260629" 或空
-    """
-    dt = _parse_dt(param)
-    logger.info("demandEnqueue started, dt=%s", dt)
+async def _do_enqueue(dt: str) -> None:
+    """后台执行: ODPS 拉取 → 过滤 → 聚合 → MySQL 入队"""
+    logger.info("demandEnqueue background task started, dt=%s", dt)
 
-    # ── 1. ODPS 拉取 ────────────────────────────────
-    odps_config = OdpsConfig()
-    fetcher = FetchDemands(odps_config)
     try:
+        # ── 1. ODPS 拉取 ────────────────────────────
+        fetcher = FetchDemands(OdpsConfig())
         demands = await fetcher.deal(dt)
+
+        if not demands:
+            logger.info("无需求数据, dt=%s", dt)
+            return
     except Exception:
         logger.exception("ODPS 查询失败, dt=%s", dt)
-        return {"code": 500, "msg": f"ODPS query failed, dt={dt}"}
-
-    if not demands:
-        logger.info("无需求数据, dt=%s", dt)
-        return {"code": 200, "msg": f"no demands found for dt={dt}"}
+        return
 
     # ── 2. MySQL 入队 ──────────────────────────────
-    db_config = LongArticlesDatabaseConfig()
-    backend = create_backend("long_articles", db_config)
+    backend = create_backend("long_articles", LongArticlesDatabaseConfig())
     try:
         await backend.init()
         mysql = MysqlManager(backend)
@@ -62,14 +56,22 @@ async def demand_enqueue(param: str) -> dict:
         n = await enqueuer.deal(demands)
     except Exception:
         logger.exception("MySQL 入队失败, dt=%s", dt)
-        return {"code": 500, "msg": f"MySQL enqueue failed, dt={dt}"}
+        return
     finally:
         await backend.close()
 
     accounts = len(set(d["channel_level3"] for d in demands))
     total_terms = sum(len(d["search_terms"]) for d in demands)
-    msg = (
-        f"enqueued {n} demands ({accounts} accounts, {total_terms} terms) for dt={dt}"
+    logger.info(
+        "demandEnqueue done: enqueued %d demands (%d accounts, %d terms) for dt=%s",
+        n, accounts, total_terms, dt,
     )
-    logger.info("demandEnqueue done: %s", msg)
-    return {"code": 200, "msg": msg}
+
+
+@xxl_job("demandEnqueue")
+async def demand_enqueue(param: str) -> dict:
+    """需求入队 handler — 秒级返回，实际工作在后台执行"""
+    dt = _parse_dt(param)
+    asyncio.create_task(_do_enqueue(dt))
+    logger.info("demandEnqueue accepted, dt=%s (background task spawned)", dt)
+    return {"code": 200, "msg": f"accepted, dt={dt}"}

src/server/app.py

@@ -11,7 +11,7 @@ from src.infra.database import DatabaseManager, create_backend
 from src.infra.observability import LogService
 from src.infra.xxl_jobs import XxlJobExecutor
 from src.handlers import discover_and_register
-from src.server.xxl_endpoints import set_xxl_executor, create_xxl_bp
+from src.server.xxl_endpoints import set_xxl_executor
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +45,6 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")
-    app.register_blueprint(create_xxl_bp())
 
     # 生命周期
     ctx = AppContext()