diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index 8283849..f993119 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -211,9 +211,29 @@ def _build_column_values( # truth. if "raw_json" in col_names and "raw_json" not in values: values["raw_json"] = rec + # Sanitize lone surrogate chars (e.g. \udf1e from emoji fragments in the + # upstream API) that break SQLite/MySQL UTF-8 encoding. Replaces them with + # the Unicode replacement character (U+FFFD). + for k, v in values.items(): + values[k] = _sanitize_surrogates(v) return values +def _sanitize_surrogates(obj: Any) -> Any: + """Recursively remove lone surrogate characters from strings/dicts/lists.""" + if isinstance(obj, str): + try: + obj.encode("utf-8") # fast-path: clean string, no copy + return obj + except UnicodeEncodeError: + return obj.encode("utf-8", "replace").decode("utf-8") + if isinstance(obj, dict): + return {k: _sanitize_surrogates(v) for k, v in obj.items()} + if isinstance(obj, list): + return [_sanitize_surrogates(v) for v in obj] + return obj + + def _build_upsert_stmt( model: type[DeclarativeBase], values: dict[str, Any], dialect_name: str ) -> sa.Insert: diff --git a/yunke-server/scripts/backfill_customers.py b/yunke-server/scripts/backfill_customers.py new file mode 100644 index 0000000..ae4a8ae --- /dev/null +++ b/yunke-server/scripts/backfill_customers.py @@ -0,0 +1,85 @@ +"""Full backfill of customers (WeChat + WhatsApp) into the DB. + +Unlike the seed scripts (tiny demo slices), this paginates the ENTIRE friend +list for both channels and upserts everything. Slow (anti-风控 paced), commits +per batch so it's resumable. Run in background; logs progress to stdout. +""" +import asyncio +import sys +import time + +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db import models # noqa: F401 +from app.scraper.wechat import scrape_departments, scrape_wechat_friends +from app.scraper.whatsapp import scrape_whatsapp_friends +from app.storage.repos import upsert_customers +from scripts.smoke_wechat import _push_settings_into_environ, build_client_from_env + +BATCH = 100 # upsert + commit every N records +PAGE_SIZE = 50 + + +def _log(msg: str) -> None: + print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True) + + +async def _root_dept(client) -> str | None: + async for d in scrape_departments(client): + if not d.get("parent_id"): + return d.get("dept_id") + return None + + +async def _backfill_channel(label: str, agen, session_maker) -> int: + total = 0 + batch: list[dict] = [] + async for rec in agen: + batch.append(rec) + if len(batch) >= BATCH: + async with session_maker() as s: + await upsert_customers(s, batch) + await s.commit() + total += len(batch) + _log(f"{label}: {total} upserted...") + batch = [] + if batch: + async with session_maker() as s: + await upsert_customers(s, batch) + await s.commit() + total += len(batch) + _log(f"{label}: DONE — {total} customers") + return total + + +async def main() -> None: + _push_settings_into_environ() + client = build_client_from_env() + if client is None: + _log("no auth env"); sys.exit(2) + engine = create_async_engine("sqlite+aiosqlite:///./smoke.db") + sm = async_sessionmaker(engine, expire_on_commit=False) + + _log("resolving root department...") + root = await _root_dept(client) + _log(f"root dept: {root}") + + _log("=== WeChat customers (full) ===") + wx = await _backfill_channel( + "wechat", scrape_wechat_friends(client, page_size=PAGE_SIZE), sm + ) + + _log("=== WhatsApp customers (full) ===") + wa = await _backfill_channel( + "whatsapp", + scrape_whatsapp_friends(client, department_id=root or "", page_size=PAGE_SIZE), + sm, + ) + + _log(f"\nTOTAL: {wx} wechat + {wa} whatsapp = {wx + wa} customers") + await client.aclose() + await engine.dispose() + + +if __name__ == "__main__": + asyncio.run(main())