From c6217ab1a9a8598cdcfe929fc2e2ff84e5daf080 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 14 Aug 2026 15:04:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(scraper):=20=E9=A1=B5=E9=9D=A2=E6=8A=93?= =?UTF-8?q?=E5=8F=96=E5=99=A8=E6=94=AF=E6=8C=81=E6=B6=88=E6=81=AF(?= =?UTF-8?q?=E5=AF=BC=E8=88=AA=E8=81=8A=E5=A4=A9=E7=AA=97=E5=8F=A3=E6=8B=A6?= =?UTF-8?q?=E6=88=AAmsg/chats=20API)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude --- yunke-server/scripts/scrape_via_page.py | 105 +++++++++++++++++++++++- 1 file changed, 104 insertions(+), 1 deletion(-) diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 7636b73..218faae 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -36,6 +36,7 @@ INTERCEPT_PATHS = { "call_list": "/wechatweb/callRecord/queryWeChatVoiceList", # voice calls "wa_friend": "/qqweb/WhatsAppUser/businessFriendList", # WA friends + groups "wa_user": "/qqweb/WhatsAppUser/businessSalesList", # WA accounts + "msg_chats": "/wechatweb/msg/chats", # chat messages } @@ -354,10 +355,110 @@ async def scrape_groups(cookie_str: str, max_employees: int = 100): return total_fetched, total_upserted +async def scrape_messages(cookie_str: str, max_conversations: int = 100): + """Scrape wechat messages via page navigation (chat window). + + Navigates to each conversation's chat window. The SPA calls /wechatweb/msg/chats + to load messages — we intercept that response. + """ + from app.db.base import async_session_maker, async_engine + from sqlalchemy import text as sa_text + from app.storage.repos import upsert_messages + + # Discover conversations from local customer table + async with async_engine.connect() as conn: + result = await conn.execute(sa_text( + "SELECT DISTINCT owner_account_id, friend_id FROM customer " + "WHERE channel='wechat' AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL " + "ORDER BY owner_account_id, friend_id" + )) + conversations = [(row[0], row[1]) for row in result.fetchall()] + + logger.info(f"Discovered {len(conversations)} conversations, will scrape {min(max_conversations, len(conversations))}") + + total_fetched = 0 + total_upserted = 0 + + async with PageScraper(cookie_str) as scraper: + for i, (owner_id, friend_id) in enumerate(conversations[:max_conversations]): + # Chat window URL: opens this specific conversation + chat_url = f"{YUNKE_BASE}/cms/wechat/salewechat/message/wechat-window/1/{owner_id}/{friend_id}" + + # Intercept the message API response + scraper.start_capture(INTERCEPT_PATHS["msg_chats"]) + + try: + await scraper.navigate_and_wait(chat_url, wait_selector=".yk-chat", timeout=10000) + await scraper.page.wait_for_timeout(3000) # let the message API complete + except Exception as e: + logger.warning(f" [{i+1}] {owner_id[:12]}:{friend_id[:12]} nav failed: {e}") + continue + + responses = scraper.get_captured() + + # Extract message records + records = [] + for resp in responses: + body = resp.get("body", {}) + data = body.get("data", {}) + rows = data.get("data", []) or [] + for row in rows: + if not isinstance(row, dict): + continue + # Normalize to Message shape + ts_raw = row.get("createTime") or row.get("timestamp") + try: + ts = int(str(ts_raw)) if ts_raw else None + except (ValueError, TypeError): + ts = None + records.append({ + "msg_id": str(row.get("msgId") or row.get("msgSvrId") or row.get("id") or f"{owner_id}:{friend_id}:{i}:{len(records)}"), + "channel": "wechat", + "owner_account_id": owner_id, + "talker_id": row.get("talker") or row.get("oriTalker") or friend_id, + "room_id": None, + "is_send": bool(row.get("isSend", 0)) if row.get("isSend") is not None else None, + "type": _normalize_msg_type(row.get("type")), + "type_raw": row.get("type"), + "text": row.get("content") or row.get("text") or "", + "file_url": row.get("file") or row.get("fileUrl"), + "timestamp": ts, + "sender_name": row.get("name") or row.get("senderName"), + "raw_json": row, + }) + + total_fetched += len(records) + + 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}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}") + + # Human-like delay + await asyncio.sleep(2 + (i % 3)) + + logger.info(f"DONE: messages fetched={total_fetched}, upserted={total_upserted}") + return total_fetched, total_upserted + + +def _normalize_msg_type(raw_type) -> str | None: + """Normalize a raw message type to our canonical type string.""" + if raw_type is None: + return None + t = str(raw_type) + mapping = {"1": "text", "3": "image_legacy", "34": "voice", "43": "video", "47": "emoji", "49": "file"} + return mapping.get(t, "unknown") + + async def main(): parser = argparse.ArgumentParser(description="页面浏览器抓取器") - parser.add_argument("--entity", required=True, choices=["friends", "groups"], help="抓取实体") + parser.add_argument("--entity", required=True, choices=["friends", "groups", "messages"], help="抓取实体") parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工") + parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话") args = parser.parse_args() cookie = os.environ.get("YK_COOKIE", "") @@ -375,6 +476,8 @@ async def main(): await scrape_friends(cookie, args.max_employees) elif args.entity == "groups": await scrape_groups(cookie, args.max_employees) + elif args.entity == "messages": + await scrape_messages(cookie, args.max_conversations) if __name__ == "__main__":