diff --git a/yunke-server/app/db/base.py b/yunke-server/app/db/base.py index 5cbab4f..45a15ee 100644 --- a/yunke-server/app/db/base.py +++ b/yunke-server/app/db/base.py @@ -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, ) diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 2f2f285..ad15148 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -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": diff --git a/yunke-server/tests/test_scrape_group_messages.py b/yunke-server/tests/test_scrape_group_messages.py new file mode 100644 index 0000000..b7a30ff --- /dev/null +++ b/yunke-server/tests/test_scrape_group_messages.py @@ -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