Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a48889ffc8 |
@@ -202,6 +202,18 @@ def _to_cents(yuan) -> int | None:
|
|||||||
return 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:
|
def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
|
||||||
cpd = item.get("couponPackDetail") or {}
|
cpd = item.get("couponPackDetail") or {}
|
||||||
br = item.get("brandInfo") or {}
|
br = item.get("brandInfo") or {}
|
||||||
@@ -231,7 +243,8 @@ def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
|
|||||||
dist = None
|
dist = None
|
||||||
|
|
||||||
dedup_raw = f"{brand}|{name}|{price_cents}"
|
dedup_raw = f"{brand}|{name}|{price_cents}"
|
||||||
return {
|
# 入库前递归清洗 NUL(美团脏数据偶含 0x00,PostgreSQL text/jsonb 拒绝整批 → DataError)
|
||||||
|
return _strip_nul({
|
||||||
"source": source["code"],
|
"source": source["code"],
|
||||||
"platform": source["platform"],
|
"platform": source["platform"],
|
||||||
"biz_line": item.get("bizLine") or cpd.get("bizLine"),
|
"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,
|
"delivery_distance_m": dist,
|
||||||
"dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
|
"dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
|
||||||
"raw": item,
|
"raw": item,
|
||||||
}
|
})
|
||||||
|
|
||||||
|
|
||||||
def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
|
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"])
|
fails.append(city["name"])
|
||||||
print(f" [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
|
print(f" [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
|
||||||
continue
|
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
|
total_up += up
|
||||||
done += 1
|
done += 1
|
||||||
if done % 20 == 0 or done == n_city:
|
if done % 20 == 0 or done == n_city:
|
||||||
|
|||||||
Reference in New Issue
Block a user