fix(meituan-etl): 清洗文本 NUL 字节 + 逐城入库容错(修全量首灌崩在脏数据) (#58)

全国 359 城全量首灌崩在厦门:美团某券文本字段含 NUL(0x00),PostgreSQL
text/jsonb 拒绝该字节,整批 upsert 抛 DataError → --once 进程崩、后续 300+ 城
全不跑(本地 20 城没撞上、跑全国才暴露)。两处修复:
- _strip_nul 递归清洗入库 dict 所有字符串(含 raw JSON)的 NUL;
- 逐城 _upsert 包 try/except + rollback,单城入库失败记 fails 跳过,不再让
  一条脏数据 / 一次抖动拖垮整轮 359 城。

---------

Co-authored-by: chenshuobo <1119780489@qq.com>
Reviewed-on: #58
Co-authored-by: chenshuobo <chenshuobo@wonderable.ai>
Co-committed-by: chenshuobo <chenshuobo@wonderable.ai>

scripts/pull_meituan_coupons.py

@@ -202,6 +202,18 @@ def _to_cents(yuan) -> int | None:
         return None
 
 
+def _strip_nul(v):
+    """递归去掉字符串里的 NUL(0x00):PostgreSQL 的 text / jsonb 字段都不接受该字节,
+    美团偶有脏数据(某券文本含 NUL)会让整批 upsert 抛 DataError。入库前统一清洗。"""
+    if isinstance(v, str):
+        return v.replace(chr(0), "")
+    if isinstance(v, dict):
+        return {k: _strip_nul(x) for k, x in v.items()}
+    if isinstance(v, list):
+        return [_strip_nul(x) for x in v]
+    return v
+
+
 def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
     cpd = item.get("couponPackDetail") or {}
     br = item.get("brandInfo") or {}
@@ -231,7 +243,8 @@ def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
         dist = None
 
     dedup_raw = f"{brand}|{name}|{price_cents}"
-    return {
+    # 入库前递归清洗 NUL(美团脏数据偶含 0x00,PostgreSQL text/jsonb 拒绝整批 → DataError)
+    return _strip_nul({
         "source": source["code"],
         "platform": source["platform"],
         "biz_line": item.get("bizLine") or cpd.get("bizLine"),
@@ -252,7 +265,7 @@ def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
         "delivery_distance_m": dist,
         "dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
         "raw": item,
-    }
+    })
 
 
 def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
@@ -372,7 +385,13 @@ def run_once(
                     fails.append(city["name"])
                     print(f"  [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
                     continue
-                up, _dup = _upsert(db, parsed, now)
+                try:
+                    up, _dup = _upsert(db, parsed, now)
+                except Exception as e:  # noqa: BLE001
+                    db.rollback()  # 事务已 abort,必须 rollback 才能继续给下一城复用 Session
+                    fails.append(city["name"])
+                    print(f"  [入库失败] {city['name']}: {type(e).__name__}: {str(e)[:80]}")
+                    continue
                 total_up += up
                 done += 1
                 if done % 20 == 0 or done == n_city: