增加同步任务和日志上传

agents/demand_belong_category_agent/run.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Run find_agent interactively."""
+"""Run demand_belong_category_agent."""
 
 from agents.demand_belong_category_agent import create_demand_belong_category_agent
 

supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py

@@ -4,10 +4,9 @@
 流程：
 1. 比对当天 ODPS / MySQL 行数，相同则跳过写入
 2. 有差异时拉取 ODPS，按 (strategy, demand_id) 只插入缺失、删除多余
-3. 查询当天 demand_name，按空格分词并去重
+3. 查询当天全部 demand_name，按空格分词写入 set
 4. 过滤 demand_belong_category 中已存在的词
-5. 剩余词按 100 个一批，调用 demand_belong_category_agent
-   （测试阶段仅调用 1 批）
+5. 剩余词按 100 词一批调用 demand_belong_category_agent（测试阶段仅 1 批）
 """
 from __future__ import annotations
 
@@ -62,62 +61,58 @@ def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str
     return list(by_key.values())
 
 
-def _tokenize_demand_names(demand_names: list[str]) -> list[str]:
-    """按空格分词，去空、去重（保持首次出现顺序）。"""
-    seen: set[str] = set()
-    tokens: list[str] = []
+def _build_word_set(demand_names: list[str]) -> set[str]:
+    """demand_name 按空格分词，全部落入同一个 set。"""
+    words: set[str] = set()
     for name in demand_names:
         for token in str(name).split():
             word = token.strip()
-            if not word or word in seen:
-                continue
-            seen.add(word)
-            tokens.append(word)
-    return tokens
+            if word:
+                words.add(word)
+    return words
 
 
 def _chunked(items: list[str], size: int) -> list[list[str]]:
     return [items[i : i + size] for i in range(0, len(items), size)]
 
 
-def _classify_new_words(biz_dt: str, *, max_batches: int | None = _MAX_CLASSIFY_BATCHES) -> dict:
-    """分词 → 过滤已存在词 → 分批调用归属分类 agent。"""
+def _classify_words(biz_dt: str, *, max_batches: int | None = _MAX_CLASSIFY_BATCHES) -> dict:
+    """查询 demand_name → 空格分词入 set → 过滤已有词 → 100 词一批调用 agent。"""
     with get_session() as session:
-        pool_repo = MultiDemandPoolDiRepository(session)
-        category_repo = DemandBelongCategoryRepository(session)
+        demand_names = MultiDemandPoolDiRepository(session).list_demand_names_by_biz_dt(biz_dt)
+        word_set = _build_word_set(demand_names)
+        existing = DemandBelongCategoryRepository(session).get_existing_names(word_set)
+        pending = word_set - existing
 
-        demand_names = pool_repo.list_demand_names_by_biz_dt(biz_dt)
-        tokens = _tokenize_demand_names(demand_names)
-        existing = category_repo.get_existing_names(tokens)
-        new_words = [w for w in tokens if w not in existing]
-
-    batches = _chunked(new_words, _WORD_BATCH_SIZE)
+    word_list = list(pending)
+    batches = _chunked(word_list, _WORD_BATCH_SIZE)
     if max_batches is not None:
         batches = batches[:max_batches]
 
     logger.info(
-        "Classify prepare: demand_names=%d tokens=%d existing=%d new=%d batches=%d (max=%s)",
+        "Classify prepare: demand_names=%d words=%d existing=%d pending=%d batches=%d (max=%s)",
         len(demand_names),
-        len(tokens),
+        len(word_set),
         len(existing),
-        len(new_words),
+        len(pending),
         len(batches),
         max_batches,
     )
 
     classified_batches = 0
     for idx, batch in enumerate(batches, start=1):
+        batch = list(dict.fromkeys(batch))
         logger.info("Classifying batch %d/%d (%d words)", idx, len(batches), len(batch))
         classify_demand_words(batch)
         classified_batches += 1
 
     return {
         "demand_names": len(demand_names),
-        "tokens": len(tokens),
+        "words": len(word_set),
         "existing_filtered": len(existing),
-        "new_words": len(new_words),
-        "batches_total": (len(new_words) + _WORD_BATCH_SIZE - 1) // _WORD_BATCH_SIZE
-        if new_words
+        "pending": len(pending),
+        "batches_total": (len(word_list) + _WORD_BATCH_SIZE - 1) // _WORD_BATCH_SIZE
+        if word_list
         else 0,
         "batches_ran": classified_batches,
     }
@@ -199,7 +194,7 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
             **_sync_diff(partition_date),
         }
 
-    classify_stats = _classify_new_words(partition_date)
+    classify_stats = _classify_words(partition_date)
 
     result = {
         "partition_date": partition_date,