fix: sanitize lone surrogate chars in stored data + backfill script

Upstream API records may contain lone surrogate chars (emoji fragments)
that break SQLite/MySQL utf-8 encoding. Now sanitized to U+FFFD before
storage. Includes backfill_customers.py for full customer backfill.

Co-Authored-By: Claude <noreply@anthropic.com>

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:

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())