feat(scraper): WA群抓取(type=1真机验证,223群)+群成员派生(12840成员/750群)

- wa-groups 实体:businessFriendList type=1 真机验证返回群,行结构是好友字段名
  (friendUid=@g.us 群ID/nickName=群名),专用 _normalize_wa_group_row
- backfill_group_members.py:无独立成员API,从群消息派生(talker=成员,sender_name=昵称)
- full_sync_all.sh:+4步(5b WA群/语音后录音重链接/8b 群消息+成员/消息上限99999)

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

yunke-server/scripts/backfill_group_members.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+"""从群消息派生 group_member（发言者=成员）。
+
+云客 API 不单独给群成员列表（chatroom payload 的 memberIds 我们抓不到，
+2026-08-20 确认），但群消息的 talker 就是发言成员的 wxid、sender_name 是其
+群昵称——368k 群消息足够派生出高质量成员表（覆盖所有发过言的成员）。
+
+昵称取该成员在群里最新一条消息的 sender_name（SQLite 的 max()+裸列行为，
+取 max(timestamp) 同行的值）。幂等可重跑；建议在 group-messages 之后执行。
+
+用法:
+    cd yunke-server && DB_URL="sqlite+aiosqlite:///./smoke.db" \
+      .venv/Scripts/python.exe -m scripts.backfill_group_members
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import text as sa_text
+
+from app.db.base import async_engine, async_session_maker
+from app.db.models import GroupMember
+from app.storage.repos import _upsert_one
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("backfill_group_members")
+
+# max(timestamp) 裸列：SQLite 保证 sender_name 取自 timestamp 最大的那行
+SQL = sa_text("""
+SELECT room_id, talker_id, sender_name, max(timestamp) AS last_ts, count(*) AS msg_count
+FROM message
+WHERE channel = 'wechat' AND room_id IS NOT NULL AND room_id != ''
+  AND talker_id IS NOT NULL AND talker_id != ''
+GROUP BY room_id, talker_id
+""")
+
+
+async def main() -> None:
+    async with async_engine.connect() as conn:
+        rows = (await conn.execute(SQL)).fetchall()
+    log.info(f"派生出 {len(rows)} 个 (群, 成员) 对")
+
+    async with async_session_maker() as session:
+        for room_id, talker_id, sender_name, last_ts, msg_count in rows:
+            await _upsert_one(session, GroupMember, {
+                "channel": "wechat",
+                "room_id": room_id,
+                "member_id": talker_id,
+                "nickname": sender_name,
+                # raw_json 附加派生统计，抽屉可见
+                "raw_json": {
+                    "derived_from": "group_messages",
+                    "msg_count": msg_count,
+                    "last_msg_ts": last_ts,
+                },
+            })
+        await session.commit()
+    log.info(f"DONE: group_member upserted={len(rows)}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

yunke-server/scripts/full_sync_all.sh

@@ -18,14 +18,21 @@ echo "$(date) --- 4/10 WA customers ---"
 $PY scripts/scrape_via_page.py --entity wa-friends
 echo "$(date) --- 5/10 WeChat groups ---"
 $PY scripts/scrape_via_page.py --entity groups --max-employees 100
+echo "$(date) --- 5b/10 WA groups ---"
+$PY scripts/scrape_via_page.py --entity wa-groups
 echo "$(date) --- 6/10 Voice calls (WeChat + WA) ---"
 $PY scripts/scrape_via_page.py --entity voice-calls
+# 语音重扫会把 record_media_hash 覆盖回 None —— 必须紧跟录音重链接
+$PY -m app.sync.cli sync --entity call_recordings
 echo "$(date) --- 7/10 Voice clips ---"
 $PY scripts/scrape_via_page.py --entity voice-clips
 echo "$(date) --- 8/10 WeChat messages ---"
-$PY scripts/scrape_via_page.py --entity messages --max-conversations 1000
+$PY scripts/scrape_via_page.py --entity messages --max-conversations 99999
+echo "$(date) --- 8b/10 WeChat group messages + members ---"
+$PY scripts/scrape_via_page.py --entity group-messages --max-conversations 99999
+$PY -m scripts.backfill_group_members
 echo "$(date) --- 9/10 WA messages ---"
-$PY scripts/scrape_via_page.py --entity wa-messages --max-conversations 500
+$PY scripts/scrape_via_page.py --entity wa-messages --max-conversations 99999
 echo "$(date) --- 10/10 Media download (audio+images) ---"
 $PY scripts/scrape_via_page.py --entity media
 

yunke-server/scripts/scrape_via_page.py

@@ -946,6 +946,84 @@ async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
     return total_fetched, total_upserted
 
 
+def _normalize_wa_group_row(row: dict) -> dict | None:
+    """businessFriendList type=1 行 → ChatGroup 形状（真机确认 2026-08-20）。
+
+    type=1 返回的是好友字段名结构：friendUid=群 id（@g.us 后缀）、nickName=
+    群名、uid=员工 WA 号、salesNickName=员工 WA 昵称——与微信群的
+    friendWxId/friendWxNickname 完全不同，故不能复用 _normalize_chat_group。
+    非 @g.us 行（异常混入）返回 None。
+    """
+    friend_uid = str(row.get("friendUid") or "")
+    if not friend_uid.endswith("@g.us"):
+        return None
+    return {
+        "room_id": friend_uid,
+        "channel": "whatsapp",
+        "room_name": row.get("nickName"),
+        "owner_account_id": row.get("uid"),
+        "owner_account_nickname": row.get("salesNickName"),
+        "owner_user_id": row.get("userId"),
+        "owner_user_phone": row.get("salePhone"),
+        "message_count": row.get("messageGroupCount"),
+        "raw_json": row,
+    }
+
+
+async def scrape_wa_groups(cookie_str: str, max_pages: int = 200):
+    """Scrape WhatsApp GROUPS via browser fetch POST businessFriendList type=1.
+
+    Same endpoint as WA friends with ``type:"1"`` — live-verified 2026-08-20:
+    rows carry friendUid ending ``@g.us`` (group id), nickName = group name.
+    Envelope: data.overviews; upsert to chat_group (channel='whatsapp').
+    """
+    from app.db.base import async_engine
+    from sqlalchemy import text as sa_text
+    from app.storage.repos import upsert_chat_group
+
+    async with async_engine.connect() as conn:
+        row = (await conn.execute(sa_text(
+            "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1"
+        ))).first()
+    department_id = row[0] if row else ""
+    if not department_id:
+        logger.warning("wa-groups: no root department in DB — abort")
+        return 0, 0
+
+    page_size = 50
+    total_fetched = 0
+    total_upserted = 0
+    async with PageScraper(cookie_str) as scraper:
+        page = await _open_fetch_page(scraper)
+        page_num = 1
+        while page_num <= max_pages:
+            body = await _browser_fetch(
+                page,
+                INTERCEPT_PATHS["wa_friend"],
+                json_body={
+                    "searchInfo": "",
+                    "friendSearchInfo": "",
+                    "page": page_num,
+                    "pageSize": page_size,
+                    "departmentId": department_id,
+                    "type": "1",
+                },
+            )
+            rows = _rows_overviews(body)
+            records = [rec for r in rows if (rec := _normalize_wa_group_row(r)) is not None]
+            total_fetched += len(records)
+            await _upsert_all(upsert_chat_group, records)
+            total_upserted += len(records)
+            logger.info(f"  wa-groups page {page_num}: {len(rows)} rows, {len(records)} groups")
+            if len(rows) < page_size:
+                break
+            page_num += 1
+            await asyncio.sleep(1.5)
+
+    logger.info(f"DONE: wa-groups fetched={total_fetched}, upserted={total_upserted}")
+    return total_fetched, total_upserted
+
+
 # ---------------------------------------------------------------------------
 # Voice call records (navigation + interception, both WeChat + WA pages)
 # ---------------------------------------------------------------------------
@@ -1299,6 +1377,7 @@ async def main():
             "accounts",      # 微信员工账号 (wechatlist page interception)
             "wa-accounts",   # WA员工账号 (waBusiness/userlist interception)
             "wa-friends",    # WA客户 (waBusiness/friendlist interception)
+            "wa-groups",     # WA群 (browser fetch businessFriendList type=1)
             "voice-calls",   # 语音通话记录 (voice-call-list + wavoicecalllist)
             "voice-clips",   # 长语音 (browser fetch call/list per employee)
             "wa-messages",   # WA消息 (browser fetch chatRecords per conversation)
@@ -1337,6 +1416,8 @@ async def main():
         await scrape_wa_accounts(cookie, args.max_pages)
     elif args.entity == "wa-friends":
         await scrape_wa_friends(cookie, args.max_pages)
+    elif args.entity == "wa-groups":
+        await scrape_wa_groups(cookie, args.max_pages)
     elif args.entity == "voice-calls":
         await scrape_voice_calls(cookie, args.max_pages)
     elif args.entity == "voice-clips":