增加当下供需gap-分词过滤

agent/llm/openrouter.py

@@ -789,3 +789,15 @@ def create_openrouter_llm_call(
         return await openrouter_llm_call(messages, model, tools, **kwargs)
 
     return llm_call
+if __name__ == '__main__':
+    result =  openrouter_llm_call(
+        [
+            {"role": "system", "content": "你是简洁的中文助手。"},
+            {"role": "user", "content": '你的模型名称叫什么'},
+        ],
+        model='anthropic/claude-sonnet-4.5',
+        temperature=0.3,
+        max_tokens=256,
+    )
+    print(result)
+

examples/demand/data_query_tools.py

@@ -63,6 +63,40 @@ def _hive_partition_dt() -> str:
     return datetime.now(_CHINA_TZ).date().strftime(_HIVE_DT_FMT)
 
 
+def _hive_yesterday_partition_dt() -> str:
+    """中国时区（Asia/Shanghai）昨天日期，格式 yyyymmdd。"""
+    return (datetime.now(_CHINA_TZ).date() - timedelta(days=1)).strftime(_HIVE_DT_FMT)
+
+
+def _fenci_demand_filter_key(name: str, merge_leve2: str, type_str: str) -> tuple[str, str, str]:
+    """当下供需gap-分词策略的去重键（不含 dt，用于跨天过滤）。"""
+    return (name.strip(), merge_leve2.strip(), type_str.strip())
+
+
+def _get_yesterday_fenci_demand_keys() -> set[tuple[str, str, str]]:
+    """
+    从 MySQL demand_content 读取昨天产生的全部需求（含未写入 ODPS 的），
+    用于「当下供需gap-分词」策略的跨天去重。
+    """
+    yesterday = _hive_yesterday_partition_dt()
+    rows = mysql_db.select(
+        "demand_content",
+        columns="merge_leve2, name, ext_data",
+        where="dt = %s",
+        where_params=(yesterday,),
+    )
+    keys: set[tuple[str, str, str]] = set()
+    for row in rows:
+        name = str(row.get("name") or "").strip()
+        merge_leve2 = str(row.get("merge_leve2") or "").strip()
+        if not name or not merge_leve2:
+            continue
+        ext_data = _parse_ext_data(row.get("ext_data"))
+        type_str = str(ext_data.get("type") or "").strip()
+        keys.add(_fenci_demand_filter_key(name, merge_leve2, type_str))
+    return keys
+
+
 def _escape_odps_string(value: object) -> str:
     return str(value).replace("'", "''")
 
@@ -129,13 +163,16 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
     执行两次 INSERT（同表、同分区），策略不同：
     1) 当下供需gap: demand_name=merge_leve2+' '+name, demand_id=md5(strategy+demand_name+type+dt)
     2) 当下供需gap-分词: demand_name=name, demand_id=md5(strategy+name+品类+type+dt)
+       - 写入前过滤掉昨天 MySQL demand_content 中已产生的全部需求（按 name+品类+type 去重）
     """
     if not rows:
         return 0
 
     china_today = _hive_partition_dt()
+    yesterday_keys = _get_yesterday_fenci_demand_keys()
     gap_parts: list[str] = []
     fenci_parts: list[str] = []
+    fenci_skipped = 0
 
     for row in rows:
         merge_leve2 = str(row.get("merge_leve2") or "").strip()
@@ -165,6 +202,11 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
             )
         )
 
+        filter_key = _fenci_demand_filter_key(name, merge_leve2, type_str)
+        if filter_key in yesterday_keys:
+            fenci_skipped += 1
+            continue
+
         demand_id_fenci = hashlib.md5(
             f"{_STRATEGY_GAP_FENCI}{name}{merge_leve2}{type_str}{china_today}".encode("utf-8")
         ).hexdigest()
@@ -178,8 +220,16 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
     if not gap_parts:
         return 0
 
+    print(
+        f"[hive] {_STRATEGY_GAP_FENCI} 过滤昨天需求: "
+        f"yesterday_dt={_hive_yesterday_partition_dt()}, "
+        f"yesterday_total={len(yesterday_keys)}, skipped={fenci_skipped}, "
+        f"to_write={len(fenci_parts)}"
+    )
+
     _insert_hive_select_parts(gap_parts, china_today)
-    _insert_hive_select_parts(fenci_parts, china_today)
+    if fenci_parts:
+        _insert_hive_select_parts(fenci_parts, china_today)
     return len(gap_parts) + len(fenci_parts)