init

src/handlers/demand_enqueue.py

@@ -5,22 +5,34 @@ XXL-JOB Admin 配置:
     处理器: demandEnqueue
     参数:   dt=yyyyMMdd (可选，不传则默认前天)
 
-设计: Admin 调度的 HTTP 请求秒级返回，实际入队逻辑通过后台异步任务执行，
-     避免 ODPS 查询耗时导致 Admin 侧 Read timed out。
+设计:
+    - Admin 调度的 HTTP 请求秒级返回，实际入队在后台异步执行
+    - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
 """
 
 import asyncio
 import logging
 from datetime import datetime, timedelta
+from typing import TYPE_CHECKING
 
 from src.config.odps import OdpsConfig
-from src.config.database import LongArticlesDatabaseConfig
-from src.infra.database import create_backend, MysqlManager
 from src.infra.xxl_jobs import xxl_job
 from src.domains.demand import FetchDemands, EnqueueDemands
 
+if TYPE_CHECKING:
+    from src.infra.database import DatabaseManager
+
 logger = logging.getLogger(__name__)
 
+# 由 create_app() 在启动时注入
+_db_ref: "DatabaseManager | None" = None
+
+
+def set_db(db: "DatabaseManager") -> None:
+    """注入 app 级别的 DatabaseManager，供 handler 复用已有连接池"""
+    global _db_ref
+    _db_ref = db
+
 
 def _parse_dt(param: str) -> str:
     if param and param.strip():
@@ -36,10 +48,8 @@ async def _do_enqueue(dt: str) -> None:
     logger.info("demandEnqueue background task started, dt=%s", dt)
 
     try:
-        # ── 1. ODPS 拉取 ────────────────────────────
         fetcher = FetchDemands(OdpsConfig())
         demands = await fetcher.deal(dt)
-
         if not demands:
             logger.info("无需求数据, dt=%s", dt)
             return
@@ -47,18 +57,13 @@ async def _do_enqueue(dt: str) -> None:
         logger.exception("ODPS 查询失败, dt=%s", dt)
         return
 
-    # ── 2. MySQL 入队 ──────────────────────────────
-    backend = create_backend("long_articles", LongArticlesDatabaseConfig())
     try:
-        await backend.init()
-        mysql = MysqlManager(backend)
+        mysql = _db_ref.mysql_manager("long_articles")
         enqueuer = EnqueueDemands(mysql)
         n = await enqueuer.deal(demands)
     except Exception:
         logger.exception("MySQL 入队失败, dt=%s", 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)

src/server/app.py

@@ -12,6 +12,7 @@ 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
+from src.handlers.demand_enqueue import set_db
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +43,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     xxl_executor = XxlJobExecutor(global_config.xxl_job)
 
     set_xxl_executor(xxl_executor)
+    set_db(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")