feat(scraper): 微信消息翻页取完整历史(paginate_chats hasLast/end游标+200页上限,移除临时skip;全量回填3478对话+95361条至2025-10)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-18 09:15:04 +08:00
parent aeb8a89b03
commit f9ed7dd821
2 changed files with 180 additions and 66 deletions
@@ -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


@@ -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