diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index 841626e..f7af4d2 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/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 diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index bde5efb..7efb22d 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/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) diff --git a/yunke-server/tests/test_scrape_wa_group_messages.py b/yunke-server/tests/test_scrape_wa_group_messages.py new file mode 100644 index 0000000..0f00009 --- /dev/null +++ b/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