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

Merged
marco merged 1 commits from intern/chenshuobo/meituan-etl-nul-fix-20260616 into main 2026-06-16 21:19:48 +08:00
+22 -3
View File
@@ -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: