feat(groups): chat_group 补 6 群主/建群列 + 复合PK迁移修复 + 存量回填

对齐云客员工聊天窗口·微信群 tab(2026-08-25 真机验证 raw_json 键): - ChatGroup 加列 owner_name/owner_wxid/owner_alias/owner_avatar/ friend_count/create_time(全 nullable) - scraper _normalize_chat_group 同名提取;createTime 兼容 ms-int 与 'YYYY-MM-DD HH:mm:ss'(smoke.db 实测 1105 行全为后者),复用 _parse_add_time 归一 ms-epoch - 迁移 099ac2097fa2:加 6 列 + 修复 Task1 评审遗留 drift——PK 由单列 room_id 变更为复合 (room_id, channel, owner_account_id) 且 owner_account_id NOT NULL(对齐模型 8a99940;smoke.db 1105 行无 NULL, 未兜底编造数据);SQLite 走 batch 重建(copy_from 显式旧结构), MySQL 渲染为直接 ALTER(离线 --sql 已验证) - 回填脚本 scripts/backfill_chat_group_fields.py(幂等,逐列只补空); smoke.db 实跑 backfilled 1105/1105,重跑 0/1105 - GroupOut 加 6 字段(model_validate 自动输出,路由无改) Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-25 14:32:12 +08:00
parent 90c28f4c7b
commit 6ff288f1d3
7 changed files with 342 additions and 1 deletions
@@ -0,0 +1,103 @@
+"""Backfill ChatGroup 群主/建群补充列 from raw_json (no re-scrape).
+
+2026-08-25 对齐云客 员工聊天窗口·微信群 tab 的 6 个补充列(迁移
+099ac2097fa2 加列):从既有行的 raw_json 提取 ownerName/ownerWchatId/
+ownerAlias/ownerImage/friendCount/createTime 写入同名列。
+
+幂等:逐列只在「列为空 且 raw_json 有可用值」时回填,重跑只补新空列。
+createTime 兼容 ms-int 与 'YYYY-MM-DD HH:mm:ss'(smoke.db 实测全量为
+后者),复用 scraper 层 _parse_add_time 归一为 ms-epoch int。
+"""
+import asyncio
+import time
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.db import models  # noqa: F401
+from app.db.models import ChatGroup
+from app.scraper.wechat import _coerce_int_or_none, _parse_add_time
+
+BATCH = 200
+
+# raw_json 键 -> 目标列(字符串直提)
+_STR_FIELDS: dict[str, str] = {
+    "ownerName": "owner_name",
+    "ownerWchatId": "owner_wxid",
+    "ownerAlias": "owner_alias",
+    "ownerImage": "owner_avatar",
+}
+
+
+def _extract(raw: dict, current: dict) -> dict:
+    """raw_json -> 列值(只为「当前为空」的列产出值)。"""
+    out: dict = {}
+    for key, col in _STR_FIELDS.items():
+        if current.get(col) in (None, "") and raw.get(key):
+            out[col] = raw[key]
+    if current.get("friend_count") is None:
+        v = _coerce_int_or_none(raw.get("friendCount"))
+        if v is not None:
+            out["friend_count"] = v
+    if current.get("create_time") is None:
+        v = _parse_add_time(raw.get("createTime"))
+        if v is not None:
+            out["create_time"] = v
+    return out
+
+
+async def main(db_url: str = "sqlite+aiosqlite:///./smoke.db") -> None:
+    engine = create_async_engine(db_url)
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    scanned = updated = 0
+    async with sm() as s:
+        rows = (
+            await s.execute(
+                select(
+                    ChatGroup.room_id,
+                    ChatGroup.channel,
+                    ChatGroup.owner_account_id,
+                    ChatGroup.raw_json,
+                    ChatGroup.owner_name,
+                    ChatGroup.owner_wxid,
+                    ChatGroup.owner_alias,
+                    ChatGroup.owner_avatar,
+                    ChatGroup.friend_count,
+                    ChatGroup.create_time,
+                )
+            )
+        ).all()
+        for rid, ch, ow, raw, *cols in rows:
+            scanned += 1
+            if not raw:
+                continue
+            current = dict(
+                zip(
+                    ("owner_name", "owner_wxid", "owner_alias", "owner_avatar",
+                     "friend_count", "create_time"),
+                    cols,
+                )
+            )
+            values = _extract(raw, current)
+            if not values:
+                continue
+            await s.execute(
+                update(ChatGroup)
+                .where(
+                    ChatGroup.room_id == rid,
+                    ChatGroup.channel == ch,
+                    ChatGroup.owner_account_id == ow,
+                )
+                .values(**values)
+            )
+            updated += 1
+            if updated % BATCH == 0:
+                await s.commit()
+                print(f"[{time.strftime('%H:%M:%S')}] {updated} updated...")
+        await s.commit()
+    print(f"backfilled {updated}/{scanned} rows")
+    await engine.dispose()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())