From f9ed7dd8218a6bc77a973208b881bd34e761e40b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 18 Aug 2026 09:15:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(scraper):=20=E5=BE=AE=E4=BF=A1=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E7=BF=BB=E9=A1=B5=E5=8F=96=E5=AE=8C=E6=95=B4=E5=8E=86?= =?UTF-8?q?=E5=8F=B2(paginate=5Fchats=20hasLast/end=E6=B8=B8=E6=A0=87+200?= =?UTF-8?q?=E9=A1=B5=E4=B8=8A=E9=99=90,=E7=A7=BB=E9=99=A4=E4=B8=B4?= =?UTF-8?q?=E6=97=B6skip;=E5=85=A8=E9=87=8F=E5=9B=9E=E5=A1=AB3478=E5=AF=B9?= =?UTF-8?q?=E8=AF=9D+95361=E6=9D=A1=E8=87=B32025-10)?= 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 | 157 ++++++++++-------- .../tests/test_scrape_via_page_pagination.py | 89 ++++++++++ 2 files changed, 180 insertions(+), 66 deletions(-) create mode 100644 yunke-server/tests/test_scrape_via_page_pagination.py diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index d663450..c94e275 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -362,11 +362,40 @@ 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). +async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=None): + """按 hasLast/end 游标翻页 chats 类接口（微信 msg/chats 与 WA chatRecords 同协议）。 - Navigates to each conversation's chat window. The SPA calls /wechatweb/msg/chats - to load messages — we intercept that response. + fetch_page(end_ms) -> 响应 dict；逐页 yield data 部分（dict）。 + 终止：data 非 dict / hasLast 假 / end 非数值 / 达到 max_pages 页上限（防死循环）。 + """ + if sleep is None: + sleep = asyncio.sleep + pages = 0 + while True: + body = await fetch_page(end_ms) + pages += 1 + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return + yield data + if not data.get("hasLast"): + return + end = data.get("end") + if not isinstance(end, (int, float)): + return + end_ms = int(end) - 1 + if pages >= max_pages: + logger.warning(f"paginate_chats: page cap {max_pages} hit, stop paging") + return + await sleep(1) + + +async def scrape_messages(cookie_str: str, max_conversations: int = 100): + """Scrape wechat messages via browser fetch POST /wechatweb/msg/chats. + + Conversation pairs come from the local customer table (channel='wechat'). + Pages backward through full history via data.end / data.hasLast + (paginate_chats, same cursor strategy as the WA chatRecords scraper). """ from app.db.base import async_session_maker, async_engine from sqlalchemy import text as sa_text @@ -386,6 +415,7 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): total_fetched = 0 total_upserted = 0 + total_pages = 0 async with PageScraper(cookie_str) as scraper: # Just load the base URL to establish cookie context — no need for specific page @@ -397,39 +427,68 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): page = fetch_page # use the stable page for all evaluate() calls for i, (owner_id, friend_id) in enumerate(conversations[:max_conversations]): + records = [] + pages = 0 try: # Use browser fetch() to call the message API from the page context # (the browser sends it with proper cookies/headers → no 风控) # NOTE: msg/chats uses FORM-ENCODED + cursor params (start/end, not pageNum) import time as _time now_ms = int(_time.time() * 1000) - body = await page.evaluate("""async ([ownerId, friendId, endMs]) => { - const params = new URLSearchParams(); - params.append('wechatId', ownerId); - params.append('friendId', friendId); - params.append('types', ''); - params.append('start', '0'); - params.append('end', String(endMs)); - const resp = await fetch('/wechatweb/msg/chats', { - method: 'POST', - headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, - body: params.toString(), - }); - return await resp.json(); - }""", [owner_id, friend_id, now_ms]) - responses = [{"body": body, "status": 200}] + async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id): + return await page.evaluate("""async ([ownerId, friendId, endMs]) => { + const params = new URLSearchParams(); + params.append('wechatId', ownerId); + params.append('friendId', friendId); + params.append('types', ''); + params.append('start', '0'); + params.append('end', String(endMs)); + const resp = await fetch('/wechatweb/msg/chats', { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: params.toString(), + }); + return await resp.json(); + }""", [_owner, _friend, end_ms]) - # Debug: log first response structure - if i == 0: - logger.info(f" DEBUG first response keys: {list(body.keys()) if isinstance(body, dict) else type(body)}") - if isinstance(body, dict): - d = body.get("data") - logger.info(f" DEBUG data type: {type(d)}, keys: {list(d.keys()) if isinstance(d, dict) else 'N/A'}") - if isinstance(d, dict): - msgs = d.get("messages") - logger.info(f" DEBUG messages type: {type(msgs)}, len: {len(msgs) if isinstance(msgs, list) else 'N/A'}") - logger.info(f" DEBUG body preview: {str(body)[:500]}") + # 翻页取完整历史：hasLast 真 → 用 data.end-1 继续向前翻（页间 sleep 1s） + async for data in paginate_chats(_fetch, now_ms): + pages += 1 + # Debug: log first response structure + if i == 0 and pages == 1: + logger.info(f" DEBUG first response keys: {list(data.keys())}") + msgs0 = data.get("messages") + logger.info(f" DEBUG messages type: {type(msgs0)}, len: {len(msgs0) if isinstance(msgs0, list) else 'N/A'}") + rows = data.get("messages") or [] + for row in rows: + if not isinstance(row, dict): + continue + # 跳过 type=-2 日期分隔线（非真实消息，仅显示日期） + if str(row.get("type")) == "-2": + 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, + # 用发送的 friend_id（与客户表一致）优先，API 返回的 talker 可能脱敏 + "talker_id": friend_id or row.get("talker") or row.get("oriTalker"), + "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, + }) except Exception as e: logger.warning(f" [{i+1}] {owner_id[:12]}:{friend_id[:12]} failed: {e}") @@ -444,42 +503,8 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): page = fetch_page continue - # Extract message records (msg/chats response: body.data.messages[]) - records = [] - for resp in responses: - body = resp.get("body", {}) - data = body.get("data") or {} - rows = data.get("messages") or [] - for row in rows: - if not isinstance(row, dict): - continue - # 跳过 type=-2 日期分隔线（非真实消息，仅显示日期） - if str(row.get("type")) == "-2": - 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, - # 用发送的 friend_id（与客户表一致）优先，API 返回的 talker 可能脱敏 - "talker_id": friend_id or row.get("talker") or row.get("oriTalker"), - "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) + total_pages += pages if records: async with async_session_maker() as session: @@ -488,12 +513,12 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): total_upserted += len(records) if (i + 1) % 10 == 0: - logger.info(f" [{i+1}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}") + logger.info(f" [{i+1}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}, pages: {total_pages}") # Human-like delay await asyncio.sleep(2 + (i % 3)) - logger.info(f"DONE: messages fetched={total_fetched}, upserted={total_upserted}") + logger.info(f"DONE: messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}") return total_fetched, total_upserted diff --git a/yunke-server/tests/test_scrape_via_page_pagination.py b/yunke-server/tests/test_scrape_via_page_pagination.py new file mode 100644 index 0000000..4a41dc1 --- /dev/null +++ b/yunke-server/tests/test_scrape_via_page_pagination.py @@ -0,0 +1,89 @@ +"""Unit tests for the hasLast/end cursor pagination helper in +:mod:`scripts.scrape_via_page` (WeChat msg/chats deep-history paging). + +No network/browser — the fetch callable is faked; these verify the +cursor protocol: page N+1 uses ``data.end - 1`` as ``end`` until +``hasLast`` is falsy, with guards for malformed ``end`` and a page cap. +""" + +from __future__ import annotations + +from scripts.scrape_via_page import paginate_chats + + +def _page(messages, has_last, end=1000): + return {"data": {"messages": messages, "hasLast": has_last, "end": end}} + + +async def _noop_sleep(_seconds: float) -> None: + return None + + +async def test_multi_page_uses_end_minus_one_cursor(): + calls: list[int] = [] + + async def fetch(end_ms: int): + calls.append(end_ms) + if len(calls) == 1: + return _page([{"id": 1}], has_last=True, end=1000) + return _page([{"id": 2}], has_last=False, end=500) + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)] + + assert [len(d["messages"]) for d in datas] == [1, 1] + assert calls == [2000, 999] # 下一页游标 = end - 1 + + +async def test_stops_when_has_last_falsy(): + async def fetch(end_ms: int): + return _page([{"id": 1}], has_last=False) + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)] + + assert len(datas) == 1 + + +async def test_stops_on_non_dict_data(): + calls = [] + + async def fetch(end_ms: int): + calls.append(end_ms) + return {"message": "调用成功", "data": None} + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)] + + assert datas == [] + assert len(calls) == 1 + + +async def test_stops_when_end_not_numeric(): + async def fetch(end_ms: int): + return _page([{"id": 1}], has_last=True, end=None) + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)] + + assert len(datas) == 1 # hasLast 真但游标缺失 → 不再无限请求 + + +async def test_page_cap_prevents_infinite_loop(): + calls = [] + + async def fetch(end_ms: int): + calls.append(end_ms) + return _page([{"id": len(calls)}], has_last=True, end=end_ms - 1) + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, max_pages=5, sleep=_noop_sleep)] + + assert len(datas) == 5 + assert len(calls) == 5 + + +async def test_empty_messages_list_still_pages(): + async def fetch(end_ms: int): + if end_ms > 1500: + return _page(None, has_last=True, end=1500) + return _page([{"id": 1}], has_last=False) + + datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)] + + assert len(datas) == 2