feat(scraper): 微信群消息抓取(368778条/750群)+浏览器守护循环+SQLite忙等

- 新 group-messages 实体:真机探明群消息=同一 msg/chats 端点 friendId 传 roomId; talker=发送者wxid、msgId=null→owner:room:createTime:talker 合成稳定主键(重扫幂等) - 浏览器守护循环:超大群撑崩渲染进程时重启浏览器跳过毒群继续(实战2次自愈); 200页上限;恢复路径补 await - base.py SQLite connect_args timeout=60:并发写忙等而非撞锁报错 (双抓取器并发写实测 database is locked 崩任务) - 5个新单测,全量410绿 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-20 09:09:32 +08:00
parent 123c23242c
commit f92aac1a3d
3 changed files with 228 additions and 2 deletions
@@ -23,10 +23,15 @@ class Base(DeclarativeBase):

 settings = get_settings()

+# SQLite: 写锁竞争时忙等 60s 而不是立刻报 "database is locked"
+# （多进程并发写的场景：页面抓取器 × 2 + 后端 API 同时跑，2026-08-19 实测撞锁崩任务）。
+_connect_args = {"timeout": 60} if settings.db_url.startswith("sqlite") else {}
+
 async_engine = create_async_engine(
    settings.db_url,
    pool_pre_ping=True,
    pool_recycle=3600,  # recycle connections before MySQL wait_timeout (default 8h) closes them
+    connect_args=_connect_args,
    future=True,
 )

@@ -565,7 +565,7 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
                for attempt in range(30):
                    try:
                        try:
-                            fetch_page.close()
+                            await fetch_page.close()
                        except Exception:
                            pass
                        fetch_page = await scraper.context.new_page()
@@ -599,6 +599,166 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
    return total_fetched, total_upserted


+def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | None:
+    """msg/chats 群消息行 → Message 形状（friendId=roomId 时端点返回的行）。
+
+    群行与 1:1 行同构（真机确认 2026-08-19），差异：talker=发送者 wxid、
+    roomid=群 id、msgId 为 null → 用 owner:room:createTime:talker 合成稳定
+    主键，重扫同一条消息得到同一 id（幂等 upsert 前提）。type=-2 日期分隔
+    线丢弃。
+    """
+    if str(row.get("type")) == "-2":
+        return None
+    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
+    talker = row.get("talker") or ""
+    synth = f"{owner_id}:{room_id}:{ts}:{talker}"
+    return {
+        "msg_id": str(row.get("msgId") or row.get("msgSvrId") or row.get("id") or synth),
+        "channel": "wechat",
+        "owner_account_id": owner_id,
+        "talker_id": talker or None,
+        "room_id": row.get("roomid") or room_id,
+        "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,
+    }
+
+
+async def scrape_group_messages(cookie_str: str, max_groups: int = 500):
+    """Scrape WeChat GROUP chat messages via browser fetch POST /wechatweb/msg/chats.
+
+    Same endpoint as 1:1 chats — pass the room id as ``friendId`` (confirmed
+    live 2026-08-19: rows come back with talker=sender wxid, roomid=the room,
+    msgId=null → stable synthetic pk, see _normalize_group_message). Groups
+    come from the local chat_group table; full-history pagination via
+    paginate_chats (hasLast/end cursor, same as 1:1).
+    """
+    import time as _time
+
+    from app.db.base import async_engine, async_session_maker
+    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='wechat' AND owner_account_id IS NOT NULL "
+            "ORDER BY last_chat_time DESC"
+        ))
+        groups = [(row[0], row[1]) for row in result.fetchall()]
+
+    logger.info(f"Discovered {len(groups)} 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
+
+    # 守护循环：超大群的单页响应可能把渲染进程/整个浏览器撑崩（真机 Target
+    # crashed, 2026-08-19）——浏览器级失败时重启浏览器、跳过毒群、从当前
+    # 位置继续；页面级失败换页重试（30 × 2min，同 scrape_messages）。
+    while idx < len(scan):
+        try:
+            async with PageScraper(cookie_str) as scraper:
+                fetch_page = await scraper.context.new_page()
+                await fetch_page.goto(f"{YUNKE_BASE}/", wait_until="domcontentloaded", timeout=15000)
+                await fetch_page.wait_for_timeout(2000)
+                page = fetch_page
+
+                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 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, _room, 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):
+                                    continue
+                                rec = _normalize_group_message(row, owner_id, room_id)
+                                if rec is not None:
+                                    records.append(rec)
+
+                    except Exception as e:
+                        logger.warning(f"  [{i+1}] {owner_id[:12]}:{str(room_id)[:16]} failed: {e}")
+                        # 页面级恢复：换新页继续；浏览器已死则抛给外层重启
+                        for attempt in range(30):
+                            try:
+                                try:
+                                    await fetch_page.close()
+                                except Exception:
+                                    pass
+                                fetch_page = await scraper.context.new_page()
+                                await fetch_page.goto(f"{YUNKE_BASE}/cms/home/statistics", wait_until="networkidle", timeout=20000)
+                                await fetch_page.wait_for_timeout(1000)
+                                page = fetch_page
+                                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: group-messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}")
+    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:
@@ -1135,6 +1295,7 @@ async def main():
            "friends",       # 微信客户 (per-employee friend pages)
            "groups",        # 微信群 (per-employee group pages)
            "messages",      # 微信消息 (browser fetch msg/chats)
+            "group-messages", # 微信群消息 (msg/chats with friendId=roomId)
            "accounts",      # 微信员工账号 (wechatlist page interception)
            "wa-accounts",   # WA员工账号 (waBusiness/userlist interception)
            "wa-friends",    # WA客户 (waBusiness/friendlist interception)
@@ -1146,7 +1307,7 @@ async def main():
        help="抓取实体",
    )
    parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工 (friends/groups/voice-clips)")
-    parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话 (messages/wa-messages)")
+    parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话 (messages/wa-messages/group-messages)")
    parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls)")
    parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)")
    args = parser.parse_args()
@@ -1168,6 +1329,8 @@ async def main():
        await scrape_groups(cookie, args.max_employees)
    elif args.entity == "messages":
        await scrape_messages(cookie, args.max_conversations)
+    elif args.entity == "group-messages":
+        await scrape_group_messages(cookie, args.max_conversations)
    elif args.entity == "accounts":
        await scrape_accounts(cookie, args.max_pages)
    elif args.entity == "wa-accounts":
@@ -0,0 +1,58 @@
+"""Unit tests for the WeChat GROUP message normalizer in
+:mod:`scripts.scrape_via_page` (msg/chats with friendId=roomId, confirmed
+live 2026-08-19). Group rows differ from 1:1: talker=sender, roomid=the room,
+msgId=null -> stable synthetic pk required for idempotent re-scrapes."""
+
+from __future__ import annotations
+
+from scripts.scrape_via_page import _normalize_group_message
+
+OWNER = "wxid_wo97u1u2x11w22"
+ROOM = "56807599995@chatroom"
+
+
+def _row(**over):
+    base = {
+        "talker": "wxid_en2iklxtjxzz22",
+        "oriTalker": ROOM,
+        "name": "小**贸",
+        "content": "hello",
+        "type": 1,
+        "roomid": ROOM,
+        "createTime": 1787125769000,
+        "msgId": None,
+    }
+    base.update(over)
+    return base
+
+
+def test_group_row_maps_sender_and_room():
+    rec = _normalize_group_message(_row(), OWNER, ROOM)
+    assert rec["channel"] == "wechat"
+    assert rec["owner_account_id"] == OWNER
+    assert rec["room_id"] == ROOM
+    assert rec["talker_id"] == "wxid_en2iklxtjxzz22"  # 发送者（非会话对端）
+    assert rec["sender_name"] == "小**贸"
+    assert rec["type_raw"] == 1
+
+
+def test_synthetic_msg_id_stable_across_rescans():
+    """msgId=null 的群行两次归一化必须得到同一 msg_id（幂等 upsert 前提）。"""
+    a = _normalize_group_message(_row(), OWNER, ROOM)
+    b = _normalize_group_message(_row(), OWNER, ROOM)
+    assert a["msg_id"] == b["msg_id"]
+    assert a["msg_id"] == f"{OWNER}:{ROOM}:1787125769000:wxid_en2iklxtjxzz22"
+
+
+def test_real_msg_id_preferred_over_synthetic():
+    rec = _normalize_group_message(_row(msgId="m999"), OWNER, ROOM)
+    assert rec["msg_id"] == "m999"
+
+
+def test_divider_rows_dropped():
+    assert _normalize_group_message(_row(type=-2, text="2026-08-19 15:49"), OWNER, ROOM) is None
+
+
+def test_roomid_falls_back_to_queried_room():
+    rec = _normalize_group_message(_row(roomid=""), OWNER, ROOM)
+    assert rec["room_id"] == ROOM