feat(scraper): WA群消息抓取(chatRecords friendUid=@g.us,chatRoomTalker=群内发送者)

- 真机探明:同 chatRecords 端点 friendUid 直接传 @g.us 群id(与微信群同套路)
- 群行 chatRoomTalker=发送者@lid、keyId 有值(缺失时合成稳定主键)
- file 字段直连 OSS URL 优先于 mediaName(WA群媒体 URL 现成!)
- 守护循环同微信群;full_sync 加 9b 步;3 新测试,416 绿

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

yunke-server/scripts/full_sync_all.sh

@@ -33,6 +33,8 @@ $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 99999
+echo "$(date) --- 9b/10 WA group messages ---"
+$PY scripts/scrape_via_page.py --entity wa-group-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

@@ -1371,6 +1371,131 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
     return total_fetched, total_upserted
 
 
+def _normalize_wa_group_message(record: dict) -> dict:
+    """chatRecords 群消息行 → Message 形状（friendUid=@g.us,真机确认 2026-08-20）。
+
+    与 WA 1:1 行同构,差异:``chatRoomTalker`` = 群内发送者(@lid)、friendUid =
+    群 id(@g.us);keyId 通常有值(与微信群 msgId=null 不同),缺失时用
+    owner:room:createTime:chatRoomTalker 合成稳定主键。
+    """
+    from app.scraper.whatsapp import _normalize_whatsapp_message
+
+    rec = _normalize_whatsapp_message(record)
+    rec["room_id"] = record.get("friendUid")
+    rec["talker_id"] = record.get("chatRoomTalker")
+    # 群行的 file 常带直连 OSS URL(yk-whatsapp.oss),优先于 1:1 用的 mediaName 不透明名
+    rec["file_url"] = record.get("file") or record.get("mediaName")
+    if not rec.get("msg_id"):
+        rec["msg_id"] = (
+            f"{record.get('uid')}:{record.get('friendUid')}:"
+            f"{record.get('createTime')}:{record.get('chatRoomTalker')}"
+        )
+    return rec
+
+
+async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500):
+    """WA 群消息:浏览器 fetch chatRecords,friendUid 直接传 @g.us 群 id
+    (真机确认 2026-08-20,与微信群的 msg/chats+roomId 同套路)。
+
+    群来自本地 chat_group(channel='whatsapp');翻页复用 paginate_chats
+    (hasLast/end 游标);守护循环同微信群抓取器(毒群重启跳过)。
+    """
+    import time as _time
+
+    from app.db.base import async_engine, async_session_maker
+    from app.scraper.whatsapp import _is_wa_divider
+    from app.storage.repos import upsert_messages
+    from sqlalchemy import text as sa_text
+
+    async with async_engine.connect() as conn:
+        result = await conn.execute(sa_text(
+            "SELECT owner_account_id, room_id FROM chat_group "
+            "WHERE channel='whatsapp' AND owner_account_id IS NOT NULL"
+        ))
+        groups = [(row[0], row[1]) for row in result.fetchall()]
+
+    logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}")
+
+    total_fetched = 0
+    total_upserted = 0
+    total_pages = 0
+    scan = groups[:max_groups]
+    idx = 0
+    relaunches = 0
+
+    while idx < len(scan):
+        try:
+            async with PageScraper(cookie_str) as scraper:
+                page = await _open_fetch_page(scraper)
+
+                while idx < len(scan):
+                    owner_id, room_id = scan[idx]
+                    i = idx
+                    records = []
+                    pages = 0
+                    try:
+                        now_ms = int(_time.time() * 1000)
+
+                        async def _fetch(end_ms: int, _owner=owner_id, _room=room_id):
+                            return await _browser_fetch(
+                                page,
+                                "/qqweb/WhatsAppUser/chatRecords",
+                                json_body={"uid": _owner, "friendUid": _room, "start": 0, "end": end_ms},
+                            )
+
+                        async for data in paginate_chats(_fetch, now_ms, max_pages=200):
+                            pages += 1
+                            rows = data.get("messages") or []
+                            for row in rows:
+                                if not isinstance(row, dict) or _is_wa_divider(row):
+                                    continue
+                                records.append(_normalize_wa_group_message(row))
+
+                    except Exception as e:
+                        logger.warning(f"  [{i+1}] {str(owner_id)[:18]}:{str(room_id)[:18]} failed: {e}")
+                        for attempt in range(30):
+                            try:
+                                try:
+                                    await page.close()
+                                except Exception:
+                                    pass
+                                page = await _open_fetch_page(scraper)
+                                break
+                            except Exception as rec:
+                                logger.warning(f"  recovery {attempt + 1}/30 failed: {rec}")
+                                if attempt == 29:
+                                    raise
+                                await asyncio.sleep(120)
+                        idx += 1
+                        continue
+
+                    total_fetched += len(records)
+                    total_pages += pages
+
+                    if records:
+                        async with async_session_maker() as session:
+                            await upsert_messages(session, records)
+                            await session.commit()
+                        total_upserted += len(records)
+
+                    if (i + 1) % 10 == 0:
+                        logger.info(f"  [{i+1}/{len(scan)}] msgs total: {total_fetched}, pages: {total_pages}")
+
+                    await asyncio.sleep(2 + (i % 3))
+                    idx += 1
+
+        except Exception as fatal:
+            relaunches += 1
+            if relaunches > 10:
+                raise
+            logger.warning(f"  browser-level failure, relaunch {relaunches}/10, skip poison group: {fatal}")
+            idx += 1
+            await asyncio.sleep(30)
+
+    logger.info(f"DONE: wa-group-messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}")
+    return total_fetched, total_upserted
+
+
 # ---------------------------------------------------------------------------
 # Media downloads (OSS CDN — NOT API calls, safe from 风控)
 # ---------------------------------------------------------------------------
@@ -1439,6 +1564,7 @@ async def main():
             "voice-calls",   # 语音通话记录 (voice-call-list + wavoicecalllist)
             "voice-clips",   # 长语音 (browser fetch call/list per employee)
             "wa-messages",   # WA消息 (browser fetch chatRecords per conversation)
+            "wa-group-messages", # WA群消息 (chatRecords with friendUid=@g.us)
             "media",         # OSS媒体下载 (录音/长语音音频/消息图片)
         ],
         help="抓取实体",
@@ -1482,6 +1608,8 @@ async def main():
         await scrape_voice_clips(cookie, args.max_employees)
     elif args.entity == "wa-messages":
         await scrape_wa_messages(cookie, args.max_conversations)
+    elif args.entity == "wa-group-messages":
+        await scrape_wa_group_messages(cookie, args.max_conversations)
     elif args.entity == "media":
         await scrape_media(cookie, args.batch_size)
 

yunke-server/tests/test_scrape_wa_group_messages.py

@@ -0,0 +1,50 @@
+"""Unit tests for the WA GROUP message normalizer in :mod:`scripts.scrape_via_page`
+(chatRecords with friendUid=@g.us, confirmed live 2026-08-20). Group rows carry
+chatRoomTalker = in-group sender id and a real keyId (unlike WeChat groups)."""
+
+from __future__ import annotations
+
+from scripts.scrape_via_page import _normalize_wa_group_message
+
+OWNER = "15638182659@s.whatsapp.net"
+ROOM = "120363406490654152@g.us"
+
+
+def _row(**over):
+    base = {
+        "createTime": 1773025474000,
+        "type": 1,
+        "content": "hello",
+        "isSend": 0,
+        "keyId": "3EB0CC278FB9C73304A622",
+        "uid": OWNER,
+        "friendUid": ROOM,
+        "chatRoomTalker": "267074588913817@lid",
+        "nickName": "A**a",
+    }
+    base.update(over)
+    return base
+
+
+def test_group_row_maps_sender_and_room():
+    rec = _normalize_wa_group_message(_row())
+    assert rec["channel"] == "whatsapp"
+    assert rec["owner_account_id"] == OWNER
+    assert rec["room_id"] == ROOM
+    assert rec["talker_id"] == "267074588913817@lid"  # 群内发送者(chatRoomTalker)
+    assert rec["msg_id"] == "3EB0CC278FB9C73304A622"  # keyId 有值,直接用
+    assert rec["sender_name"] == "A**a"
+
+
+def test_keyid_missing_falls_back_to_stable_synthetic():
+    rec = _normalize_wa_group_message(_row(keyId=None))
+    expect = f"{OWNER}:{ROOM}:1773025474000:267074588913817@lid"
+    assert rec["msg_id"] == expect
+    again = _normalize_wa_group_message(_row(keyId=None))
+    assert rec["msg_id"] == again["msg_id"]  # 重扫幂等
+
+
+def test_media_url_preserved():
+    url = "https://yk-whatsapp.oss-cn-beijing.aliyuncs.com/app-file/x.jpg"
+    rec = _normalize_wa_group_message(_row(type=1, file=url))
+    assert rec["file_url"] == url