feat: wechat voice-call scraper + group message coverage
- scrape_wechat_voice_calls: wraps POST /wechatweb/callRecord/queryWeChatVoiceList with page-number pagination (reuses paginate_by_page); normalizes records to the PhoneCall column shape with channel='voice_call'. Field mapping per phase0-findings §4: id->call_id, duration(str s)->call_duration(int), startTime(YYYY-MM-DD HH:MM:SS)->start_call_time(ms epoch), callStatus-> call_status, talker->call_number, talkerNickName->plan_customer_name, ossFileName->record_file (may be null). isSend (0=in/1=out) is inverted to incoming_call (1=in/0=out) so the column's meaning matches the phone_call channel. - upsert_voice_call: idempotent Core upsert into PhoneCall by call_id (sqlite/mysql dialect-aware). Adds call_duration/start_call_time to the _INT_FIELDS coercion net (scraper already emits ints; defensive for hand-built records). - Group-message test (Task 2.3): locks in that scrape_wechat_messages already handles chatroom conversations (friendId=@chatroom, roomid set, talker=member wxid) — yields room_id populated and talker_id = the sending member. No scraper change required. Suite: 123 -> 130 passing. ruff clean. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -24,6 +24,11 @@ Endpoints covered today:
|
||||
and yields the per-staff-wechat conversation list (M2). Used by the sync
|
||||
orchestrator to decide which conversations have recent traffic to pull.
|
||||
Page-number pagination; the owning ``wechatId`` is sent in every request.
|
||||
* :func:`scrape_wechat_voice_calls` — wraps ``POST /wechatweb/callRecord/queryWeChatVoiceList``
|
||||
and yields normalized WeChat voice-call records (M2). Each record is mapped
|
||||
to the ``PhoneCall`` column shape with ``channel="voice_call"``; the
|
||||
recording URL (``ossFileName``) lands on ``record_file``. Page-number
|
||||
pagination.
|
||||
|
||||
All yield dicts (not ORM). Persistence is the caller's job (the
|
||||
:mod:`app.storage.repos` module bridges these dicts to the ORM tables). Each
|
||||
@@ -33,8 +38,9 @@ dict carries:
|
||||
for friends, :func:`_normalize_message` for messages,
|
||||
:func:`_normalize_department` for departments,
|
||||
:func:`_normalize_wxuser_account` for accounts,
|
||||
:func:`_normalize_chat_group` for groups, and
|
||||
:func:`_normalize_talker` for talkers),
|
||||
:func:`_normalize_chat_group` for groups,
|
||||
:func:`_normalize_talker` for talkers, and
|
||||
:func:`_normalize_voice_call` for voice calls),
|
||||
* ``channel == "wechat"`` (only on communication-channel entities; the
|
||||
organization entities ``Department`` / ``WechatAccount`` carry no channel),
|
||||
* ``raw_json`` — the original upstream record, verbatim, so nothing captured
|
||||
@@ -52,6 +58,7 @@ disambiguations where the source has two fields for one column:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
@@ -66,6 +73,7 @@ __all__ = [
|
||||
"scrape_wechat_groups",
|
||||
"scrape_wechat_messages",
|
||||
"scrape_wechat_talkers",
|
||||
"scrape_wechat_voice_calls",
|
||||
]
|
||||
|
||||
FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es"
|
||||
@@ -74,6 +82,7 @@ DEPT_TREE_PATH = "/usercenter/dept/tree"
|
||||
WXUSER_LIST_PATH = "/wechatweb/wxUser/list"
|
||||
CHATROOM_LIST_PATH = "/wechatweb/chatroom/getChatroomList"
|
||||
TALKER_LIST_PATH = "/wechatweb/wxFriends/talkerList"
|
||||
VOICE_LIST_PATH = "/wechatweb/callRecord/queryWeChatVoiceList"
|
||||
|
||||
# Upstream sends form-encoded bodies for ``msg/chats`` (NOT JSON), so we have
|
||||
# to override the client's default JSON content-type on every request.
|
||||
@@ -971,3 +980,190 @@ async def scrape_wechat_talkers(
|
||||
wechat_id=wechat_id,
|
||||
):
|
||||
yield _normalize_talker(record, wechat_id)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Voice-call scraper (M2, Task 2.4) — POST /wechatweb/callRecord/queryWeChatVoiceList
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
def _parse_voice_total(resp: Any) -> int | None:
|
||||
"""Read the declared row count from the ``queryWeChatVoiceList`` envelope.
|
||||
|
||||
Shape: ``data.total`` (an int). Returns ``None`` if the path is absent so
|
||||
the paginator falls back to the empty-page stop signal. (Live ``total``
|
||||
is 333 per phase0-findings.md §4.)
|
||||
"""
|
||||
try:
|
||||
body = resp.json()
|
||||
except (AttributeError, ValueError):
|
||||
return None
|
||||
data = body.get("data") if isinstance(body, dict) else None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
total = data.get("total")
|
||||
return int(total) if isinstance(total, (int, float)) and not isinstance(
|
||||
total, bool
|
||||
) else None
|
||||
|
||||
|
||||
def _parse_voice_rows(resp: Any) -> list[dict[str, Any]]:
|
||||
"""Extract the voice-call list from the ``queryWeChatVoiceList`` envelope.
|
||||
|
||||
Shape: ``data.data`` (a list of records). Missing / non-list values
|
||||
collapse to an empty list so the paginator sees "empty page".
|
||||
"""
|
||||
try:
|
||||
body = resp.json()
|
||||
except (AttributeError, ValueError):
|
||||
return []
|
||||
data = body.get("data") if isinstance(body, dict) else None
|
||||
if not isinstance(data, dict):
|
||||
return []
|
||||
rows = data.get("data")
|
||||
return rows if isinstance(rows, list) else []
|
||||
|
||||
|
||||
def _build_voice_body(page_num: int, page_size: int) -> dict[str, Any]:
|
||||
"""Build the ``queryWeChatVoiceList`` request body for a given page.
|
||||
|
||||
The live capture carries no filter parameters — just the page slice.
|
||||
"""
|
||||
return {"pageNum": page_num, "pageSize": page_size}
|
||||
|
||||
|
||||
def _parse_voice_start_time(value: Any) -> int | None:
|
||||
"""Coerce a voice-call ``startTime`` to a ms-epoch int (the PhoneCall column type).
|
||||
|
||||
The voice-call endpoint reports ``startTime`` as a ``YYYY-MM-DD HH:MM:SS``
|
||||
datetime string (confirmed by the Task 2.4 spec). To be robust against the
|
||||
upstream also emitting a numeric ms-epoch (as other call endpoints do), the
|
||||
following shapes are accepted:
|
||||
|
||||
* ``int`` / ``float`` — treated as ms-epoch already, returned as ``int``.
|
||||
* numeric ``str`` (e.g. ``"1700000000123"``) -> ``int``.
|
||||
* ``"YYYY-MM-DD HH:MM:SS"`` datetime string -> parsed and converted to
|
||||
ms-epoch. The CRM is a Chinese deployment and the wall-clock times are
|
||||
presumably CST (UTC+8); this scraper parses them as **naive UTC** for
|
||||
determinism — a documented simplification. If production later needs
|
||||
CST-accurate buckets, swap the ``tzinfo`` attached here.
|
||||
* ``None`` / empty / unparseable -> ``None`` (no spurious value stored).
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return int(value)
|
||||
if isinstance(value, (int, float)):
|
||||
return int(value)
|
||||
if isinstance(value, str):
|
||||
s = value.strip()
|
||||
if not s:
|
||||
return None
|
||||
# Numeric string -> ms-epoch int (no datetime round-trip).
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
pass
|
||||
# Documented "YYYY-MM-DD HH:MM:SS" datetime string -> ms-epoch.
|
||||
# ``.replace(tzinfo=...)`` is chained on the strptime so ruff's
|
||||
# DTZ007 sees the naive datetime being made aware (and so the
|
||||
# resulting epoch is well-defined).
|
||||
try:
|
||||
parsed = dt.datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace(
|
||||
tzinfo=dt.UTC
|
||||
)
|
||||
except ValueError:
|
||||
return None
|
||||
return int(parsed.timestamp() * 1000)
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_voice_call(record: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Map one upstream ``queryWeChatVoiceList`` record to the ``PhoneCall`` shape.
|
||||
|
||||
Adds ``channel="voice_call"`` and preserves the original record under
|
||||
``raw_json``. Field mapping per phase0-findings.md §4 + the PhoneCall
|
||||
column set:
|
||||
|
||||
* ``id`` -> ``call_id`` (PK)
|
||||
* ``duration`` -> ``call_duration`` (string seconds coerced to int)
|
||||
* ``startTime`` -> ``start_call_time`` (datetime string -> ms epoch;
|
||||
see :func:`_parse_voice_start_time`)
|
||||
* ``callStatus`` -> ``call_status`` (1=connected w/ recording,
|
||||
2=no answer)
|
||||
* ``isSend`` -> ``incoming_call`` with INVERTED polarity.
|
||||
Voice-call ``isSend`` is ``0=incoming / 1=outgoing``; ``PhoneCall.
|
||||
incoming_call`` (matching the ``phone_call`` channel) is
|
||||
``1=incoming / 0=outgoing``. To keep the column's meaning identical
|
||||
across both call channels we store ``incoming_call = 1 - isSend``
|
||||
(``None`` preserved when ``isSend`` is missing/non-numeric).
|
||||
* ``userId`` -> ``user_id`` (owning staff)
|
||||
* ``userName`` -> ``user_id_name``
|
||||
* ``talker`` -> ``call_number`` (the customer wxid; for a room
|
||||
call this is the ``@chatroom`` id — there is no separate member-sender
|
||||
field on the voice-call record)
|
||||
* ``talkerNickName`` -> ``plan_customer_name``
|
||||
* ``ossFileName`` -> ``record_file`` (the recording URL; may be null)
|
||||
|
||||
Non-applicable PhoneCall columns (``ring_time`` / ``sim_phone`` /
|
||||
``sim_number`` / ``imei`` / ``is_upload`` / ``download_count`` / ``ai_asr``
|
||||
/ ``qc_status`` / ``department_id``) are left ``None`` — the voice-call
|
||||
payload carries none of them. The owning ``weChatId`` has no matching
|
||||
PhoneCall column (there is no ``owner_account_id`` on this model), so it
|
||||
is preserved only in ``raw_json``. ``None``-safe on every mapped field.
|
||||
"""
|
||||
is_send = _coerce_int_or_none(record.get("isSend"))
|
||||
# Flip isSend (0=incoming/1=outgoing) to incoming_call (1=incoming/0=outgoing).
|
||||
incoming_call = None if is_send is None else 1 - is_send
|
||||
return {
|
||||
"call_id": record.get("id"),
|
||||
"channel": "voice_call",
|
||||
"start_call_time": _parse_voice_start_time(record.get("startTime")),
|
||||
"call_duration": _coerce_int_or_none(record.get("duration")),
|
||||
"call_status": _coerce_int_or_none(record.get("callStatus")),
|
||||
"incoming_call": incoming_call,
|
||||
"user_id": record.get("userId"),
|
||||
"user_id_name": record.get("userName"),
|
||||
"call_number": record.get("talker"),
|
||||
"plan_customer_name": record.get("talkerNickName"),
|
||||
"record_file": record.get("ossFileName"),
|
||||
"version": record.get("version"),
|
||||
"raw_json": record,
|
||||
}
|
||||
|
||||
|
||||
async def scrape_wechat_voice_calls(
|
||||
client: YunkeClient,
|
||||
*,
|
||||
page_size: int = 50,
|
||||
) -> AsyncIterator[dict[str, Any]]:
|
||||
"""Yield normalized WeChat voice-call records from ``queryWeChatVoiceList``.
|
||||
|
||||
Walks ``POST /wechatweb/callRecord/queryWeChatVoiceList`` page by page
|
||||
until the upstream reports no more rows (empty ``data.data`` array — the
|
||||
primary stop signal), mapping each record to the ``PhoneCall`` column
|
||||
shape with ``channel="voice_call"`` plus a verbatim ``raw_json`` copy.
|
||||
|
||||
The envelope is the same flat shape as ``wxUser/list`` (no
|
||||
``basicResult`` wrapper): the row list and pagination fields sit directly
|
||||
under ``data`` (``data.data`` / ``data.total`` / ``data.pageNum`` /
|
||||
``data.pageSize``).
|
||||
|
||||
Parameters
|
||||
----------
|
||||
client:
|
||||
Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are
|
||||
handled by the client.
|
||||
page_size:
|
||||
Page size to request. Defaults to 50 (a sane middle ground between
|
||||
request count and per-response payload size for this endpoint).
|
||||
"""
|
||||
async for record in paginate_by_page(
|
||||
client,
|
||||
VOICE_LIST_PATH,
|
||||
page_size=page_size,
|
||||
parse_total=_parse_voice_total,
|
||||
parse_rows=_parse_voice_rows,
|
||||
build_body=_build_voice_body,
|
||||
):
|
||||
yield _normalize_voice_call(record)
|
||||
|
||||
Reference in New Issue
Block a user