diff --git a/yunke-server/app/api/routers/calls.py b/yunke-server/app/api/routers/calls.py index e547daf..f7996c5 100644 --- a/yunke-server/app/api/routers/calls.py +++ b/yunke-server/app/api/routers/calls.py @@ -9,7 +9,7 @@ Ordered by ``start_call_time desc`` (nulls last) — most recent calls first. from __future__ import annotations -from typing import Annotated +from typing import Annotated, Literal import sqlalchemy as sa from fastapi import APIRouter, Depends, Query @@ -35,13 +35,22 @@ async def list_calls( user_id: Annotated[str | None, Query()] = None, start: Annotated[int | None, Query(ge=0)] = None, end: Annotated[int | None, Query(ge=0)] = None, + scope: Annotated[Literal["friend", "group"] | None, Query()] = None, ) -> Page[PhoneCallOut]: - """List phone/voice calls with optional filters, newest first.""" + """List phone/voice calls with optional filters, newest first. + + ``scope`` filters voice calls by peer kind: ``group`` keeps only group + calls (``call_number`` ends ``@chatroom``), ``friend`` excludes them — + verified exactly equivalent to the CRM's ``isRoom`` flag (2026-08-18). + """ conditions = [] if channel is not None: conditions.append(PhoneCall.channel == channel) if user_id is not None: conditions.append(PhoneCall.user_id == user_id) + if scope is not None: + is_room = PhoneCall.call_number.endswith("@chatroom") + conditions.append(is_room if scope == "group" else ~is_room) if start is not None: conditions.append(PhoneCall.start_call_time >= start) if end is not None: diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index c94e275..2f2f285 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -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: diff --git a/yunke-server/tests/test_api_calls.py b/yunke-server/tests/test_api_calls.py index cdae684..d01f400 100644 --- a/yunke-server/tests/test_api_calls.py +++ b/yunke-server/tests/test_api_calls.py @@ -291,3 +291,59 @@ async def test_record_media_hash_serializes(make_client): assert by_id["c1"]["record_media_hash"] is None assert by_id["c2"]["record_media_hash"] is None assert by_id["c3"]["record_media_hash"] is None + + +# --------------------------------------------------------------------------- +# scope filter (voice-call 好友/群 server-side filter, 2026-08-18) +# --------------------------------------------------------------------------- + +async def _seed_voice_scopes(session: AsyncSession) -> None: + """Friend + group voice calls (group peer id ends '@chatroom').""" + session.add_all( + [ + PhoneCall( + call_id="v_friend", + channel="voice_call", + user_id="u2", + user_id_name="Bob", + start_call_time=1000, + call_number="q290059956", + ), + PhoneCall( + call_id="v_group", + channel="voice_call", + user_id="u2", + user_id_name="Bob", + start_call_time=2000, + call_number="45877925713@chatroom", + ), + ] + ) + await session.commit() + + +async def test_scope_group_returns_only_chatroom_calls(make_client): + """scope=group -> only '@chatroom' rows, total reflects the filter.""" + client = await make_client(_seed_voice_scopes) + r = await client.get("/api/calls", params={"channel": "voice_call", "scope": "group"}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 1 + assert [it["call_id"] for it in body["items"]] == ["v_group"] + + +async def test_scope_friend_excludes_chatroom_calls(make_client): + """scope=friend -> only non-'@chatroom' rows.""" + client = await make_client(_seed_voice_scopes) + r = await client.get("/api/calls", params={"channel": "voice_call", "scope": "friend"}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 1 + assert [it["call_id"] for it in body["items"]] == ["v_friend"] + + +async def test_scope_invalid_value_rejected(make_client): + """scope is Literal['friend','group'] — anything else is a 422.""" + client = await make_client(_seed_voice_scopes) + r = await client.get("/api/calls", params={"scope": "bogus"}) + assert r.status_code == 422 diff --git a/yunkes/src/api/index.ts b/yunkes/src/api/index.ts index 04d9419..838c169 100644 --- a/yunkes/src/api/index.ts +++ b/yunkes/src/api/index.ts @@ -176,6 +176,8 @@ export interface CallQuery { /** ms-epoch inclusive window compared against start_call_time. */ start?: number end?: number + /** voice-call peer kind: group = call_number ends '@chatroom'. */ + scope?: 'friend' | 'group' page?: number page_size?: number } @@ -188,6 +190,7 @@ export function getCalls(query: CallQuery = {}): Promise> { user_id: query.user_id, start: query.start, end: query.end, + scope: query.scope, page: query.page ?? 1, page_size: query.page_size ?? 20, }), diff --git a/yunkes/src/views/CallList.vue b/yunkes/src/views/CallList.vue index 12404fc..9666de0 100644 --- a/yunkes/src/views/CallList.vue +++ b/yunkes/src/views/CallList.vue @@ -28,7 +28,7 @@ const callType = ref( // 微信语音 tab scope filter: 对齐云客 微信好友/微信群 二分(isRoom 0=好友, 1=群)。 // 仅在 voice_call tab 生效;client-side 过滤当前页。 -const voiceScope = ref<'all' | 'friend' | 'group'>('all') +const voiceScope = ref<'friend' | 'group'>('friend') const calls = ref([]) const total = ref(0) @@ -64,6 +64,8 @@ async function load(): Promise { try { const res = await getCalls({ channel: callType.value, + // 微信语音 好友/群 scope 为服务端过滤(@chatroom 后缀判群,与云客 isRoom 等价) + scope: callType.value === 'voice_call' ? voiceScope.value : undefined, page: page.value, page_size: pageSize.value, }) @@ -97,7 +99,7 @@ watch( callType.value = next page.value = 1 // 复位 scope,避免把 微信群/好友 筛选带到 WA语音。 - voiceScope.value = 'all' + voiceScope.value = 'friend' load() } }, @@ -136,19 +138,11 @@ async function exportCalls(): Promise { } /** - * 当前表格要渲染的行。非 voice_call tab 原样返回;voice_call tab 按 voiceScope - * 依 isRoom(0=好友, 1=群)做页内过滤。云客 isRoom 可能是 number/string/boolean, - * 三种都判一下。 + * 好友/群 scope 切换 → 服务端重查(与云客一致按 isRoom 分舱,分页/总数都对)。 */ -const displayCalls = computed(() => { - if (callType.value !== 'voice_call' || voiceScope.value === 'all') { - return calls.value - } - return calls.value.filter((row) => { - const v = cell(row, 'isRoom') - const isGroup = v === 1 || v === '1' || v === true - return voiceScope.value === 'group' ? isGroup : !isGroup - }) +watch(voiceScope, () => { + page.value = 1 + load() }) /** @@ -283,14 +277,13 @@ onMounted(load) size="small" style="margin-bottom: 8px" > - 全部 微信好友 微信群