增加需求过滤

examples/demand/data_query_tools.py

@@ -73,6 +73,38 @@ def _fenci_demand_filter_key(name: str, merge_leve2: str, type_str: str) -> tupl
     return (name.strip(), merge_leve2.strip(), type_str.strip())
 
 
+def _get_demand_filter_words(bizdate: str | None = None) -> list[str]:
+    """
+    从 ODPS demand_filter_word 读取当天过滤词。
+    dt 为中国时区当天（yyyymmdd）；查询失败时返回空列表（不过滤）。
+    """
+    dt = bizdate or _hive_partition_dt()
+    sql = f"""
+SELECT  filter_word
+FROM    demand_filter_word
+WHERE   dt = '{dt}'
+"""
+    records = get_odps_data(sql)
+    if not records:
+        return []
+    words: list[str] = []
+    seen: set[str] = set()
+    for record in records:
+        word = str(record[0] or "").strip()
+        if not word or word in seen:
+            continue
+        seen.add(word)
+        words.append(word)
+    return words
+
+
+def _text_contains_filter_word(text: str, filter_words: list[str]) -> bool:
+    """若 text 包含任一过滤词（等价 LIKE '%word%'）返回 True。"""
+    if not text or not filter_words:
+        return False
+    return any(word in text for word in filter_words if word)
+
+
 def _get_yesterday_fenci_demand_keys() -> set[tuple[str, str, str]]:
     """
     从 MySQL demand_content 读取昨天产生的全部需求（含未写入 ODPS 的），
@@ -160,6 +192,7 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
     将行数据映射并写入 loghubods.dwd_multi_demand_pool_di（尽力插入，不校验结果）。
 
     分区与 demand_id 的日期均为中国时区当天（yyyymmdd），不使用行内 dt 字段。
+    写入前从 demand_filter_word（dt=当天）拉取过滤词，demand_name 包含任一过滤词则跳过。
     执行两次 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)
@@ -169,10 +202,16 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
         return 0
 
     china_today = _hive_partition_dt()
+    filter_words = _get_demand_filter_words(china_today)
     yesterday_keys = _get_yesterday_fenci_demand_keys()
     gap_parts: list[str] = []
     fenci_parts: list[str] = []
     fenci_skipped = 0
+    filter_word_skipped = 0
+
+    print(
+        f"[hive] 过滤词加载: dt={china_today}, count={len(filter_words)}"
+    )
 
     for row in rows:
         merge_leve2 = str(row.get("merge_leve2") or "").strip()
@@ -192,6 +231,11 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
         extend_json = json.dumps({"品类": merge_leve2}, ensure_ascii=False)
 
         demand_name_gap = f"{merge_leve2} {name}"
+        # 等价 SQL LIKE '%filter_word%'：demand_name 包含任一过滤词则不写入 ODPS
+        if _text_contains_filter_word(demand_name_gap, filter_words):
+            filter_word_skipped += 1
+            continue
+
         demand_id_gap = hashlib.md5(
             f"{_STRATEGY_GAP}{demand_name_gap}{type_str}{china_today}".encode("utf-8")
         ).hexdigest()
@@ -217,6 +261,11 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
             )
         )
 
+    print(
+        f"[hive] 过滤词跳过: skipped={filter_word_skipped}, "
+        f"gap_to_write={len(gap_parts)}, fenci_to_write={len(fenci_parts)}"
+    )
+
     if not gap_parts:
         return 0