1

examples/content_finder/db/schedule.py

@@ -202,18 +202,28 @@ def get_one_today_unprocessed_demand(*, dt: int) -> Optional[Dict[str, Any]]:
     - dt 与表字段一致：一般为 YYYYMMDD 整数（如 20260402）
 
     Returns:
-        {"demand_content_id": int, "query": str, "suggestion": Optional[str], "score": Any} 或 None
+        {
+            "demand_content_id": int,
+            "query": str,
+            "suggestion": Optional[str],
+            "score": Any,
+            "merge_leve2": Optional[str],
+            "category_rank": int,
+        } 或 None
     """
     sql = """
     SELECT x.demand_content_id,
            x.query,
            x.suggestion,
-           x.score
+           x.score,
+           x.merge_leve2,
+           x.rn AS category_rank
     FROM (
         SELECT dc.id AS demand_content_id,
                dc.name AS query,
                dc.suggestion AS suggestion,
                dc.score AS score,
+               dc.merge_leve2 AS merge_leve2,
                ROW_NUMBER() OVER (
                    PARTITION BY COALESCE(dc.merge_leve2, '')
                    ORDER BY dc.score DESC, dc.id DESC

examples/content_finder/server.py

@@ -251,7 +251,8 @@ def _has_running_content_task() -> bool:
 async def scheduled_tick():
     """
     按 SCHEDULE_DISPATCH_INTERVAL_SECONDS 派发：若当前并发有空槽，则从 demand_content 取
-    当天(dt=今日)、尚未出现在 demand_find_task 中且 score 最高的一条需求并执行。
+    当天(dt=今日)、尚未出现在 demand_find_task 中的 1 条需求并执行。
+    调度顺序按“品类分层轮转”：先各品类 top1（层内 score 高者优先），再各品类 top2，依此类推。
     """
     if SCHEDULE_DISPATCH_NOT_BEFORE_HOUR is not None:
         now = datetime.now(SCHEDULER_TZ)
@@ -307,9 +308,15 @@ async def scheduled_tick():
         return
 
     score = item.get("score")
+    merge_leve2 = (item.get("merge_leve2") or "").strip()
+    category_rank = item.get("category_rank")
     logger.info(
-        f"定时任务领取（当天 score 最高）：demand_content_id={demand_content_id}, "
-        f"dt={dt}, score={score}"
+        "定时任务领取（品类分层轮转）：demand_content_id=%s, dt=%s, score=%s, merge_leve2=%s, category_rank=%s",
+        demand_content_id,
+        dt,
+        score,
+        merge_leve2 or "<EMPTY>",
+        category_rank,
     )
     create_task_record(demand_content_id)
     # 后台执行：由 execute_task 内部 semaphore 控制并发占用