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

fix(calls+scraper): 语音群scope服务端过滤+对话发现缺口补全+抓取韧性三连修

Browse Source 
- /api/calls 加 scope=friend|group(@chatroom后缀判群,与isRoom实测等价120=120),CallList去掉「全部」tab改服务端查
- 对话发现:全部customer pair ∪ 语音pair(weChatId,talker),未扫过的排最前——修复纯语音关系客户聊天记录永远空(35%语音客户pair从未扫过)
- voice-calls改浏览器直发(isRoom=0/1双趟+WA callSource=4,拦截方式已两次失效)
- 断连恢复重试30×2min(云客约3h一次断连,已实战扛过多次)
- WA messages upsert改整列表直调(_upsert_all逐条传dict被迭代成key字符串→TypeError)
- 405测试绿含scope新增3个

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-19 16:48:22 +08:00
parent f9ed7dd821
commit ab5a918a3b
5 changed files with 293 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/scripts/scrape_via_page.py
+214 -62
View File
@@ -390,26 +390,91 @@ async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=No
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 scrape_messages(cookie_str: str, max_conversations: int = 100):
"""Scrape wechat messages via browser fetch POST /wechatweb/msg/chats.
Conversation pairs come from the local customer table (channel='wechat').
Pages backward through full history via data.end / data.hasLast
(paginate_chats, same cursor strategy as the WA chatRecords scraper).
Conversation pairs come from _discover_conversations (customer table ∪
voice-call pairs). Pages backward through full history via data.end /
data.hasLast (paginate_chats, same cursor strategy as the WA chatRecords
scraper).
"""
from app.db.base import async_session_maker, async_engine
from sqlalchemy import text as sa_text
from app.db.base import async_session_maker
from app.storage.repos import upsert_messages
# Discover conversations from local customer table — prioritize recent activity
async with async_engine.connect() as conn:
result = await conn.execute(sa_text(
"SELECT owner_account_id, friend_id FROM customer "
"WHERE channel='wechat' 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()]
conversations = await _discover_conversations("wechat", "voice_call")
logger.info(f"Discovered {len(conversations)} conversations, will scrape {min(max_conversations, len(conversations))}")
@@ -492,15 +557,27 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
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
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
# 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:
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)
@@ -715,42 +792,111 @@ async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
async def scrape_voice_calls(cookie_str: str, max_pages: int = 200):
"""Scrape voice call records from BOTH channel pages, upsert to phone_call.
"""Scrape voice call records from BOTH channels, 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.
* 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:
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,
)
# --- 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 = []
for resp in responses:
for row in _rows_flat(resp.get("body", {})):
rec = normalize(row)
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" {label}: {len(records)} calls")
await asyncio.sleep(3)
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
@@ -850,24 +996,16 @@ 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
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 app.db.base import async_engine
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
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()]
conversations = await _discover_conversations("whatsapp", "wa_voice_call")
logger.info(f"Discovered {len(conversations)} WA conversations, will scrape {min(max_conversations, len(conversations))}")
@@ -903,16 +1041,30 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
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)
# 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:
await _upsert_all(upsert_messages, records)
# upsert_messages 收整列表(逐条版 _upsert_all 会把 dict 迭代成
# key 字符串 → TypeError, 2026-08-19 WA 阶段首条即崩的根因)
async with async_session_maker() as session:
await upsert_messages(session, records)
await session.commit()
total_upserted += len(records)
if (i + 1) % 10 == 0: