From 9bca5fc08a1d345682c7b3ca7771e3783945297c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 21 Aug 2026 12:20:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(sync):=20=E6=B6=88=E6=81=AF=E5=85=A8?= =?UTF-8?q?=E9=9D=A2=E5=A2=9E=E9=87=8F=E5=90=8C=E6=AD=A5(=E6=B0=B4?= =?UTF-8?q?=E4=BD=8D+=E8=B7=B3=E8=BF=87+=E8=A7=A6=E5=BA=95=E5=81=9C,4h?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=E4=BB=8E=E5=B0=8F=E6=97=B6=E7=BA=A7=E9=99=8D?= =?UTF-8?q?=E5=88=B0=E5=88=86=E9=92=9F=E7=BA=A7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每对话水位=本地max(timestamp);请求带start=水位只拉新消息 - 云客last_chat_time≤水位的对话整跳过(0请求)——实测7046对话3150被跳过(45%) - 翻页遇≤水位消息即停(已同步区),页数从几十页降到1页 - 微信1:1/WA 1:1/微信群/WA群四个抓取器全接入;CLI --full-history强制全量 - 实测:每对话1页(~3s),3895待访对话约3.5h(含补积压的首次全量); 稳态后每4h只访有新活动的对话(几十~几百个,15-30min) Co-Authored-By: Claude --- yunke-server/scripts/scrape_via_page.py | 143 +++++++++++++++++++----- 1 file changed, 116 insertions(+), 27 deletions(-) diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 0d9bfad..a8835dd 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -571,13 +571,47 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl return fresh + known -async def scrape_messages(cookie_str: str, max_conversations: int = 100): +async def _conversation_watermarks(channel: str) -> dict[tuple[str, str], int]: + """每对话的本地最新消息时间(ms)——增量同步水位。""" + from app.db.base import async_engine + from sqlalchemy import text as sa_text + + async with async_engine.connect() as conn: + result = await conn.execute(sa_text( + "SELECT owner_account_id, talker_id, max(timestamp) FROM message " + "WHERE channel=:ch AND (room_id IS NULL OR room_id='') " + "AND talker_id IS NOT NULL AND talker_id != '' " + "GROUP BY owner_account_id, talker_id" + ), {"ch": channel}) + return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()} + + +async def _customer_activity(channel: str) -> dict[tuple[str, str], int]: + """客户表 last_chat_time(4h 增量的 friends 实时刷新)——判断对话有无新消息。""" + from app.db.base import async_engine + from sqlalchemy import text as sa_text + + async with async_engine.connect() as conn: + result = await conn.execute(sa_text( + "SELECT owner_account_id, friend_id, last_chat_time FROM customer " + "WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL" + ), {"ch": channel}) + return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()} + + +def _page_min_ts(data: dict) -> int: + """页内最早消息时间(0 当无有效 ts)。""" + ts_list = [m.get("timestamp") for m in (data.get("messages") or []) if isinstance(m, dict) and m.get("timestamp")] + return min(ts_list) if ts_list else 0 + + +async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False): """Scrape wechat messages via browser fetch POST /wechatweb/msg/chats. - Conversation pairs come from _discover_conversations (customer table ∪ - voice-call pairs). Pages backward through full history via data.end / - data.hasLast (paginate_chats, same cursor strategy as the WA chatRecords - scraper). + **增量同步(默认)**:每对话水位 = 本地 max(timestamp);请求带 + ``start=水位`` 只拉新消息,翻页遇到 ≤水位的消息即停(已同步区)。 + 客户表 last_chat_time(friends 实时刷新)≤ 水位的对话**整跳过**(0 请求) + ——活跃对话才访问。``full_history=True`` 强制全量(首次/修复用)。 """ from sqlalchemy import text as sa_text @@ -585,8 +619,25 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): from app.storage.repos import upsert_messages conversations = await _discover_conversations("wechat", "voice_call") + watermarks = await _conversation_watermarks("wechat") + cust_activity = await _customer_activity("wechat") - logger.info(f"Discovered {len(conversations)} conversations, will scrape {min(max_conversations, len(conversations))}") + # 增量模式:云客侧无新活动的对话跳过(0 请求)——速度来源 + skipped = 0 + if not full_history: + kept = [] + for pair in conversations: + wm = watermarks.get(pair, 0) + act = cust_activity.get(pair, 0) + if wm > 0 and act > 0 and act <= wm + 60_000: + skipped += 1 + else: + kept.append(pair) + conversations = kept + logger.info( + f"Discovered conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)" + + (" [FULL HISTORY]" if full_history else " [incremental]") + ) total_fetched = 0 total_upserted = 0 @@ -611,13 +662,15 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): import time as _time now_ms = int(_time.time() * 1000) - async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id): - return await page.evaluate("""async ([ownerId, friendId, endMs]) => { + watermark = 0 if full_history else watermarks.get((owner_id, friend_id), 0) + + async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id, _wm=watermark): + return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => { const params = new URLSearchParams(); params.append('wechatId', ownerId); params.append('friendId', friendId); params.append('types', ''); - params.append('start', '0'); + params.append('start', String(startMs)); params.append('end', String(endMs)); const resp = await fetch('/wechatweb/msg/chats', { method: 'POST', @@ -625,11 +678,14 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): body: params.toString(), }); return await resp.json(); - }""", [_owner, _friend, end_ms]) + }""", [_owner, _friend, _wm, end_ms]) # 翻页取完整历史：hasLast 真 → 用 data.end-1 继续向前翻（页间 sleep 1s） async for data in paginate_chats(_fetch, now_ms): pages += 1 + # 增量触底停:页内出现 ≤水位 的消息 = 已同步区,不再向前翻 + if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark: + break # Debug: log first response structure if i == 0 and pages == 1: logger.info(f" DEBUG first response keys: {list(data.keys())}") @@ -764,7 +820,7 @@ def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | N } -async def scrape_group_messages(cookie_str: str, max_groups: int = 500): +async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False): """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 @@ -787,7 +843,10 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500): )) groups = [(row[0], row[1]) for row in result.fetchall()] - logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}") + room_wms = await _room_watermarks() + + logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}" + + (" [FULL]" if full_history else " [incremental]")) total_fetched = 0 total_upserted = 0 @@ -814,14 +873,15 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500): pages = 0 try: now_ms = int(_time.time() * 1000) + watermark = 0 if full_history else room_wms.get(room_id, 0) - async def _fetch(end_ms: int, _owner=owner_id, _room=room_id): - return await page.evaluate("""async ([ownerId, friendId, endMs]) => { + async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark): + return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => { const params = new URLSearchParams(); params.append('wechatId', ownerId); params.append('friendId', friendId); params.append('types', ''); - params.append('start', '0'); + params.append('start', String(startMs)); params.append('end', String(endMs)); const resp = await fetch('/wechatweb/msg/chats', { method: 'POST', @@ -829,10 +889,12 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500): body: params.toString(), }); return await resp.json(); - }""", [_owner, _room, end_ms]) + }""", [_owner, _room, _wm, end_ms]) async for data in paginate_chats(_fetch, now_ms, max_pages=200): pages += 1 + if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark: + break rows = data.get("messages") or [] for row in rows: if not isinstance(row, dict): @@ -1499,7 +1561,7 @@ async def scrape_voice_clips(cookie_str: str, max_employees: int = 200, page_siz # --------------------------------------------------------------------------- -async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): +async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False): """Scrape WhatsApp chat records via browser fetch POST /qqweb/WhatsAppUser/chatRecords (JSON body {uid, friendUid, start, end}). @@ -1515,8 +1577,24 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): from app.storage.repos import upsert_messages conversations = await _discover_conversations("whatsapp", "wa_voice_call") + watermarks = await _conversation_watermarks("whatsapp") + cust_activity = await _customer_activity("whatsapp") - logger.info(f"Discovered {len(conversations)} WA conversations, will scrape {min(max_conversations, len(conversations))}") + skipped = 0 + if not full_history: + kept = [] + for pair in conversations: + wm = watermarks.get(pair, 0) + act = cust_activity.get(pair, 0) + if wm > 0 and act > 0 and act <= wm + 60_000: + skipped += 1 + else: + kept.append(pair) + conversations = kept + logger.info( + f"Discovered WA conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)" + + (" [FULL]" if full_history else " [incremental]") + ) total_fetched = 0 total_upserted = 0 @@ -1526,13 +1604,14 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): for i, (uid, friend_uid) in enumerate(conversations[:max_conversations]): records = [] + watermark = 0 if full_history else watermarks.get((uid, friend_uid), 0) try: end_ms = int(_time.time() * 1000) while True: body = await _browser_fetch( page, "/qqweb/WhatsAppUser/chatRecords", - json_body={"uid": uid, "friendUid": friend_uid, "start": 0, "end": end_ms}, + json_body={"uid": uid, "friendUid": friend_uid, "start": watermark, "end": end_ms}, ) data = body.get("data") if isinstance(body, dict) else None if not isinstance(data, dict): @@ -1543,6 +1622,9 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): records.append(_normalize_whatsapp_message(row)) if not data.get("hasLast"): break + # 增量触底停:翻到已同步区 + if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark: + break end = data.get("end") if not isinstance(end, (int, float)): break @@ -1614,7 +1696,7 @@ def _normalize_wa_group_message(record: dict) -> dict: return rec -async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500): +async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False): """WA 群消息:浏览器 fetch chatRecords,friendUid 直接传 @g.us 群 id (真机确认 2026-08-20,与微信群的 msg/chats+roomId 同套路)。 @@ -1635,7 +1717,10 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500): )) groups = [(row[0], row[1]) for row in result.fetchall()] - logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}") + room_wms = await _room_watermarks() + + logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}" + + (" [FULL]" if full_history else " [incremental]")) total_fetched = 0 total_upserted = 0 @@ -1656,16 +1741,19 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500): pages = 0 try: now_ms = int(_time.time() * 1000) + watermark = 0 if full_history else room_wms.get(room_id, 0) - async def _fetch(end_ms: int, _owner=owner_id, _room=room_id): + async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark): return await _browser_fetch( page, "/qqweb/WhatsAppUser/chatRecords", - json_body={"uid": _owner, "friendUid": _room, "start": 0, "end": end_ms}, + json_body={"uid": _owner, "friendUid": _room, "start": _wm, "end": end_ms}, ) async for data in paginate_chats(_fetch, now_ms, max_pages=200): pages += 1 + if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark: + break rows = data.get("messages") or [] for row in rows: if not isinstance(row, dict) or _is_wa_divider(row): @@ -1794,6 +1882,7 @@ async def main(): 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)") + parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)") args = parser.parse_args() cookie = os.environ.get("YK_COOKIE", "") @@ -1812,9 +1901,9 @@ async def main(): elif args.entity == "groups": await scrape_groups(cookie, args.max_employees) elif args.entity == "messages": - await scrape_messages(cookie, args.max_conversations) + await scrape_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "group-messages": - await scrape_group_messages(cookie, args.max_conversations) + await scrape_group_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "accounts": await scrape_accounts(cookie, args.max_pages) elif args.entity == "wa-accounts": @@ -1828,9 +1917,9 @@ async def main(): elif args.entity == "voice-clips": await scrape_voice_clips(cookie, args.max_employees) elif args.entity == "wa-messages": - await scrape_wa_messages(cookie, args.max_conversations) + await scrape_wa_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "wa-group-messages": - await scrape_wa_group_messages(cookie, args.max_conversations) + await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "media": await scrape_media(cookie, args.batch_size)