xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scraper): 页面抓取器支持全实体(账号/WA客户/语音/长语音/WA消息,全浏览器方式)

Browse Source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-15 14:44:20 +08:00
parent 7beae8cf55
commit 4a148706b9
1 changed files with 531 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/scripts/scrape_via_page.py
+531 -25
View File
@@ -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__":