diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 864c06a..0182712 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -9,7 +9,14 @@ 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 读取。 """ @@ -365,12 +372,13 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): from sqlalchemy import text as sa_text from app.storage.repos import upsert_messages - # Discover conversations from local customer table + # Discover conversations from local customer table — prioritize recent activity async with async_engine.connect() as conn: result = await conn.execute(sa_text( - "SELECT DISTINCT owner_account_id, friend_id FROM customer " + "SELECT owner_account_id, friend_id FROM customer " "WHERE channel='wechat' AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL " - "ORDER BY owner_account_id, friend_id" + "AND last_chat_time IS NOT NULL " + "ORDER BY last_chat_time DESC" )) conversations = [(row[0], row[1]) for row in result.fetchall()] @@ -380,45 +388,68 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): total_upserted = 0 async with PageScraper(cookie_str) as scraper: - page = scraper.page - # First navigate to the CRM to establish session context - await page.goto(f"{YUNKE_BASE}/cms/wechat/wechatlist", wait_until="domcontentloaded") - await page.wait_for_timeout(3000) + # 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]): try: # Use browser fetch() to call the message API from the page context # (the browser sends it with proper cookies/headers → no 风控) - body = await page.evaluate("""async ([ownerId, friendId]) => { + # NOTE: msg/chats uses FORM-ENCODED + cursor params (start/end, not pageNum) + import time as _time + now_ms = int(_time.time() * 1000) + body = await page.evaluate("""async ([ownerId, friendId, endMs]) => { + const params = new URLSearchParams(); + params.append('wechatId', ownerId); + params.append('friendId', friendId); + params.append('types', ''); + params.append('start', '0'); + params.append('end', String(endMs)); const resp = await fetch('/wechatweb/msg/chats', { method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - pageNum: 1, - pageSize: 50, - wechatId: ownerId, - friendId: friendId, - type: '', - }), + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: params.toString(), }); return await resp.json(); - }""", [owner_id, friend_id]) + }""", [owner_id, friend_id, now_ms]) responses = [{"body": body, "status": 200}] + # Debug: log first response structure + if i == 0: + logger.info(f" DEBUG first response keys: {list(body.keys()) if isinstance(body, dict) else type(body)}") + if isinstance(body, dict): + d = body.get("data") + logger.info(f" DEBUG data type: {type(d)}, keys: {list(d.keys()) if isinstance(d, dict) else 'N/A'}") + if isinstance(d, dict): + msgs = d.get("messages") + logger.info(f" DEBUG messages type: {type(msgs)}, len: {len(msgs) if isinstance(msgs, list) else 'N/A'}") + logger.info(f" DEBUG body preview: {str(body)[:500]}") + except Exception as e: logger.warning(f" [{i+1}] {owner_id[:12]}:{friend_id[:12]} failed: {e}") - # Re-navigate to reset the page if session expired - await page.goto(f"{YUNKE_BASE}/cms/wechat/wechatlist", wait_until="domcontentloaded") - await page.wait_for_timeout(2000) + # Recreate the fetch page if context was destroyed + try: + 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 continue - # Extract message records + # Extract message records (msg/chats response: body.data.messages[]) records = [] for resp in responses: body = resp.get("body", {}) data = body.get("data") or {} - rows = data.get("data") or [] + rows = data.get("messages") or [] for row in rows: if not isinstance(row, dict): continue @@ -471,11 +502,472 @@ def _normalize_msg_type(raw_type) -> str | None: 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_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): + """Scrape WeChat staff accounts: navigate /cms/wechat/wechatlist, + intercept /wechatweb/wxUser/list, upsert to wechat_account (+staff).""" + from app.scraper.wechat import _normalize_wxuser_account + from app.storage.repos import upsert_wechat_account + + total_fetched = 0 + async with PageScraper(cookie_str) as scraper: + responses = await scraper.scrape_entity_pages( + list_url=f"{YUNKE_BASE}/cms/wechat/wechatlist", + api_path=INTERCEPT_PATHS["wxUser_list"], + max_pages=max_pages, + ) + records = [] + for resp in responses: + for row in _rows_flat(resp.get("body", {})): + records.append(_normalize_wxuser_account(row)) + total_fetched = len(records) + await _upsert_all(upsert_wechat_account, records) + + logger.info(f"DONE: accounts fetched={total_fetched}, upserted={total_fetched}") + return total_fetched, total_fetched + + +# --------------------------------------------------------------------------- +# WhatsApp staff accounts (navigation + interception) +# --------------------------------------------------------------------------- + + +async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100): + """Scrape WA staff accounts: navigate /cms/wechat/waBusiness/userlist, + intercept /qqweb/WhatsAppUser/businessSalesList, upsert to whatsapp_account.""" + from app.scraper.whatsapp import _normalize_whatsapp_account + from app.storage.repos import upsert_whatsapp_account + + async with PageScraper(cookie_str) as scraper: + responses = await scraper.scrape_entity_pages( + list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/userlist", + api_path=INTERCEPT_PATHS["wa_user"], + max_pages=max_pages, + ) + records = [] + for resp in responses: + for row in _rows_overviews(resp.get("body", {})): + records.append(_normalize_whatsapp_account(row)) + total = len(records) + await _upsert_all(upsert_whatsapp_account, records) + + logger.info(f"DONE: wa-accounts fetched={total}, upserted={total}") + return total, total + + +# --------------------------------------------------------------------------- +# WhatsApp customers (navigation + interception) +# --------------------------------------------------------------------------- + + +async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000): + """Scrape WA customers: navigate /cms/wechat/waBusiness/friendlist, + intercept /qqweb/WhatsAppUser/businessFriendList, upsert to customer + (channel='whatsapp'). Live total ~11k rows => high default max_pages.""" + from app.scraper.whatsapp import _normalize_whatsapp_friend + from app.storage.repos import upsert_customer + + total_fetched = 0 + total_upserted = 0 + async with PageScraper(cookie_str) as scraper: + responses = await scraper.scrape_entity_pages( + list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/friendlist", + api_path=INTERCEPT_PATHS["wa_friend"], + max_pages=max_pages, + ) + # Upsert in chunks so one giant page-walk doesn't build a 11k-record list + # before the first commit. + chunk: list[dict] = [] + for resp in responses: + records = [_normalize_whatsapp_friend(row) for row in _rows_overviews(resp.get("body", {}))] + total_fetched += len(records) + chunk.extend(records) + if len(chunk) >= 500: + await _upsert_all(upsert_customer, chunk) + total_upserted += len(chunk) + chunk = [] + await _upsert_all(upsert_customer, chunk) + total_upserted += len(chunk) + + logger.info(f"DONE: wa-friends 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 channel pages, upsert to phone_call. + + * WeChat page /cms/wechat/voice-call-list -> channel='voice_call' + * WA page /cms/wechat/waBusiness/wavoicecalllist -> channel='wa_voice_call' + + Both pages call the SAME unified API (/wechatweb/callRecord/ + queryWeChatVoiceList) with callSource 0/4 — the SPA sends it, we intercept. + """ + 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 + + total_fetched = 0 + total_upserted = 0 + async with PageScraper(cookie_str) as scraper: + for label, url, normalize in ( + ("wechat", f"{YUNKE_BASE}/cms/wechat/voice-call-list", _normalize_voice_call), + ("wa", f"{YUNKE_BASE}/cms/wechat/waBusiness/wavoicecalllist", _normalize_wa_voice_call), + ): + logger.info(f"voice-calls: navigating {label} page...") + responses = await scraper.scrape_entity_pages( + list_url=url, + api_path=INTERCEPT_PATHS["call_list"], + max_pages=max_pages, + ) + records = [] + for resp in responses: + for row in _rows_flat(resp.get("body", {})): + rec = normalize(row) + if rec.get("call_id") is not None: + records.append(rec) + total_fetched += len(records) + await _upsert_all(upsert_voice_call, records) + total_upserted += len(records) + logger.info(f" {label}: {len(records)} calls") + await asyncio.sleep(3) + + 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): + """Scrape WhatsApp chat records via browser fetch POST + /qqweb/WhatsAppUser/chatRecords (JSON body {uid, friendUid, start, end}). + + Conversation pairs come from the local customer table + (channel='whatsapp'). Pages backward via data.end / data.hasLast + (same cursor strategy as the WeChat msg/chats scraper). + """ + import time as _time + from app.db.base import async_engine + from app.scraper.whatsapp import _is_wa_divider, _normalize_whatsapp_message + 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, friend_id FROM customer " + "WHERE channel='whatsapp' AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL " + "AND last_chat_time IS NOT NULL " + "ORDER BY last_chat_time DESC" + )) + conversations = [(row[0], row[1]) for row in result.fetchall()] + + logger.info(f"Discovered {len(conversations)} WA conversations, will scrape {min(max_conversations, len(conversations))}") + + 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 = [] + 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}, + ) + 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 + 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}") + try: + await page.close() + except Exception: + pass + page = await _open_fetch_page(scraper) + continue + + total_fetched += len(records) + if records: + await _upsert_all(upsert_messages, records) + 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 + + +# --------------------------------------------------------------------------- +# 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", "groups", "messages"], help="抓取实体") - parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工") - parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话") + parser.add_argument( + "--entity", + required=True, + choices=[ + "friends", # 微信客户 (per-employee friend pages) + "groups", # 微信群 (per-employee group pages) + "messages", # 微信消息 (browser fetch msg/chats) + "accounts", # 微信员工账号 (wechatlist page interception) + "wa-accounts", # WA员工账号 (waBusiness/userlist interception) + "wa-friends", # WA客户 (waBusiness/friendlist interception) + "voice-calls", # 语音通话记录 (voice-call-list + wavoicecalllist) + "voice-clips", # 长语音 (browser fetch call/list per employee) + "wa-messages", # WA消息 (browser fetch chatRecords per conversation) + "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)") + 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() cookie = os.environ.get("YK_COOKIE", "") @@ -495,6 +987,20 @@ async def main(): await scrape_groups(cookie, args.max_employees) elif args.entity == "messages": await scrape_messages(cookie, args.max_conversations) + 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 == "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) + elif args.entity == "media": + await scrape_media(cookie, args.batch_size) if __name__ == "__main__":