#!/usr/bin/env python """页面浏览器抓取器 — 当 API 接口被风控时，用 Playwright 浏览器方式抓取数据。 原理：SPA 页面加载时会自己调用 friend_list/es 等接口渲染表格。 我们拦截页面的网络响应，获取原始 JSON —— 效果等同 API 调用， 但因为是浏览器页面上下文（非脚本直调），不触发风控。 用法： cd yunke-server DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity friends DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity groups DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity messages DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity accounts DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-accounts DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-friends DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity voice-calls DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity voice-clips DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-messages DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity media Cookie 从 .env 的 YK_COOKIE 读取。 """ from __future__ import annotations import argparse import asyncio import json import logging import os from urllib.parse import urlparse logger = logging.getLogger("scrape_via_page") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") YUNKE_BASE = "https://crm2.yunkecn.com" # API endpoints we intercept (the SPA calls these to render tables) INTERCEPT_PATHS = { "friend_list": "/wechatweb/wxFriends/friend_list/es", # friends + groups (friendTypeYunke) "wxUser_list": "/wechatweb/wxUser/list", # wechat accounts "chatroom": "/wechatweb/chatroom/getChatroomList", # legacy groups "voice_list": "/wechatweb/call/list", # voice clips "call_list": "/wechatweb/callRecord/queryWeChatVoiceList", # voice calls "wa_friend": "/qqweb/WhatsAppUser/businessFriendList", # WA friends + groups "wa_user": "/qqweb/WhatsAppUser/businessSalesList", # WA accounts "msg_chats": "/wechatweb/msg/chats", # chat messages } def parse_cookies(cookie_str: str, domain: str = "crm2.yunkecn.com") -> list[dict]: """Parse a cookie string like 'user=xxx; userToken=yyy' into Playwright cookie dicts.""" cookies = [] for pair in cookie_str.split(";"): pair = pair.strip() if "=" not in pair: continue name, value = pair.split("=", 1) cookies.append({ "name": name.strip(), "value": value.strip(), "domain": domain, "path": "/", }) return cookies class PageScraper: """Navigates 云客 pages in a browser, intercepts API responses, extracts JSON.""" def __init__(self, cookie_str: str, headless: bool = True): self.cookie_str = cookie_str self.headless = headless self.browser = None self.context = None self.page = None self._captured: list[dict] = [] self._capture_path: str | None = None async def __aenter__(self): from playwright.async_api import async_playwright self._pw = await async_playwright().start() self.browser = await self._pw.chromium.launch(headless=self.headless) self.context = await self.browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", viewport={"width": 1920, "height": 1080}, ) # Set cookies cookies = parse_cookies(self.cookie_str) await self.context.add_cookies(cookies) self.page = await self.context.new_page() # Intercept API responses self.page.on("response", self._on_response) return self async def __aexit__(self, *args): if self.browser: await self.browser.close() if self._pw: await self._pw.stop() async def _on_response(self, response): """Capture API JSON responses that match our intercept paths.""" url = response.url path = urlparse(url).path if self._capture_path and path == self._capture_path: try: body = await response.json() self._captured.append({"url": url, "status": response.status, "body": body}) logger.info(f" captured {path}: status={response.status}, keys={list(body.keys()) if isinstance(body, dict) else 'N/A'}") except Exception as e: logger.warning(f" capture failed for {path}: {e}") def start_capture(self, api_path: str): """Start capturing responses for a specific API path.""" self._capture_path = api_path self._captured.clear() def get_captured(self) -> list[dict]: """Return captured responses and stop capturing.""" self._capture_path = None return self._captured async def navigate_and_wait(self, url: str, wait_selector: str = ".ant-table-row", timeout: int = 15000): """Navigate to a URL and wait for table rows to appear.""" await self.page.goto(url, wait_until="domcontentloaded", timeout=timeout) try: await self.page.wait_for_selector(wait_selector, timeout=timeout) except Exception: pass # might be empty page await self.page.wait_for_timeout(2000) # extra settle time async def get_total_pages(self) -> int: """Extract total page count from Ant Design pagination.""" try: total_text = await self.page.evaluate("""() => { const el = document.querySelector('.ant-pagination-total-text'); return el ? el.textContent : ''; }""") if total_text: import re m = re.search(r'(\d+)', total_text) if m: total = int(m.group(1)) page_size = 10 # 云客 default return (total + page_size - 1) // page_size except Exception: pass return 1 async def click_next_page(self) -> bool: """Click the 'next page' button. Returns True if successful.""" try: next_btn = self.page.locator('.ant-pagination-next:not(.ant-pagination-disabled)') if await next_btn.count() > 0: await next_btn.click() await self.page.wait_for_timeout(2000) # wait for data load return True except Exception: pass return False async def scrape_entity_pages( self, list_url: str, api_path: str, max_pages: int = 100, ) -> list[dict]: """Navigate to a list page, paginate through all pages, capturing API responses. Returns a list of captured JSON response bodies (one per page). """ self.start_capture(api_path) await self.navigate_and_wait(list_url) await self.page.wait_for_timeout(1000) # let API call complete all_responses = list(self.get_captured()) total_pages = await self.get_total_pages() logger.info(f" total pages: {min(total_pages, max_pages)}") page_num = 1 while page_num < min(total_pages, max_pages): self.start_capture(api_path) clicked = await self.click_next_page() if not clicked: break await self.page.wait_for_timeout(1500) all_responses.extend(self.get_captured()) page_num += 1 if page_num % 10 == 0: logger.info(f" page {page_num}/{min(total_pages, max_pages)}") logger.info(f" captured {len(all_responses)} API responses across {page_num} pages") return all_responses async def get_employee_links(self) -> list[dict]: """Extract employee wxIds + friend-count links from the staff list page.""" await self.navigate_and_wait(f"{YUNKE_BASE}/cms/wechat/wechatlist") employees = await self.page.evaluate("""() => { const rows = document.querySelectorAll('.ant-table-row'); const result = []; for (const tr of rows) { // Find the friend-count link (href contains wechatdetail/1/) const friendLink = tr.querySelector('a[href*="wechatdetail/1/"]'); const groupLink = tr.querySelector('a[href*="wechatdetail/2/"]'); if (friendLink) { const href = friendLink.getAttribute('href'); const wxId = href.match(/wechatdetail\\/1\\/([^?]+)/)?.[1]; const friendCount = parseInt(friendLink.textContent) || 0; const groupHref = groupLink?.getAttribute('href'); const groupCount = parseInt(groupLink?.textContent) || 0; result.push({ wxId, friendCount, groupCount, friendHref: href, groupHref }); } } return result; }""") # Paginate through ALL staff list pages all_employees = list(employees) page = 1 while await self.click_next_page(): page += 1 await self.page.wait_for_timeout(2000) more = await self.page.evaluate("""() => { const rows = document.querySelectorAll('.ant-table-row'); const result = []; for (const tr of rows) { const friendLink = tr.querySelector('a[href*="wechatdetail/1/"]'); const groupLink = tr.querySelector('a[href*="wechatdetail/2/"]'); if (friendLink) { const href = friendLink.getAttribute('href'); const wxId = href.match(/wechatdetail\\/1\\/([^?]+)/)?.[1]; const friendCount = parseInt(friendLink.textContent) || 0; const groupHref = groupLink?.getAttribute('href'); const groupCount = parseInt(groupLink?.textContent) || 0; result.push({ wxId, friendCount, groupCount, friendHref: href, groupHref }); } } return result; }""") all_employees.extend(more) logger.info(f" staff page {page}: +{len(more)} employees (total: {len(all_employees)})") # Dedup by wxId seen = set() deduped = [] for emp in all_employees: if emp["wxId"] not in seen: seen.add(emp["wxId"]) deduped.append(emp) logger.info(f"Found {len(deduped)} unique employees with friend links ({page} pages)") return deduped async def scrape_friends(cookie_str: str, max_employees: int = 100): """微信客户:浏览器直发 friend_list/es,**逐员工**(userId 过滤)翻页。 部门级全量分页排序不稳定(实测:云客声明 8691 vs 到手 distinct 6563, 8682 行里 2119 行跨页重复挤占页容量漏抓真客户,2026-08-20);逐员工范围 小漂移可忽略,且与云客后台「按员工看客户列表」口径一致。员工清单来自 本地 wechat_account(user_id 列)。 """ from app.db.base import async_engine from sqlalchemy import text as sa_text from app.scraper.wechat import _normalize from app.storage.repos import upsert_customer dept = await _root_dept_id() async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT wx_id FROM wechat_account WHERE wx_id IS NOT NULL AND wx_id != ''" )) employees = [row[0] for row in result.fetchall()][:max_employees] import datetime as _dt today = _dt.date.today().isoformat() total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) # ---- 阶段1:逐员工全量(保底完整性;该路径 lastChatTime 有滞后) ---- for i, wx_id in enumerate(employees): # 真机抓包(2026-08-20 云客客户列表页)确认的过滤参数:wechatId(非 # userId!)+ymd 窗口+friendTypeYunke=1+isDel=0 base_body = { "ymdStart": "0000-00-00", "ymdEnd": today, "wechatId": wx_id, "salesname": "", "friendTypeYunke": "1", "isDel": 0, } # total-aware 翻页:该端点分页不稳(中途偶发短页),按声明 total # 翻满或连续空页才停;跨页重复行 upsert 幂等无害。 rows: list[dict] = [] declared: int | None = None empty_streak = 0 page_num = 1 while page_num <= 400: resp = await _browser_fetch( page, INTERCEPT_PATHS["friend_list"], json_body={**base_body, "pageNum": page_num, "pageSize": 50}, ) basic = ((resp.get("data") or {}).get("basicResult") or {}) declared = basic.get("total") or declared page_rows = basic.get("data") or [] if not page_rows: empty_streak += 1 if empty_streak >= 2 or (declared and len(rows) >= declared): break else: empty_streak = 0 rows.extend(page_rows) if declared and len(rows) >= declared: break page_num += 1 await asyncio.sleep(1) records = [] for row in rows: rec = _normalize(row) if rec.get("friend_id"): records.append(rec) total_fetched += len(records) if records: await _upsert_all(upsert_customer, records) total_upserted += len(records) logger.info( f" [{i+1}/{len(employees)}] {wx_id[:16]}..: declared={declared} " f"rows={len(rows)} (total fetched {total_fetched})" ) await asyncio.sleep(2 + (i % 3)) # ---- 阶段2:全局实时刷新(云客列表页的参数集,跨员工按最近沟通实时排序; # 阶段1的 lastChatTime 有滞后,这里用实时值覆盖,保证本地排序与云客第一页一致) ---- global_body = { "friendName": "", "friendRemark": "", "saleName": "", "friendPhone": "", "realName": "", "emotionOrder": "", "emotionStatus": "", "tagGroupId": "", "isDel": 0, "isOnJob": 1, "showSale": "1", "friendTab": "all", "mergeSearchType": "saleName", "mergeSearchLabel": "", "ymdStart": today, "ymdEnd": today, "friendId": "", "departmentId": dept or "", "userId": "", } realtime_rows: list[dict] = [] for page_num in range(1, 41): # 前 40 页 x50 = 2000 行,覆盖全部近期活跃客户 resp = await _browser_fetch( page, INTERCEPT_PATHS["friend_list"], json_body={**global_body, "pageNum": page_num, "pageSize": 50}, ) basic = ((resp.get("data") or {}).get("basicResult") or {}) page_rows = basic.get("data") or [] if not page_rows: break realtime_rows.extend(page_rows) if len(page_rows) < 50: break await asyncio.sleep(1) # 只刷 last_chat_time 且只增不减(全局结果里也含旧 ES 副本,后写会 # 把实时值覆盖回旧值——max 条件天然免疫,2026-08-20 实测踩坑) from app.db.base import async_session_maker as _asm from sqlalchemy import text as _st updated = 0 async with _asm() as session: for row in realtime_rows: fid, ts = row.get("friendWxId"), row.get("lastChatTime") if not fid or not ts: continue r = await session.execute(_st( "UPDATE customer SET last_chat_time = :ts " "WHERE channel='wechat' AND owner_account_id = :o AND friend_id = :f " "AND (last_chat_time IS NULL OR last_chat_time < :ts)" ), {"ts": int(ts), "o": row.get("wxId"), "f": fid}) updated += r.rowcount await session.commit() logger.info(f" 实时刷新: {len(realtime_rows)} 行(全局实时排序), last_chat_time 更新 {updated} 条") logger.info(f"DONE: friends fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted async def scrape_groups(cookie_str: str, max_employees: int = 100): """微信群:浏览器直发 friend_list/es,**逐员工**(wechatId 过滤)翻页。 与好友共用端点,仅 friendTypeYunke=2;部门级分页不稳(见 scrape_friends 三连修,2026-08-20),同模式逐员工 + total-aware 翻页。""" import datetime as _dt from app.db.base import async_engine from sqlalchemy import text as sa_text from app.scraper.wechat import _normalize_chat_group from app.storage.repos import upsert_chat_group async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT wx_id FROM wechat_account WHERE wx_id IS NOT NULL AND wx_id != ''" )) employees = [row[0] for row in result.fetchall()][:max_employees] today = _dt.date.today().isoformat() total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) for i, wx_id in enumerate(employees): base_body = { "ymdStart": "0000-00-00", "ymdEnd": today, "wechatId": wx_id, "salesname": "", "friendTypeYunke": "2", "isDel": 0, } rows: list[dict] = [] declared: int | None = None empty_streak = 0 page_num = 1 while page_num <= 200: resp = await _browser_fetch( page, INTERCEPT_PATHS["friend_list"], json_body={**base_body, "pageNum": page_num, "pageSize": 50}, ) basic = ((resp.get("data") or {}).get("basicResult") or {}) declared = basic.get("total") or declared page_rows = basic.get("data") or [] if not page_rows: empty_streak += 1 if empty_streak >= 2 or (declared and len(rows) >= declared): break else: empty_streak = 0 rows.extend(page_rows) if declared and len(rows) >= declared: break page_num += 1 await asyncio.sleep(1) records = [] for row in rows: rec = _normalize_chat_group(row) if rec.get("room_id"): records.append(rec) total_fetched += len(records) if records: await _upsert_all(upsert_chat_group, records) total_upserted += len(records) if (i + 1) % 10 == 0: logger.info(f" [{i+1}/{len(employees)}] employees done, groups total: {total_fetched}") await asyncio.sleep(2 + (i % 3)) logger.info(f"DONE: groups fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=None): """按 hasLast/end 游标翻页 chats 类接口（微信 msg/chats 与 WA chatRecords 同协议）。 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 _discover_conversations(channel: str, voice_channel: str) -> list[tuple[str, str]]: """Conversation-pair discovery for the message scrapers (wechat + WA). Sources (union, dedup): * customer table — ALL pairs. The old ``last_chat_time IS NOT NULL`` filter skipped customers whose relationship is voice-call-only (NULL last_chat) and thus their chat history was never scraped → empty chat pages (diagnosed 2026-08-18). * voice-call pairs ``(weChatId, talker)`` from phone_call raw_json — peers that had calls but have NO customer row at all (deleted friends). Order: pairs with no messages yet FIRST (new coverage survives an interruption), then the rest by customer last_chat_time DESC. """ import json as _json from app.db.base import async_engine from sqlalchemy import text as sa_text async with async_engine.connect() as conn: cust = ( await conn.execute(sa_text( "SELECT owner_account_id, friend_id FROM customer " "WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL " "ORDER BY last_chat_time DESC" ), {"ch": channel}) ).fetchall() voice = ( await conn.execute(sa_text( "SELECT raw_json FROM phone_call WHERE channel=:vc" ), {"vc": voice_channel}) ).fetchall() with_msgs = ( await conn.execute(sa_text( "SELECT DISTINCT owner_account_id, talker_id FROM message " "WHERE channel=:ch AND (room_id IS NULL OR room_id='')" ), {"ch": channel}) ).fetchall() have = {(r[0], r[1]) for r in with_msgs} seen: set[tuple[str, str]] = set() fresh: list[tuple[str, str]] = [] known: list[tuple[str, str]] = [] def _add(key: tuple[str, str]) -> None: if key in seen: return seen.add(key) (fresh if key not in have else known).append(key) for owner, friend in cust: _add((owner, friend)) voice_only = 0 for (rj,) in voice: try: d = _json.loads(rj) except (TypeError, ValueError): continue owner, talker = d.get("weChatId"), d.get("talker") if not owner or not talker or str(talker).endswith("@chatroom"): continue before = len(seen) _add((str(owner), str(talker))) if len(seen) > before and (str(owner), str(talker)) not in cust: voice_only += 1 logger.info( f"discovery[{channel}]: {len(cust)} customer pairs + {voice_only} voice-only pairs, " f"{len(fresh)} unscanned (first) / {len(known)} already have messages" ) return fresh + known async def _room_watermarks() -> dict[str, int]: """每群本地最新消息时间(ms)——群消息增量水位(按 room_id)。""" 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 room_id, max(timestamp) FROM message " "WHERE room_id IS NOT NULL AND room_id != '' " "GROUP BY room_id" )) return {r[0]: r[1] or 0 for r in result.fetchall()} 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. **增量同步(默认)**:每对话水位 = 本地 max(timestamp);请求带 ``start=水位`` 只拉新消息,翻页遇到 ≤水位的消息即停(已同步区)。 客户表 last_chat_time(friends 实时刷新)≤ 水位的对话**整跳过**(0 请求) ——活跃对话才访问。``full_history=True`` 强制全量(首次/修复用)。 """ from sqlalchemy import text as sa_text from app.db.base import async_session_maker 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") # 增量模式:云客侧无新活动的对话跳过(0 请求)——速度来源 skipped = 0 if not full_history: kept = [] for pair in conversations: wm = watermarks.get(pair, 0) act = cust_activity.get(pair, 0) # 注意:不能用 +60s 缓冲——若新消息恰好落在水位后 60s 内, # act 永远 ≤ wm+60s → 每轮都跳过 → 该消息**永久漏掉**。 # act==wm 表示云客最新消息==我们最新消息,精确相等才跳。 if wm > 0 and act > 0 and act <= wm: 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 total_pages = 0 async with PageScraper(cookie_str) as scraper: # Just load the base URL to establish cookie context — no need for specific page # The fetch() calls work from any page (even login redirect) as long as cookies are set 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 # 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) 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', String(startMs)); 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, _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())}") 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["isSend"]) if row.get("isSend") is not None else (bool(row["mine"]) if row.get("mine") 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"), # fileTh/fileHd 常在 file 为空时仍带完整 OSS URL——映射进列,媒体下载器才够得着 "file_th_url": row.get("fileTh"), "file_hd_url": row.get("fileHd"), "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}") # Recreate the fetch page if context was destroyed. # A network blip / yunke outage also breaks this recovery goto — # retry patiently (30 × 2min = 1h) so a longer outage doesn't # kill the whole run (crashed twice on 2026-08-18 this way, # outages ~3h apart, ~1.5h into each run). 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) continue total_fetched += len(records) total_pages += pages if records: async with async_session_maker() as session: await upsert_messages(session, records) # 时效:用本对话最新消息时间刷新 customer.last_chat_time # (云客列表按最近沟通实时排序;friends 快照每日一次,消息增量每4h) max_ts = max((r.get("timestamp") or 0) for r in records) if max_ts: await session.execute(sa_text( "UPDATE customer SET last_chat_time = :ts " "WHERE channel='wechat' AND owner_account_id = :o AND friend_id = :f " "AND (last_chat_time IS NULL OR last_chat_time < :ts)" ), {"ts": max_ts, "o": owner_id, "f": friend_id}) await session.commit() total_upserted += len(records) if (i + 1) % 10 == 0: 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}, pages={total_pages}") 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["isSend"]) if row.get("isSend") is not None else (bool(row["mine"]) if row.get("mine") 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"), # 同 1:1:fileTh/fileHd 的 OSS URL 必须映射进列,媒体下载器才够得着 "file_th_url": row.get("fileTh"), "file_hd_url": row.get("fileHd"), "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, 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 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()] 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 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) watermark = 0 if full_history else room_wms.get(room_id, 0) 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', String(startMs)); 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, _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): 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. 微信 type=2 = 图片(新版,fileTh 带完整 OSS URL;真机 A**杰对话确认 2026-08-20——旧映射漏掉它导致图片被存成 unknown 前端不渲染)。 """ if raw_type is None: return None t = str(raw_type) mapping = { "1": "text", "2": "image", # 新版图片(fileTh URL) "3": "image_legacy", # 旧版图片(md5/键) "34": "voice", "43": "video", "47": "emoji", "49": "file", } return mapping.get(t, "unknown") # --------------------------------------------------------------------------- # Shared helpers for the fetch()-based entities # --------------------------------------------------------------------------- async def _open_fetch_page(scraper: PageScraper): """Open a stable page in the scraper's browser context for fetch() calls. The fetch() calls work from any same-origin page (even a login redirect) as long as the context carries the session cookies — same trick the message scraper uses. """ page = await scraper.context.new_page() await page.goto(f"{YUNKE_BASE}/", wait_until="domcontentloaded", timeout=15000) await page.wait_for_timeout(2000) return page async def _browser_fetch(page, path: str, json_body: dict | None = None, form_body: dict | None = None): """POST from inside the browser context (proper cookies/referrer => no 风控). ``json_body`` sends a JSON body; ``form_body`` sends form-urlencoded. Returns the parsed JSON response. """ if json_body is not None: return await page.evaluate( """async ([path, body]) => { const resp = await fetch(path, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body), }); return await resp.json(); }""", [path, json_body], ) return await page.evaluate( """async ([path, params]) => { const ps = new URLSearchParams(); for (const [k, v] of Object.entries(params)) ps.append(k, String(v)); const resp = await fetch(path, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body: ps.toString(), }); return await resp.json(); }""", [path, form_body or {}], ) def _rows_flat(body: dict) -> list[dict]: """Extract rows from the flat envelope: ``data.data[]`` (wxUser/list, queryWeChatVoiceList, call/list).""" data = body.get("data") if isinstance(body, dict) else None if not isinstance(data, dict): return [] rows = data.get("data") return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else [] def _rows_basic_result(body: dict) -> list[dict]: """friend_list/es 信封: ``data.basicResult.data[]``(好友与群共用)。""" data = body.get("data") if isinstance(body, dict) else None basic = data.get("basicResult") if isinstance(data, dict) else None rows = basic.get("data") if isinstance(basic, dict) else None return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else [] async def _root_dept_id() -> str: """本地 department 表的公司根部门 id(多个抓取端点的必填 body 字段)。""" from app.db.base import async_engine from sqlalchemy import text as sa_text async with async_engine.connect() as conn: row = (await conn.execute(sa_text( "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1" ))).first() return row[0] if row else "" async def _fetch_pages( page, path: str, *, base_body: dict, parse_rows, page_size: int = 50, max_pages: int = 500, form: bool = False, page_key: str = "pageNum", ) -> list[dict]: """按页码翻页浏览器直发 POST,直到短页;收集全部 rows 返回。 2026-08-20 起 list 类实体一律走此路径(导航+拦截被登录墙杀死,0 捕获)。 ``form=True`` 发 form-urlencoded(Spring @RequestParam,如 wxUser/list); ``page_key`` 兼容 WA 的 ``page``。首页响应 code 异常时醒目告警 cookie 失效(所有 list 端点健康时 code=10000)。 """ rows_all: list[dict] = [] first_row_sig: str | None = None page_num = 1 while page_num <= max_pages: body = {**base_body, page_key: page_num, "pageSize": page_size} if form: resp = await _browser_fetch(page, path, form_body=body) else: resp = await _browser_fetch(page, path, json_body=body) if page_num == 1 and isinstance(resp, dict): code = resp.get("code") if code is not None and str(code) not in ("10000",): logger.warning( f"⚠️ {path} 首页 code={code} ({resp.get('message')}) — " "cookie 可能已过期/失效,请从浏览器重新复制 YK_COOKIE 到 .env!" ) rows = parse_rows(resp) # 粘页检测:端点忽略页码参数时第 2 页起首行与第 1 页相同(实测 wxUser/list # 的页码字段是 page 而非 pageNum,2026-08-20)——丢弃重复页并停止 if rows: import json as _json sig = _json.dumps(rows[0], sort_keys=True, default=str)[:200] if page_num == 1: first_row_sig = sig elif sig == first_row_sig: logger.warning(f" {path} page {page_num} 与首页相同 — 端点忽略页码参数,停止翻页") break rows_all.extend(rows) logger.info(f" {path.rsplit('/', 1)[-1]} page {page_num}: {len(rows)} rows") if len(rows) < page_size: break page_num += 1 await asyncio.sleep(1.5) return rows_all def _rows_overviews(body: dict) -> list[dict]: """Extract rows from the WA envelope: ``data.overviews[]`` (businessSalesList / businessFriendList).""" data = body.get("data") if isinstance(body, dict) else None if not isinstance(data, dict): return [] rows = data.get("overviews") return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else [] async def _upsert_all(upsert_fn, records: list[dict]) -> None: """Batch-upsert records (each fn per record) in one committed session.""" from app.db.base import async_session_maker if not records: return async with async_session_maker() as session: for rec in records: await upsert_fn(session, rec) await session.commit() # --------------------------------------------------------------------------- # WeChat staff accounts (navigation + interception) # --------------------------------------------------------------------------- async def scrape_accounts(cookie_str: str, max_pages: int = 100): """微信员工账号:浏览器直发 wxUser/list(form-encoded,2026-08-20 起改 fetch)。 body 必填 isOnJob/departmentId/startdate/enddate(缺则 -90000);时间窗只 影响每账号统计数字不影响返回集合,取近 7 天。""" import datetime as _dt from app.scraper.wechat import _normalize_wxuser_account from app.storage.repos import upsert_wechat_account dept = await _root_dept_id() end_date = _dt.datetime.now().date() start_date = end_date - _dt.timedelta(days=7) async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) rows = await _fetch_pages( page, INTERCEPT_PATHS["wxUser_list"], base_body={ "isOnJob": "1", "departmentId": dept, "startdate": start_date.isoformat(), "enddate": end_date.isoformat(), "userId": "", "salesname": "", "weChatNickName": "", }, parse_rows=_rows_flat, form=True, page_key="page", # 此端点页码字段是 page(实测 pageNum 被无视) max_pages=max_pages, ) records = [] for idx, row in enumerate(rows): rec = _normalize_wxuser_account(row) # 记录云客 API 返回顺序(对齐员工列表排列) rec["list_order"] = idx records.append(rec) await _upsert_all(upsert_wechat_account, records) total = len(records) logger.info(f"DONE: accounts fetched={total}, upserted={total}") return total, total # --------------------------------------------------------------------------- # WhatsApp staff accounts (browser fetch) # --------------------------------------------------------------------------- async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100): """WA 员工账号:浏览器直发 businessSalesList(JSON body {pageNum,pageSize})。""" from app.scraper.whatsapp import _normalize_whatsapp_account from app.storage.repos import upsert_whatsapp_account async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) rows = await _fetch_pages( page, INTERCEPT_PATHS["wa_user"], base_body={}, parse_rows=_rows_overviews, max_pages=max_pages, ) records = [_normalize_whatsapp_account(row) for row in rows] await _upsert_all(upsert_whatsapp_account, records) total = len(records) logger.info(f"DONE: wa-accounts fetched={total}, upserted={total}") return total, total # --------------------------------------------------------------------------- # WhatsApp customers (browser fetch) # --------------------------------------------------------------------------- async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000): """WA 客户:浏览器直发 businessFriendList type=0,**逐员工**(uid 过滤)翻页。 全局查询有 ~1 万行硬上限(实测云客声明 11931 vs 全局翻页只到 10000, 2026-08-20);逐员工 ``uid`` 过滤(员工 WA 号)范围小无上限,页码字段是 ``page``,total-aware 翻页(短页继续,满声明数或连续空页停)。""" from app.db.base import async_engine from sqlalchemy import text as sa_text from app.scraper.whatsapp import _normalize_whatsapp_friend from app.storage.repos import upsert_customer dept = await _root_dept_id() async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT uid FROM whatsapp_account WHERE uid IS NOT NULL AND uid != ''" )) accounts = [row[0] for row in result.fetchall()] total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) for i, wa_uid in enumerate(accounts): base_body = { "searchInfo": "", "friendSearchInfo": "", "departmentId": dept, "type": "0", "uid": wa_uid, } rows: list[dict] = [] declared: int | None = None empty_streak = 0 page_num = 1 while page_num <= max_pages: resp = await _browser_fetch( page, INTERCEPT_PATHS["wa_friend"], json_body={**base_body, "page": page_num, "pageSize": 50}, ) data = resp.get("data") or {} declared = data.get("totalCount") or declared page_rows = data.get("overviews") or [] if not page_rows: empty_streak += 1 if empty_streak >= 2 or (declared and len(rows) >= declared): break else: empty_streak = 0 rows.extend(page_rows) if declared and len(rows) >= declared: break page_num += 1 await asyncio.sleep(1) records = [] for row in rows: rec = _normalize_whatsapp_friend(row) if rec.get("friend_id"): records.append(rec) total_fetched += len(records) if records: await _upsert_all(upsert_customer, records) total_upserted += len(records) logger.info( f" [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} rows={len(rows)}" ) await asyncio.sleep(2 + (i % 3)) logger.info(f"DONE: wa-friends fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted def _normalize_wa_group_row(row: dict) -> dict | None: """businessFriendList type=1 行 → ChatGroup 形状（真机确认 2026-08-20）。 type=1 返回的是好友字段名结构：friendUid=群 id（@g.us 后缀）、nickName= 群名、uid=员工 WA 号、salesNickName=员工 WA 昵称——与微信群的 friendWxId/friendWxNickname 完全不同，故不能复用 _normalize_chat_group。 非 @g.us 行（异常混入）返回 None。 """ friend_uid = str(row.get("friendUid") or "") if not friend_uid.endswith("@g.us"): return None return { "room_id": friend_uid, "channel": "whatsapp", "room_name": row.get("nickName"), "owner_account_id": row.get("uid"), "owner_account_nickname": row.get("salesNickName"), "owner_user_id": row.get("userId"), "owner_user_phone": row.get("salePhone"), "message_count": row.get("messageGroupCount"), "raw_json": row, } async def scrape_wa_groups(cookie_str: str, max_pages: int = 200): """Scrape WhatsApp GROUPS via browser fetch POST businessFriendList type=1. Same endpoint as WA friends with ``type:"1"`` — live-verified 2026-08-20: rows carry friendUid ending ``@g.us`` (group id), nickName = group name. Envelope: data.overviews; upsert to chat_group (channel='whatsapp'). """ from app.db.base import async_engine from sqlalchemy import text as sa_text from app.storage.repos import upsert_chat_group async with async_engine.connect() as conn: row = (await conn.execute(sa_text( "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1" ))).first() department_id = row[0] if row else "" if not department_id: logger.warning("wa-groups: no root department in DB — abort") return 0, 0 page_size = 50 total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) page_num = 1 while page_num <= max_pages: body = await _browser_fetch( page, INTERCEPT_PATHS["wa_friend"], json_body={ "searchInfo": "", "friendSearchInfo": "", "page": page_num, "pageSize": page_size, "departmentId": department_id, "type": "1", }, ) rows = _rows_overviews(body) records = [rec for r in rows if (rec := _normalize_wa_group_row(r)) is not None] total_fetched += len(records) await _upsert_all(upsert_chat_group, records) total_upserted += len(records) logger.info(f" wa-groups page {page_num}: {len(rows)} rows, {len(records)} groups") if len(rows) < page_size: break page_num += 1 await asyncio.sleep(1.5) logger.info(f"DONE: wa-groups fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted # --------------------------------------------------------------------------- # Voice call records (navigation + interception, both WeChat + WA pages) # --------------------------------------------------------------------------- async def scrape_voice_calls(cookie_str: str, max_pages: int = 200): """Scrape voice call records from BOTH channels, upsert to phone_call. * WeChat: browser-fetch queryWeChatVoiceList (callSource=0) with TWO passes isRoom=0(好友)/1(群). The voice-call-list page only sends isRoom=0 on load — group voice calls live behind the 微信群 tab, so pure navigation+interception never captured them (root cause of the empty 群 scope tab). Body must include callSource+departmentId+isRoom+ startTime+endTime or the API returns 200 with empty data. * WA: browser-fetch the same unified endpoint with callSource=4 (no group concept; interception stopped capturing on 2026-08-18). """ from app.db.base import async_engine from sqlalchemy import text as sa_text from app.scraper.wechat import _normalize_voice_call from app.scraper.whatsapp import _normalize_wa_voice_call from app.storage.repos import upsert_voice_call # queryWeChatVoiceList requires departmentId — take the company-root dept async with async_engine.connect() as conn: row = (await conn.execute(sa_text( "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1" ))).first() department_id = row[0] if row else "" if not department_id: logger.warning("voice-calls: no root department in DB — skipping wechat fetch passes") page_size = 100 total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: # --- WeChat: browser fetch, isRoom 0(好友)/1(群) two passes --- if department_id: page = await _open_fetch_page(scraper) for is_room, label in (("0", "好友"), ("1", "群")): records = [] page_num = 1 while page_num <= max_pages: body = await _browser_fetch( page, INTERCEPT_PATHS["call_list"], json_body={ "pageNum": page_num, "pageSize": page_size, "callSource": "0", "isRoom": is_room, "startTime": "2000-01-01 00:00:00", "endTime": "2099-12-31 23:59:59", "sort": "", "ids": [], "userId": "", "departmentId": department_id, }, ) rows = _rows_flat(body) for row in rows: rec = _normalize_voice_call(row) if rec.get("call_id") is not None: records.append(rec) if len(rows) < page_size: break page_num += 1 await asyncio.sleep(1) total_fetched += len(records) await _upsert_all(upsert_voice_call, records) total_upserted += len(records) logger.info(f" wechat isRoom={is_room}({label}): {len(records)} calls, {page_num} page(s)") await asyncio.sleep(2) # --- WA: browser fetch (callSource=4, no group concept; same unified endpoint) --- # Interception on the wavoicecalllist page used to work but captured 0 # responses on 2026-08-18 (page no longer fires the API on load) — direct # browser fetch with the full body is the reliable path, same as WeChat. if department_id: records = [] page_num = 1 while page_num <= max_pages: body = await _browser_fetch( page, INTERCEPT_PATHS["call_list"], json_body={ "pageNum": page_num, "pageSize": page_size, "callSource": "4", "isRoom": "0", "startTime": "2000-01-01 00:00:00", "endTime": "2099-12-31 23:59:59", "sort": "", "ids": [], "userId": "", "departmentId": department_id, }, ) rows = _rows_flat(body) for row in rows: rec = _normalize_wa_voice_call(row) if rec.get("call_id") is not None: records.append(rec) if len(rows) < page_size: break page_num += 1 await asyncio.sleep(1) total_fetched += len(records) await _upsert_all(upsert_voice_call, records) total_upserted += len(records) logger.info(f" wa(callSource=4): {len(records)} calls, {page_num} page(s)") logger.info(f"DONE: voice-calls fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted # --------------------------------------------------------------------------- # Long voice clips (browser fetch, per employee) # --------------------------------------------------------------------------- async def scrape_voice_clips(cookie_str: str, max_employees: int = 200, page_size: int = 50): """Scrape long voice clips via browser fetch POST /wechatweb/call/list. The endpoint is per-employee: employee user_ids come from the LOCAL wechat_account / whatsapp_account tables (channel resolved the same way). JSON body {pageNum, pageSize, userId} (真机确认 2026-08-07 spec §2). """ from app.db.base import async_engine from app.scraper.voice_clip import _normalize_voice_clip from app.storage.repos import upsert_voice_clip from sqlalchemy import text as sa_text # Resolve employee user_id -> channel from local staff-account tables. # WeChat wins when an id appears in both tables. user_channel: dict[str, str] = {} async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT DISTINCT user_id FROM whatsapp_account " "WHERE user_id IS NOT NULL AND user_id != ''" )) for (uid,) in result.fetchall(): user_channel[uid] = "whatsapp" result = await conn.execute(sa_text( "SELECT DISTINCT user_id FROM wechat_account " "WHERE user_id IS NOT NULL AND user_id != ''" )) for (uid,) in result.fetchall(): user_channel[uid] = "wechat" employees = sorted(user_channel.items()) logger.info(f"Found {len(employees)} employees; scraping {min(max_employees, len(employees))}") total_fetched = 0 total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) for i, (user_id, channel) in enumerate(employees[:max_employees]): records = [] try: page_num = 1 while True: body = await _browser_fetch( page, INTERCEPT_PATHS["voice_list"], json_body={"pageNum": page_num, "pageSize": page_size, "userId": user_id}, ) rows = _rows_flat(body if isinstance(body, dict) else {}) if not rows: break records.extend(_normalize_voice_clip(row, channel=channel) for row in rows) data = body.get("data") or {} total = data.get("total") if isinstance(total, (int, float)) and page_num * page_size >= int(total): break page_num += 1 await asyncio.sleep(1) except Exception as e: logger.warning(f" [{i+1}] user {user_id} failed: {e}") try: await page.close() except Exception: pass page = await _open_fetch_page(scraper) continue total_fetched += len(records) if records: await _upsert_all(upsert_voice_clip, records) total_upserted += len(records) if (i + 1) % 10 == 0 or not records: logger.info(f" [{i+1}/{min(max_employees, len(employees))}] {user_id}({channel}): {len(records)} clips (total: {total_fetched})") await asyncio.sleep(2 + (i % 3)) logger.info(f"DONE: voice-clips fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted # --------------------------------------------------------------------------- # WhatsApp messages (browser fetch, per conversation, backward cursor) # --------------------------------------------------------------------------- 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}). Conversation pairs come from _discover_conversations (customer table ∪ voice-call pairs). Pages backward via data.end / data.hasLast (same cursor strategy as the WeChat msg/chats scraper). """ import time as _time from sqlalchemy import text as sa_text from app.db.base import async_session_maker from app.scraper.whatsapp import _is_wa_divider, _normalize_whatsapp_message 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") skipped = 0 if not full_history: kept = [] for pair in conversations: wm = watermarks.get(pair, 0) act = cust_activity.get(pair, 0) # 注意:不能用 +60s 缓冲——若新消息恰好落在水位后 60s 内, # act 永远 ≤ wm+60s → 每轮都跳过 → 该消息**永久漏掉**。 # act==wm 表示云客最新消息==我们最新消息,精确相等才跳。 if wm > 0 and act > 0 and act <= wm: 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 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) 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": watermark, "end": end_ms}, ) data = body.get("data") if isinstance(body, dict) else None if not isinstance(data, dict): break msgs = data.get("messages") or [] for row in msgs: if isinstance(row, dict) and not _is_wa_divider(row): 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 end_ms = int(end) - 1 await asyncio.sleep(1) except Exception as e: logger.warning(f" [{i+1}] {str(uid)[:20]}:{str(friend_uid)[:20]} failed: {e}") # Recreate the fetch page; retry patiently on outages (same # resilience as the WeChat message scraper — 30 × 2min). for attempt in range(30): try: try: await page.close() except Exception: pass page = await _open_fetch_page(scraper) break except Exception as rec: logger.warning(f" recovery {attempt + 1}/30 failed: {rec}") if attempt == 29: raise await asyncio.sleep(120) continue total_fetched += len(records) if records: # upsert_messages 收整列表(逐条版 _upsert_all 会把 dict 迭代成 # key 字符串 → TypeError, 2026-08-19 WA 阶段首条即崩的根因) async with async_session_maker() as session: await upsert_messages(session, records) # 时效:刷新 customer.last_chat_time(同微信 1:1,云客按最近沟通实时排序) max_ts = max((r.get("timestamp") or 0) for r in records) if max_ts: await session.execute(sa_text( "UPDATE customer SET last_chat_time = :ts " "WHERE channel='whatsapp' AND owner_account_id = :o AND friend_id = :f " "AND (last_chat_time IS NULL OR last_chat_time < :ts)" ), {"ts": max_ts, "o": uid, "f": friend_uid}) await session.commit() total_upserted += len(records) if (i + 1) % 10 == 0: logger.info(f" [{i+1}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}") await asyncio.sleep(2 + (i % 3)) logger.info(f"DONE: wa-messages fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted def _normalize_wa_group_message(record: dict) -> dict: """chatRecords 群消息行 → Message 形状（friendUid=@g.us,真机确认 2026-08-20）。 与 WA 1:1 行同构,差异:``chatRoomTalker`` = 群内发送者(@lid)、friendUid = 群 id(@g.us);keyId 通常有值(与微信群 msgId=null 不同),缺失时用 owner:room:createTime:chatRoomTalker 合成稳定主键。 """ from app.scraper.whatsapp import _normalize_whatsapp_message rec = _normalize_whatsapp_message(record) rec["room_id"] = record.get("friendUid") rec["talker_id"] = record.get("chatRoomTalker") # 群行的 file 常带直连 OSS URL(yk-whatsapp.oss),优先于 1:1 用的 mediaName 不透明名 rec["file_url"] = record.get("file") or record.get("mediaName") if not rec.get("msg_id"): rec["msg_id"] = ( f"{record.get('uid')}:{record.get('friendUid')}:" f"{record.get('createTime')}:{record.get('chatRoomTalker')}" ) return rec 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 同套路)。 群来自本地 chat_group(channel='whatsapp');翻页复用 paginate_chats (hasLast/end 游标);守护循环同微信群抓取器(毒群重启跳过)。 """ import time as _time from app.db.base import async_engine, async_session_maker from app.scraper.whatsapp import _is_wa_divider 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='whatsapp' AND owner_account_id IS NOT NULL" )) groups = [(row[0], row[1]) for row in result.fetchall()] 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 total_pages = 0 scan = groups[:max_groups] idx = 0 relaunches = 0 while idx < len(scan): try: async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) while idx < len(scan): owner_id, room_id = scan[idx] i = idx records = [] 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, _wm=watermark): return await _browser_fetch( page, "/qqweb/WhatsAppUser/chatRecords", 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): continue records.append(_normalize_wa_group_message(row)) except Exception as e: logger.warning(f" [{i+1}] {str(owner_id)[:18]}:{str(room_id)[:18]} failed: {e}") for attempt in range(30): try: try: await page.close() except Exception: pass page = await _open_fetch_page(scraper) 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: wa-group-messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}") return total_fetched, total_upserted # --------------------------------------------------------------------------- # Media downloads (OSS CDN — NOT API calls, safe from 风控) # --------------------------------------------------------------------------- async def scrape_media(cookie_str: str, batch_size: int = 50): """Download media artifacts into the local object store: * call recordings (phone_call.record_file OSS URLs) * voice clip audio (voice_clip.oss_file_name OSS URLs) * message images/videos (message.file_url / file_th_url / file_hd_url) These are plain OSS CDN downloads (not CRM API calls), so httpx direct fetching is safe. The YunkeClient is only used as the httpx transport. """ from app.config import get_settings from app.db.base import async_session_maker from app.scraper.media import ( download_call_recordings, download_message_images, download_voice_clip_audio, ) from app.session.yunke_client import YunkeClient from app.storage.object_store import ObjectStore settings = get_settings() cookie_dict = {c["name"]: c["value"] for c in parse_cookies(cookie_str)} # Session-mode client purely as an httpx transport for OSS downloads — # userid/company only matter for login(), which we never call here. client = YunkeClient( base_url=YUNKE_BASE, userid=settings.yk_userid or "oss", company=settings.yk_company or "oss", cookies=cookie_dict, ) store = ObjectStore() totals: dict[str, tuple[int, int]] = {} try: async with async_session_maker() as session: totals["call_recordings"] = await download_call_recordings(client, store, session, batch_size=batch_size) totals["voice_clip_audio"] = await download_voice_clip_audio(client, store, session, batch_size=batch_size) totals["message_images"] = await download_message_images(client, store, session, batch_size=batch_size) finally: await client.aclose() for name, (attempted, localized) in totals.items(): logger.info(f" {name}: attempted={attempted}, localized={localized}") logger.info(f"DONE: media {totals}") return totals async def main(): parser = argparse.ArgumentParser(description="页面浏览器抓取器") parser.add_argument( "--entity", required=True, choices=[ "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) "wa-groups", # WA群 (browser fetch businessFriendList type=1) "voice-calls", # 语音通话记录 (voice-call-list + wavoicecalllist) "voice-clips", # 长语音 (browser fetch call/list per employee) "wa-messages", # WA消息 (browser fetch chatRecords per conversation) "wa-group-messages", # WA群消息 (chatRecords with friendUid=@g.us) "media", # OSS媒体下载 (录音/长语音音频/消息图片) ], 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/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", "") if not cookie: # Try reading from .env from app.config import get_settings cookie = get_settings().yk_cookie if not cookie: logger.error("YK_COOKIE not set in .env or environment") return logger.info(f"Starting page-based scrape: entity={args.entity}") if args.entity == "friends": await scrape_friends(cookie, args.max_employees) elif args.entity == "groups": await scrape_groups(cookie, args.max_employees) elif args.entity == "messages": await scrape_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "group-messages": 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": await scrape_wa_accounts(cookie, args.max_pages) elif args.entity == "wa-friends": await scrape_wa_friends(cookie, args.max_pages) elif args.entity == "wa-groups": await scrape_wa_groups(cookie, args.max_pages) elif args.entity == "voice-calls": await scrape_voice_calls(cookie, args.max_pages) 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, args.full_history) elif args.entity == "wa-group-messages": await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history) elif args.entity == "media": await scrape_media(cookie, args.batch_size) if __name__ == "__main__": asyncio.run(main())