From 02dd77a62f1baa80fa8c7c08db56f638ea3e939c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 20 Aug 2026 09:21:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(scraper):=20WA=E7=BE=A4=E6=8A=93=E5=8F=96(?= =?UTF-8?q?type=3D1=E7=9C=9F=E6=9C=BA=E9=AA=8C=E8=AF=81,223=E7=BE=A4)+?= =?UTF-8?q?=E7=BE=A4=E6=88=90=E5=91=98=E6=B4=BE=E7=94=9F(12840=E6=88=90?= =?UTF-8?q?=E5=91=98/750=E7=BE=A4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .../scripts/backfill_group_members.py | 63 +++++++++++++++ yunke-server/scripts/full_sync_all.sh | 11 ++- yunke-server/scripts/scrape_via_page.py | 81 +++++++++++++++++++ 3 files changed, 153 insertions(+), 2 deletions(-) create mode 100644 yunke-server/scripts/backfill_group_members.py diff --git a/yunke-server/scripts/backfill_group_members.py b/yunke-server/scripts/backfill_group_members.py new file mode 100644 index 0000000..6852f5e --- /dev/null +++ b/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()) diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index 53907a3..841626e 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/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 diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index ad15148..791e0eb 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/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":