diff --git a/yunke-server/app/scraper/call.py b/yunke-server/app/scraper/call.py new file mode 100644 index 0000000..117d025 --- /dev/null +++ b/yunke-server/app/scraper/call.py @@ -0,0 +1,376 @@ +"""Phone-call + SMS scrapers (M4, ``/pc/`` CRM module). + +Endpoints covered: + +* :func:`scrape_phone_calls` — wraps ``POST /pc/callLog/list`` and yields + normalized phone-call dicts mapped to :class:`app.db.models.PhoneCall` with + ``channel="phone_call"``. Page-number pagination. +* :func:`scrape_sms` — wraps ``POST /pc/customer/smsLogsList`` and yields + normalized SMS dicts mapped to :class:`app.db.models.Sms` with + ``channel="sms"`` (the model's default). Page-number pagination. + +Both reuse :func:`app.scraper.base.paginate_by_page` and emit snake_case dicts +plus a verbatim ``raw_json`` copy, mirroring :mod:`app.scraper.wechat`. + +Envelope shapes (Phase-0 confirmed, data dictionary §4): + +* ``callLog/list``:: + + {"success": true, "data": {"data": [...], "page": .., "pageCount": .., + "totalCount": .., "phoneHidden": .., + "hideRecordPlay": ..}} + + Rows under ``data.data``; declared row count on ``data.totalCount``; page + count on ``data.pageCount``. The paginator consumes the row-count ceiling + via ``parse_total`` and otherwise relies on the empty-page stop signal. + +* ``smsLogsList``:: + + {"success": true, "data": {"data": [...], "dataCount": .., "power": ..}} + + Rows under ``data.data``; declared row count on ``data.dataCount`` (NOT + ``totalCount``); **no** ``pageCount``. The paginator relies solely on the + empty-page stop signal (the over-fetch ceiling keyed off ``dataCount`` + still applies). + +Field mapping notes: + +* Phone-call ``incomingCall`` is ``1=incoming / 0=outgoing`` — the SAME + polarity as ``PhoneCall.incoming_call``. It is therefore stored DIRECTLY, + with NO polarity inversion (unlike the voice-call channel's ``isSend``, + which is inverted in :func:`app.scraper.wechat._normalize_voice_call`). +* Phone-call ``callSeconds`` (int) is preferred over the legacy + ``callDuration`` ``"M'SS\""`` string for ``call_duration``. +* Phone-call ``startCallTime`` (``YYYY-MM-DD HH:MM:SS``) and SMS + ``createTime`` (``YYYY-MM-DD HH:MM``, one precision step shorter) are both + parsed to ms-epoch via the shared + :func:`app.scraper.parsers.parse_datetime_to_ms`, treated as naive UTC for + determinism (documented simplification — see the helper's docstring). +* SMS ``smsSub`` (a short preview) is carried on ``content_preview``; an + empty / falsy preview is normalized to ``None`` so the column reads as + "no preview" rather than a literal empty string. + +Auth is the same crm2 session cookie every other scraper uses (Phase-0 §1 +confirmed cookie-only auth suffices for every data endpoint). +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +from app.scraper.base import paginate_by_page +from app.scraper.parsers import coerce_int_or_none, parse_datetime_to_ms +from app.session.yunke_client import YunkeClient + +__all__ = ["scrape_phone_calls", "scrape_sms"] + +CALL_LOG_PATH = "/pc/callLog/list" +SMS_LOG_PATH = "/pc/customer/smsLogsList" + +# Datetime strptime patterns for the two endpoints. The callLog's +# ``startCallTime`` carries seconds; the smsLogsList's ``createTime`` is one +# precision step shorter (no seconds — default to :00 via the pattern). +_CALL_DATETIME_FORMATS = ["%Y-%m-%d %H:%M:%S"] +_SMS_DATETIME_FORMATS = ["%Y-%m-%d %H:%M"] + + +# =========================================================================== +# Phone-call scraper (M4) — POST /pc/callLog/list +# =========================================================================== + + +def _parse_call_total(resp: Any) -> int | None: + """Read the declared row count from the ``callLog/list`` envelope. + + Shape: ``data.totalCount`` (an int). Returns ``None`` if the path is + absent so the paginator falls back to the empty-page stop signal. + (Live ``totalCount`` is 6047 per the data dictionary §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("totalCount") + if isinstance(total, bool): + return None + return int(total) if isinstance(total, (int, float)) else None + + +def _parse_call_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the call list from the ``callLog/list`` 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_call_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``callLog/list`` 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 _normalize_phone_call(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``callLog/list`` record to the ``PhoneCall`` column shape. + + Adds ``channel="phone_call"`` and preserves the original record under + ``raw_json``. Field mapping per data dictionary §4 + the PhoneCall column + set: + + * ``id`` -> ``call_id`` (PK) + * ``startCallTime`` -> ``start_call_time`` (``YYYY-MM-DD HH:MM:SS`` -> + ms epoch; see :func:`parse_datetime_to_ms`) + * ``callSeconds`` -> ``call_duration`` (int seconds, PREFERRED over + the legacy ``callDuration`` ``"M'SS\""`` string) + * ``callStatus`` -> ``call_status`` (0=未接 / 2=接通) + * ``incomingCall`` -> ``incoming_call`` (1=呼入 incoming / 0=呼出 + outgoing — stored DIRECTLY, NO inversion; this API's polarity already + matches ``PhoneCall.incoming_call``) + * ``userId`` / ``userIdName`` -> owning staff pair + * ``departmentId`` -> ``department_id`` + * ``callNumber`` -> ``call_number`` (the customer phone) + * ``simPhone`` / ``simNumber`` / ``imei`` -> SIM/IMEI metadata + * ``recordFile`` -> ``record_file`` (OSS recording URL; may be null) + * ``isUpload`` -> ``is_upload`` (bool field; repo coerces) + * ``downloadCount`` -> ``download_count`` + * ``aiAsr`` -> ``ai_asr`` (ASR transcription text; may be null) + * ``qcStatus`` -> ``qc_status`` + * ``planCustomerName`` -> ``plan_customer_name`` + + ``call_duration`` prefers ``callSeconds`` (int) and falls back to + ``callDuration`` only when ``callSeconds`` is missing — the live capture + sends both, but ``callSeconds`` is the cleaner (already-int) source. + Coerced to ``int | None`` so a numeric string still lands cleanly. + + The non-applicable ``ring_time`` column is left ``None`` — the callLog + payload carries only ``callSeconds`` (talk time), not a separate ring + duration. ``None``-safe on every mapped field. + """ + # Prefer callSeconds (int seconds) over callDuration ("M'SS\"" string). + # Fall back to callDuration only when callSeconds is missing; coerce both + # via the shared int coercion so a numeric string still lands cleanly. + call_seconds = record.get("callSeconds") + if call_seconds is None: + call_seconds = record.get("callDuration") + + return { + "call_id": record.get("id"), + "channel": "phone_call", + "start_call_time": parse_datetime_to_ms( + record.get("startCallTime"), _CALL_DATETIME_FORMATS + ), + "call_duration": coerce_int_or_none(call_seconds), + "call_status": coerce_int_or_none(record.get("callStatus")), + # NOTE: stored DIRECTLY — this API's incomingCall (1=incoming/0=outgoing) + # already matches PhoneCall.incoming_call polarity (no inversion). + "incoming_call": coerce_int_or_none(record.get("incomingCall")), + "user_id": record.get("userId"), + "user_id_name": record.get("userIdName"), + "department_id": record.get("departmentId"), + "call_number": record.get("callNumber"), + "sim_phone": record.get("simPhone"), + "sim_number": record.get("simNumber"), + "imei": record.get("imei"), + "record_file": record.get("recordFile"), + "is_upload": record.get("isUpload"), + "download_count": coerce_int_or_none(record.get("downloadCount")), + "ai_asr": record.get("aiAsr"), + "qc_status": coerce_int_or_none(record.get("qcStatus")), + "plan_customer_name": record.get("planCustomerName"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_phone_calls( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized phone-call records from ``POST /pc/callLog/list``. + + Walks the endpoint 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="phone_call"`` plus a + verbatim ``raw_json`` copy. + + The envelope keeps the row list under ``data.data`` (same as ``wxUser/ + list``) but the declared row count is on ``data.totalCount`` (NOT + ``data.total``) and the page count is on ``data.pageCount``. The paginator + consumes ``totalCount`` for its over-fetch ceiling and otherwise relies on + the empty-page stop signal. + + 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, + CALL_LOG_PATH, + page_size=page_size, + parse_total=_parse_call_total, + parse_rows=_parse_call_rows, + build_body=_build_call_body, + ): + yield _normalize_phone_call(record) + + +# =========================================================================== +# SMS scraper (M4) — POST /pc/customer/smsLogsList +# =========================================================================== + + +def _parse_sms_total(resp: Any) -> int | None: + """Read the declared row count from the ``smsLogsList`` envelope. + + Shape: ``data.dataCount`` (an int) — note ``dataCount``, NOT + ``totalCount`` (and there is no ``pageCount`` on this envelope). Returns + ``None`` if the path is absent so the paginator falls back to the + empty-page stop signal. (Live ``dataCount`` is 53 per data dictionary §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("dataCount") + if isinstance(total, bool): + return None + return int(total) if isinstance(total, (int, float)) else None + + +def _parse_sms_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the SMS list from the ``smsLogsList`` 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_sms_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``smsLogsList`` 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 _normalize_sms(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``smsLogsList`` record to the ``Sms`` column shape. + + Adds ``channel="sms"`` (the model's default, but carried explicitly so the + repo upsert is independent of the column-default application order) and + preserves the original record under ``raw_json``. Field mapping per data + dictionary §4 + the Sms column set: + + * ``id`` -> ``sms_id`` (PK) + * ``customerPhone`` -> ``customer_phone`` + * ``sms`` -> ``content`` (full SMS text) + * ``smsSub`` -> ``content_preview`` (short preview; an EMPTY + / falsy preview is normalized to ``None`` so the column reads as "no + preview" rather than a literal empty string) + * ``sender`` -> ``sender`` + * ``direction`` -> ``direction`` (0=收 incoming / 1=发 outgoing) + * ``userId`` -> ``user_id`` (owning staff) + * ``senderDepartment`` / ``senderDepartmentId`` -> dept name / id + * ``receiveName`` -> ``receive_name`` + * ``createTime`` -> ``create_time`` (``YYYY-MM-DD HH:MM`` -> ms + epoch; seconds default to :00 via the strptime pattern) + * ``simPhone`` / ``simNumber`` -> SIM metadata + * ``source`` -> ``source`` + + ``None``-safe on every mapped field. + """ + return { + "sms_id": record.get("id"), + "channel": "sms", + "customer_phone": record.get("customerPhone"), + "content": record.get("sms"), + # Empty preview -> None (so the column reads "no preview" rather than + # a literal empty string). The repo does not normalize this; the + # scraper owns the falsy->None collapse. + "content_preview": record.get("smsSub") or None, + "sender": record.get("sender"), + "direction": coerce_int_or_none(record.get("direction")), + "user_id": record.get("userId"), + "sender_department": record.get("senderDepartment"), + "sender_department_id": record.get("senderDepartmentId"), + "receive_name": record.get("receiveName"), + "create_time": parse_datetime_to_ms( + record.get("createTime"), _SMS_DATETIME_FORMATS + ), + "sim_phone": record.get("simPhone"), + "sim_number": record.get("simNumber"), + "source": record.get("source"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_sms( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized SMS records from ``POST /pc/customer/smsLogsList``. + + Walks the endpoint page by page until the upstream reports no more rows + (empty ``data.data`` array — the primary stop signal), mapping each record + to the ``Sms`` column shape with ``channel="sms"`` plus a verbatim + ``raw_json`` copy. + + The envelope keeps the row list under ``data.data`` but the declared row + count is on ``data.dataCount`` (NOT ``totalCount`` and NOT ``total``) and + there is **no** ``pageCount``. The paginator therefore relies on + ``dataCount`` only for the over-fetch ceiling and otherwise on the + empty-page stop signal — the same shape as the talker-list endpoint. + + 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, + SMS_LOG_PATH, + page_size=page_size, + parse_total=_parse_sms_total, + parse_rows=_parse_sms_rows, + build_body=_build_sms_body, + ): + yield _normalize_sms(record) diff --git a/yunke-server/app/scraper/parsers.py b/yunke-server/app/scraper/parsers.py index b9d7928..184187a 100644 --- a/yunke-server/app/scraper/parsers.py +++ b/yunke-server/app/scraper/parsers.py @@ -21,9 +21,16 @@ without the scraper taking a DB dependency. from __future__ import annotations +import datetime as dt +from collections.abc import Sequence from typing import Any -__all__ = ["coerce_int_or_none", "normalize_wechat_type", "normalize_whatsapp_type"] +__all__ = [ + "coerce_int_or_none", + "normalize_wechat_type", + "normalize_whatsapp_type", + "parse_datetime_to_ms", +] # Phase-0 confirmed WeChat message type codes (see # docs/superpowers/specs/phase0-findings.md §2). Captured by live inspection @@ -133,3 +140,61 @@ def coerce_int_or_none(value: Any) -> int | None: except ValueError: return None return None + + +def parse_datetime_to_ms( + value: Any, formats: Sequence[str] +) -> int | None: + """Coerce a datetime-shaped upstream value to a ms-epoch ``int``. + + Centralizes the datetime -> ms-epoch parsing shared by the call scrapers + (``PhoneCall.start_call_time`` from ``callLog.startCallTime`` / + ``Sms.create_time`` from ``smsLogsList.createTime``) and the voice-call + scraper (``PhoneCall.start_call_time`` from ``startTime``). The accepted + shapes: + + * ``int`` / ``float`` — treated as ms-epoch already, returned as ``int``. + * numeric ``str`` (e.g. ``"1700000000123"``) -> ``int`` (no datetime + round-trip). + * datetime ``str`` matching one of ``formats`` -> parsed with + ``datetime.strptime`` 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, attach + a CST tzinfo to the parsed datetime before the timestamp call). The + ``formats`` are tried in order; the first match wins. ``.replace( + tzinfo=UTC)`` is chained on the strptime result so ruff's DTZ007 sees + the naive datetime being made aware (and so the resulting epoch is + well-defined). + * ``None`` / empty / unparseable by any format -> ``None`` (no spurious + value stored). + + ``formats`` is the ordered list of ``strptime`` patterns to attempt. The + call scraper passes ``["%Y-%m-%d %H:%M:%S"]`` (``startCallTime``); the SMS + scraper passes ``["%Y-%m-%d %H:%M"]`` (``createTime``, one precision step + shorter — no seconds). + """ + 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 + # Try each strptime format in order; first match wins. + for fmt in formats: + try: + parsed = dt.datetime.strptime(s, fmt).replace(tzinfo=dt.UTC) + except ValueError: + continue + return int(parsed.timestamp() * 1000) + return None + return None diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index 46a05f6..0b4e6df 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -58,12 +58,11 @@ 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 from app.scraper.base import paginate_by_cursor, paginate_by_page -from app.scraper.parsers import normalize_wechat_type +from app.scraper.parsers import normalize_wechat_type, parse_datetime_to_ms from app.session.yunke_client import YunkeClient __all__ = [ @@ -1035,47 +1034,18 @@ def _build_voice_body(page_num: int, page_size: int) -> dict[str, Any]: 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: + Thin wrapper over the shared :func:`app.scraper.parsers.parse_datetime_to_ms` + so the voice-call scraper speaks the same datetime -> ms-epoch dialect as + the phone-call + SMS scrapers while keeping its capture-specific name. The + voice-call endpoint reports ``startTime`` as a ``YYYY-MM-DD HH:MM:SS`` + datetime string (confirmed by the Task 2.4 spec); the shared helper also + accepts a numeric ms-epoch (int / float / numeric str) for robustness. - * ``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). + See :func:`parse_datetime_to_ms` for the full accepted-shapes contract and + the documented naive-UTC simplification (the CRM wall-clock times are + presumably CST; this parses them as UTC for determinism). """ - 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 + return parse_datetime_to_ms(value, ["%Y-%m-%d %H:%M:%S"]) def _normalize_voice_call(record: dict[str, Any]) -> dict[str, Any]: diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index ffbde0f..8283849 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -61,6 +61,7 @@ from app.db.models import ( Media, Message, PhoneCall, + Sms, Staff, WechatAccount, WhatsappAccount, @@ -73,6 +74,8 @@ __all__ = [ "upsert_department", "upsert_media", "upsert_messages", + "upsert_phone_call", + "upsert_sms", "upsert_staff", "upsert_staff_from_account", "upsert_voice_call", @@ -104,6 +107,7 @@ _INT_FIELDS: frozenset[str] = frozenset( "call_duration", "start_call_time", "qc_status", + "create_time", } ) @@ -505,6 +509,42 @@ async def upsert_voice_call(session: AsyncSession, rec: dict[str, Any]) -> None: await _upsert_one(session, PhoneCall, rec) +# --------------------------------------------------------------------------- +# PhoneCall (phone_call) + Sms upsert (M4) +# --------------------------------------------------------------------------- + + +async def upsert_phone_call(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`PhoneCall` from a phone-call record (M4). + + Same ``call_id`` twice -> one row, with non-PK fields refreshed from the + later record. PK = ``call_id`` (the source ``id``). The phone-call channel + (``channel="phone_call"``) shares the :class:`PhoneCall` table with the + voice-call channel; the per-row ``channel`` value disambiguates them. + + Applies the documented int coercions (``call_duration`` / + ``start_call_time`` / ``call_status`` / ``incoming_call`` / ``qc_status`` + -> int when they arrive as numeric strings; the phone-call scraper already + emits ints, but a hand-built record may not). ``is_upload`` is a bool + field and goes through :func:`_coerce_bool` like the other bool columns. + """ + await _upsert_one(session, PhoneCall, rec) + + +async def upsert_sms(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`Sms` (PK = ``sms_id``) (M4). + + Same ``sms_id`` twice -> one row, with non-PK fields refreshed from the + later record. Applies the documented int coercions (``direction`` / + ``create_time`` -> int when they arrive as numeric strings; the SMS + scraper already emits ints, but a hand-built record may not). The + ``channel`` column defaults to ``"sms"`` at the model level; the scraper + carries it explicitly so the upsert is independent of the column-default + application order. + """ + await _upsert_one(session, Sms, rec) + + # --------------------------------------------------------------------------- # Media upsert (M2, Task 2.6) # --------------------------------------------------------------------------- diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 54b3430..4c75500 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -46,6 +46,7 @@ from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import SyncRun +from app.scraper.call import scrape_phone_calls, scrape_sms from app.scraper.media import download_message_media from app.scraper.wechat import ( scrape_departments, @@ -62,6 +63,8 @@ from app.storage.repos import ( upsert_customer, upsert_department, upsert_messages, + upsert_phone_call, + upsert_sms, upsert_voice_call, upsert_wechat_account, ) @@ -73,6 +76,8 @@ __all__ = [ "SyncResult", "sync_all_wechat", "sync_departments", + "sync_phone_calls", + "sync_sms", "sync_wechat_accounts", "sync_wechat_friends", "sync_wechat_groups", @@ -93,6 +98,10 @@ ENTITY_WECHAT_FRIEND = "wechat_friend" ENTITY_WECHAT_GROUP = "wechat_group" ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call" ENTITY_MESSAGE = "message" +# M4 (/pc/ CRM module) — phone calls + SMS share no channel column with the +# wechat-family entities; they track their own SyncRun / SyncState rows. +ENTITY_PHONE_CALL = "phone_call" +ENTITY_SMS = "sms" CHANNEL_WECHAT = "wechat" @@ -432,6 +441,54 @@ async def sync_wechat_voice_calls( ) +async def sync_phone_calls( + session: AsyncSession, client: YunkeClient +) -> SyncResult: + """Full re-scrape of all phone-call records (``/pc/callLog/list``) and + idempotent upsert into :class:`PhoneCall` (``channel="phone_call"``). + + Mirrors :func:`sync_wechat_voice_calls`: the callLog endpoint has no + ``since`` param, so "incremental" = re-scrape everything and rely on the + repo's idempotent Core upsert (same ``call_id`` twice -> one refreshed + row). Records a SyncRun row (``entity_type='phone_call'``) and stamps the + watermark's ``last_synced_at`` (no cursor advance — there is no upstream + window to close). A scraper failure is recorded as a failed SyncRun and a + failed SyncResult is returned (no re-raise), so a phone-call scrape that + 500s does NOT abort a larger batch. + """ + return await _sync_full_rescrape( + session, + entity_type=ENTITY_PHONE_CALL, + channel=None, + scope_id=None, + scrape=lambda: scrape_phone_calls(client), + upsert=upsert_phone_call, + ) + + +async def sync_sms( + session: AsyncSession, client: YunkeClient +) -> SyncResult: + """Full re-scrape of all SMS records (``/pc/customer/smsLogsList``) and + idempotent upsert into :class:`Sms`. + + Mirrors :func:`sync_phone_calls`: the smsLogsList endpoint has no + ``since`` param, so "incremental" = re-scrape everything and rely on the + repo's idempotent Core upsert (same ``sms_id`` twice -> one refreshed + row). Records a SyncRun row (``entity_type='sms'``) and stamps the + watermark's ``last_synced_at``. A scraper failure is recorded as a failed + SyncRun + SyncResult without re-raising. + """ + return await _sync_full_rescrape( + session, + entity_type=ENTITY_SMS, + channel=None, + scope_id=None, + scrape=lambda: scrape_sms(client), + upsert=upsert_sms, + ) + + # --------------------------------------------------------------------------- # Coordinator # --------------------------------------------------------------------------- diff --git a/yunke-server/tests/cassettes/callLog_fixture.py b/yunke-server/tests/cassettes/callLog_fixture.py new file mode 100644 index 0000000..58e2710 --- /dev/null +++ b/yunke-server/tests/cassettes/callLog_fixture.py @@ -0,0 +1,210 @@ +"""Fixture payloads for ``POST /pc/callLog/list`` (M4 phone-call scraper). + +These records mirror the real field names confirmed by Phase-0 + the data +dictionary §4 (电话记录 phone_call): ``id, startCallTime(YYYY-MM-DD HH:MM:SS), +callDuration("M'SS\"")/callSeconds(int秒), callStatus(0未接/2接通), +incomingCall(0呼出/1呼入), userId/userIdName(员工), departmentId, callNumber, +simPhone, simNumber, imei, recordFile(录音OSS URL, may be null), isUpload, +downloadCount, aiAsr, qcStatus, planCustomerName, version``. Live ``totalCount`` +is 6047 across ``pageCount=303`` pages. + +The envelope differs from the wechat ``data.data`` / ``data.total`` shape: the +row list sits under ``data.data`` (same) but the declared row count is +``data.totalCount`` (NOT ``data.total``) and the page count is ``data.pageCount``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "page":.., "pageCount":.., "totalCount":.., + "phoneHidden":false, "hideRecordPlay":false}} + +The cassette ships four records across two pages so a test can drive the +page-number paginator across the empty-page stop signal: + +* page 1 -> 2 of the 4 fixture records, ``totalCount=4`` +* page 2 -> the other 2 records, ``totalCount=4`` +* (page 3 -> empty ``data`` array, served inline by the test) + +Records deliberately exercise the documented mapping edges: + +* ``pc_2001`` — outgoing connected call (``incomingCall=0``; ``callStatus=2``; + ``callSeconds=151``; non-null ``recordFile``; full SIM/IMEI metadata). +* ``pc_2002`` — incoming no-answer call with NO recording (``incomingCall=1``; + ``callStatus=0``; ``recordFile=null``). +* ``pc_2003`` — incoming connected call with a recording + AI ASR + QC data. +* ``pc_2004`` — outgoing connected call (exercises the other polarity pair). + +NOTE on ``incomingCall`` polarity: this API's ``incomingCall`` is +``1=incoming / 0=outgoing`` — the SAME polarity as ``PhoneCall.incoming_call`` +(no inversion needed, unlike the voice-call channel's ``isSend``). +""" + +# The module name mirrors the upstream endpoint name (``callLog``); the M4 spec +# uses this naming family. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped phone-call records (4 total across 2 pages). +# --------------------------------------------------------------------------- + +CALL_RECORDS: list[dict[str, Any]] = [ + { + # Outgoing connected call with a recording. + "id": "pc_2001", + "startCallTime": "2026-01-15 10:30:00", # -> ms epoch UTC + "callSeconds": 151, # preferred over callDuration string + "callDuration": "2'31\"", # legacy "M'SS\"" string; scraper prefers callSeconds + "callStatus": 2, # 2=接通 (connected) + "incomingCall": 0, # 0=呼出 (outgoing) -> incoming_call=0 + "userId": "staff_user_001", + "userIdName": "Alice Wang", + "departmentId": "dept_sales_east", + "callNumber": "8613800000001", + "simPhone": "8613800000111", + "simNumber": "8986000112345678901", + "imei": "356938035643801", + "recordFile": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/phone-call/" + "8m0atd/imei001/1700000000000_1700000151000.mp3" + ), + "isUpload": 1, + "downloadCount": 3, + "aiAsr": "您好，请问您方便了解一下我们的产品吗？", + "qcStatus": 1, + "planCustomerName": "VIP Customer A", + "version": "v1.20260731.pc2001", + }, + { + # Incoming no-answer call, no recording. + "id": "pc_2002", + "startCallTime": "2026-01-15 11:00:00", + "callSeconds": 0, + "callDuration": "0'0\"", + "callStatus": 0, # 0=未接 (no answer) + "incomingCall": 1, # 1=呼入 (incoming) -> incoming_call=1 + "userId": "staff_user_001", + "userIdName": "Alice Wang", + "departmentId": "dept_sales_east", + "callNumber": "8613800000002", + "simPhone": "8613800000111", + "simNumber": "8986000112345678901", + "imei": "356938035643801", + "recordFile": None, # no recording -> record_file None + "isUpload": 0, + "downloadCount": 0, + "aiAsr": None, + "qcStatus": 0, + "planCustomerName": "Lead B", + "version": "v1.20260731.pc2002", + }, + { + # Incoming connected call with recording + AI ASR text. + "id": "pc_2003", + "startCallTime": "2026-01-15 12:00:00", + "callSeconds": 45, + "callDuration": "0'45\"", + "callStatus": 2, # connected + "incomingCall": 1, # incoming -> incoming_call=1 + "userId": "staff_user_002", + "userIdName": "Bob Chen", + "departmentId": "dept_sales_west", + "callNumber": "8613800000003", + "simPhone": "8613800000222", + "simNumber": "8986000223456789012", + "imei": "356938035643802", + "recordFile": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/phone-call/" + "8m0atd/imei002/1700000003000_1700000045000.mp3" + ), + "isUpload": 1, + "downloadCount": 1, + "aiAsr": "好的，我考虑一下。", + "qcStatus": 2, + "planCustomerName": "Customer C", + "version": "v1.20260731.pc2003", + }, + { + # Outgoing connected call (the other polarity pair). + "id": "pc_2004", + "startCallTime": "2026-01-16 09:00:00", + "callSeconds": 207, + "callDuration": "3'27\"", + "callStatus": 2, # connected + "incomingCall": 0, # outgoing -> incoming_call=0 + "userId": "staff_user_001", + "userIdName": "Alice Wang", + "departmentId": "dept_sales_east", + "callNumber": "8613800000004", + "simPhone": "8613800000111", + "simNumber": "8986000112345678901", + "imei": "356938035643801", + "recordFile": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/phone-call/" + "8m0atd/imei001/1700000005000_1700000212000.mp3" + ), + "isUpload": 1, + "downloadCount": 2, + "aiAsr": None, + "qcStatus": 0, + "planCustomerName": "Customer D", + "version": "v1.20260731.pc2004", + }, +] + +# Default page size for the fixture (and the scraper). totalCount=4 means the +# fixture fills exactly 2 pages of size 2 plus a terminating empty page. +TOTAL_COUNT = len(CALL_RECORDS) +DEFAULT_PAGE_SIZE = 2 +# Mirrors the live capture's declared page count (purely cosmetic for the +# fixture; the paginator relies on the empty-page stop signal, not pageCount). +PAGE_COUNT = 2 + + +def make_call_log_response( + records: list[dict[str, Any]], + *, + total_count: int, + page_num: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + page_count: int | None = None, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``callLog/list`` response envelope. + + Shape (row list under ``data.data``; row count on ``data.totalCount``; + page count on ``data.pageCount``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "page":.., "pageCount":.., "totalCount":.., + "phoneHidden":false, "hideRecordPlay":false}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-{page_num}", + "data": { + "data": list(records), + "page": page_num, + "pageCount": page_count if page_count is not None else page_count_for(page_size, total_count), + "totalCount": total_count, + "phoneHidden": False, + "hideRecordPlay": False, + }, + } + + +def page_count_for(page_size: int, total_count: int) -> int: + """Compute the declared page count for a fixture (ceil(total/size)). + + Helper so the envelope's ``pageCount`` is internally consistent with the + declared ``totalCount`` + page size without forcing each test to recompute + it. The paginator does NOT consume ``pageCount`` (it relies on the + empty-page stop signal), so this is cosmetic — but it keeps the fixture + capture-shaped. + """ + if page_size <= 0: + return 0 + return (total_count + page_size - 1) // page_size diff --git a/yunke-server/tests/cassettes/smsLogs_fixture.py b/yunke-server/tests/cassettes/smsLogs_fixture.py new file mode 100644 index 0000000..96800cb --- /dev/null +++ b/yunke-server/tests/cassettes/smsLogs_fixture.py @@ -0,0 +1,136 @@ +"""Fixture payloads for ``POST /pc/customer/smsLogsList`` (M4 SMS scraper). + +These records mirror the real field names confirmed by Phase-0 + the data +dictionary §4 (短信记录 sms): ``id, customerPhone, sms(全文)/smsSub(预览), +sender, direction(0收/1发), userId, senderDepartment/senderDepartmentId, +receiveName, createTime(YYYY-MM-DD HH:MM), simPhone/simNumber, source``. Live +``dataCount`` is 53. + +The envelope differs from the callLog + wechat shapes: the row list sits under +``data.data`` (same) but the declared row count is ``data.dataCount`` (NOT +``totalCount`` and NOT ``total``), and there is **no** ``pageCount``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "dataCount":.., "power":...}} + +Because there is no ``pageCount``, the scraper relies solely on the empty-page +stop signal (advance ``pageNum`` until a page comes back with an empty +``data.data`` array). + +The cassette ships three records across two pages so a test can drive the +page-number paginator across the empty-page stop signal: + +* page 1 -> 2 of the 3 fixture records, ``dataCount=3`` +* page 2 -> the remaining 1 record, ``dataCount=3`` +* (page 3 -> empty ``data`` array, served inline by the test) + +Records deliberately exercise the documented mapping edges: + +* ``sms_3001`` — incoming SMS (``direction=0`` 收) with full + preview text. +* ``sms_3002`` — outgoing SMS (``direction=1`` 发) to a different customer. +* ``sms_3003`` — incoming SMS with an empty preview (exercises the + ``content_preview`` None path; ``sms`` still carries the full text). + +``createTime`` is a ``YYYY-MM-DD HH:MM`` datetime string (NO seconds — one +format precision shorter than the callLog's ``startCallTime``); the scraper +parses it to a ms-epoch int treating the missing seconds as ``:00``. +""" + +# The module name mirrors the upstream endpoint name (``smsLogsList`` -> +# ``smsLogs``); the M4 spec uses this naming family. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped SMS records (3 total across 2 pages: page1=2, page2=1). +# --------------------------------------------------------------------------- + +SMS_RECORDS: list[dict[str, Any]] = [ + { + # Incoming SMS (0=收) with full + preview text. + "id": "sms_3001", + "customerPhone": "8613800000001", + "sms": "您好，我在网上看到您的产品介绍，想了解一下价格和详情，方便回电吗？", + "smsSub": "您好，我在网上看到您的产品介绍...", + "sender": "8613800000001", + "direction": 0, # 0=收 (incoming) + "userId": "staff_user_001", + "senderDepartment": "销售一部", + "senderDepartmentId": "dept_sales_east", + "receiveName": "Alice Wang", + "createTime": "2026-01-15 10:30", # YYYY-MM-DD HH:MM (no seconds) + "simPhone": "8613800000111", + "simNumber": "8986000112345678901", + "source": "card", + }, + { + # Outgoing SMS (1=发) to a different customer. + "id": "sms_3002", + "customerPhone": "8613800000002", + "sms": "王先生您好，这是您咨询的订单详情，请查收附件。", + "smsSub": "王先生您好，这是您咨询的订单详情...", + "sender": "system", + "direction": 1, # 1=发 (outgoing) + "userId": "staff_user_002", + "senderDepartment": "销售二部", + "senderDepartmentId": "dept_sales_west", + "receiveName": "Bob Chen", + "createTime": "2026-01-15 11:00", + "simPhone": "8613800000222", + "simNumber": "8986000223456789012", + "source": "manual", + }, + { + # Incoming SMS with an EMPTY preview (content_preview None path). + "id": "sms_3003", + "customerPhone": "8613800000003", + "sms": "好的，收到。", + "smsSub": "", # empty preview -> content_preview None + "sender": "8613800000003", + "direction": 0, # incoming + "userId": "staff_user_001", + "senderDepartment": "销售一部", + "senderDepartmentId": "dept_sales_east", + "receiveName": "Alice Wang", + "createTime": "2026-01-16 09:00", + "simPhone": "8613800000111", + "simNumber": "8986000112345678901", + "source": "reply", + }, +] + +# Default page size for the fixture (and the scraper). dataCount=3 with +# page_size=2 fills page 1 (2 records) + page 2 (1 record) + a terminating +# empty page. +DATA_COUNT = len(SMS_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_sms_logs_response( + records: list[dict[str, Any]], + *, + data_count: int, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``smsLogsList`` response envelope. + + Shape (row list under ``data.data``; row count on ``data.dataCount``; + no ``pageCount`` / ``pageNum``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "dataCount":.., "power":true}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or "trace-1", + "data": { + "data": list(records), + "dataCount": data_count, + "power": True, + }, + } diff --git a/yunke-server/tests/test_orchestrator.py b/yunke-server/tests/test_orchestrator.py index 8ec6c1f..733234d 100644 --- a/yunke-server/tests/test_orchestrator.py +++ b/yunke-server/tests/test_orchestrator.py @@ -518,3 +518,164 @@ async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch): assert all(r.status == "success" for r in results) finally: await engine.dispose() + + +# =========================================================================== +# Phone calls + SMS — full re-scrape entities (M4) +# =========================================================================== + + +async def test_phone_calls_full_rescrape_upserts_all_records_sync_run_and_watermark( + monkeypatch, +): + """Phone-call list has no `since` param: 'incremental' = re-scrape all and + rely on upsert idempotency. The orchestrator upserts every record, records + a success SyncRun (entity_type='phone_call'), and stamps the watermark's + last_synced_at.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + call_recs = [ + {"call_id": "pc1", "channel": "phone_call", "version": "v1"}, + {"call_id": "pc2", "channel": "phone_call", "version": "v1"}, + ] + + async def fake_scrape_phone_calls(client): + for r in call_recs: + yield r + + upserted: list[dict] = [] + + async def fake_upsert_phone_call(session, rec): + upserted.append(rec) + + monkeypatch.setattr( + orchestrator, "scrape_phone_calls", fake_scrape_phone_calls + ) + monkeypatch.setattr( + orchestrator, "upsert_phone_call", fake_upsert_phone_call + ) + + async with SessionLocal() as session: + result = await orchestrator.sync_phone_calls(session, client=object()) + await session.commit() + + assert result.status == "success" + assert result.fetched == 2 + assert result.upserted == 2 + assert [r["call_id"] for r in upserted] == ["pc1", "pc2"] + + async with SessionLocal() as session: + run = ( + await session.execute( + sa.select(SyncRun).where(SyncRun.entity_type == "phone_call") + ) + ).scalar_one() + assert run.status == "success" + assert run.records_fetched == 2 + assert run.records_upserted == 2 + assert run.error is None + + state = ( + await session.execute( + sa.select(SyncState).where( + SyncState.entity_type == "phone_call" + ) + ) + ).scalar_one() + assert state.last_synced_at is not None + finally: + await engine.dispose() + + +async def test_sms_full_rescrape_upserts_all_records_sync_run_and_watermark( + monkeypatch, +): + """SMS list has no `since` param: same full-re-scrape shape as phone calls. + Records a success SyncRun (entity_type='sms') and stamps the watermark.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + sms_recs = [ + {"sms_id": "sms1", "channel": "sms", "content": "a"}, + {"sms_id": "sms2", "channel": "sms", "content": "b"}, + {"sms_id": "sms3", "channel": "sms", "content": "c"}, + ] + + async def fake_scrape_sms(client): + for r in sms_recs: + yield r + + upserted: list[dict] = [] + + async def fake_upsert_sms(session, rec): + upserted.append(rec) + + monkeypatch.setattr(orchestrator, "scrape_sms", fake_scrape_sms) + monkeypatch.setattr(orchestrator, "upsert_sms", fake_upsert_sms) + + async with SessionLocal() as session: + result = await orchestrator.sync_sms(session, client=object()) + await session.commit() + + assert result.status == "success" + assert result.fetched == 3 + assert result.upserted == 3 + assert [r["sms_id"] for r in upserted] == ["sms1", "sms2", "sms3"] + + async with SessionLocal() as session: + run = ( + await session.execute( + sa.select(SyncRun).where(SyncRun.entity_type == "sms") + ) + ).scalar_one() + assert run.status == "success" + assert run.records_fetched == 3 + assert run.records_upserted == 3 + assert run.error is None + + state = ( + await session.execute( + sa.select(SyncState).where(SyncState.entity_type == "sms") + ) + ).scalar_one() + assert state.last_synced_at is not None + finally: + await engine.dispose() + + +async def test_failing_phone_call_scraper_records_failed_run(monkeypatch): + """Same failure policy as the wechat entities: a scraper that raises + records a failed SyncRun (entity_type='phone_call') and returns a failed + result without re-raising.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + async def fake_scrape_phone_calls(client): + raise RuntimeError("callLog 500") + yield {} # pragma: no cover - make this an async generator + + monkeypatch.setattr( + orchestrator, "scrape_phone_calls", fake_scrape_phone_calls + ) + + async with SessionLocal() as session: + result = await orchestrator.sync_phone_calls(session, client=object()) + await session.commit() + + assert result.status == "failed" + assert "callLog 500" in (result.error or "") + + async with SessionLocal() as session: + run = ( + await session.execute( + sa.select(SyncRun).where(SyncRun.entity_type == "phone_call") + ) + ).scalar_one() + assert run.status == "failed" + assert "callLog 500" in (run.error or "") + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index c709656..529d288 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -29,6 +29,7 @@ from app.db.models import ( GroupMember, Message, PhoneCall, + Sms, Staff, WechatAccount, WhatsappAccount, @@ -41,6 +42,8 @@ from app.storage.repos import ( upsert_customers, upsert_department, upsert_messages, + upsert_phone_call, + upsert_sms, upsert_staff, upsert_staff_from_account, upsert_voice_call, @@ -1219,3 +1222,183 @@ async def test_upsert_messages_with_whatsapp_channel(): assert fetched.is_send is False finally: await engine.dispose() + + +# --------------------------------------------------------------------------- +# PhoneCall (phone_call) upsert (M4) +# --------------------------------------------------------------------------- + + +async def test_upsert_phone_call_is_idempotent_and_updates_record_file(): + """Same call_id twice with different record_file/version -> 1 row, refreshed. + + The phone-call channel (``channel="phone_call"``) shares the + :class:`PhoneCall` table with the voice-call channel; the per-row + ``channel`` value disambiguates them. + """ + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "call_id": "pc_dup", + "channel": "phone_call", + "record_file": "https://example.com/call_v1.mp3", + "call_duration": 10, + "call_status": 2, + "incoming_call": 0, + "version": "v1", + } + rec2 = { + "call_id": "pc_dup", + "channel": "phone_call", + "record_file": "https://example.com/call_v2.mp3", + "call_duration": 12, + "call_status": 0, + "incoming_call": 1, + "version": "v2", + } + + await upsert_phone_call(session, rec1) + await session.commit() + await upsert_phone_call(session, rec2) + await session.commit() + + rows = ( + await session.execute( + sa.select(PhoneCall) + .where(PhoneCall.call_id == "pc_dup") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(rows) == 1 + assert rows[0].call_id == "pc_dup" + assert rows[0].channel == "phone_call" + assert rows[0].record_file == "https://example.com/call_v2.mp3" + assert rows[0].call_duration == 12 + assert rows[0].call_status == 0 + assert rows[0].incoming_call == 1 + assert rows[0].version == "v2" + finally: + await engine.dispose() + + +async def test_upsert_phone_call_null_record_file_and_coerces_ints(): + """null record_file -> None; numeric-str call_duration/start_call_time/ + call_status/incoming_call/qc_status coerced to int; is_upload -> bool.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "call_id": "pc_null", + "channel": "phone_call", + "record_file": None, + "call_duration": "151", # str -> int + "start_call_time": "1700000000123", # numeric str -> int (ms epoch) + "call_status": "2", # str -> int + "incoming_call": "1", # str -> int (1=incoming/0=outgoing) + "qc_status": "1", # str -> int + "is_upload": "1", # str -> bool + "download_count": 3, + } + await upsert_phone_call(session, rec) + await session.commit() + + fetched = ( + await session.execute( + sa.select(PhoneCall).where(PhoneCall.call_id == "pc_null") + ) + ).scalar_one() + assert fetched.record_file is None + assert fetched.call_duration == 151 + assert fetched.start_call_time == 1700000000123 + assert fetched.call_status == 2 + assert fetched.incoming_call == 1 + assert fetched.qc_status == 1 + assert fetched.is_upload is True + assert fetched.download_count == 3 + finally: + await engine.dispose() + + +# --------------------------------------------------------------------------- +# Sms upsert (M4) +# --------------------------------------------------------------------------- + + +async def test_upsert_sms_is_idempotent_and_updates_content(): + """Same sms_id twice with different content/version -> 1 row, refreshed.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "sms_id": "sms_dup", + "channel": "sms", + "customer_phone": "8613800000001", + "content": "first message", + "content_preview": "first...", + "direction": 0, + "create_time": 1_700_000_000_000, + "version": "v1", + } + rec2 = { + "sms_id": "sms_dup", + "channel": "sms", + "customer_phone": "8613800000001", + "content": "updated message", + "content_preview": "updated...", + "direction": 0, + "create_time": 1_700_000_000_000, + "version": "v2", + } + + await upsert_sms(session, rec1) + await session.commit() + await upsert_sms(session, rec2) + await session.commit() + + rows = ( + await session.execute( + sa.select(Sms) + .where(Sms.sms_id == "sms_dup") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(rows) == 1 + assert rows[0].sms_id == "sms_dup" + assert rows[0].channel == "sms" + assert rows[0].content == "updated message" + assert rows[0].content_preview == "updated..." + assert rows[0].version == "v2" + finally: + await engine.dispose() + + +async def test_upsert_sms_coerces_direction_and_create_time(): + """direction numeric str -> int; create_time numeric str -> int (ms epoch).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "sms_id": "sms_coerce", + "channel": "sms", + "direction": "1", # str -> int (1=发/outgoing) + "create_time": "1700000000123", # numeric str -> int (ms epoch) + "content": "hi", + } + await upsert_sms(session, rec) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Sms).where(Sms.sms_id == "sms_coerce") + ) + ).scalar_one() + assert fetched.direction == 1 + assert fetched.create_time == 1700000000123 + assert fetched.content == "hi" + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_scrape_calls.py b/yunke-server/tests/test_scrape_calls.py new file mode 100644 index 0000000..4c53aad --- /dev/null +++ b/yunke-server/tests/test_scrape_calls.py @@ -0,0 +1,407 @@ +"""Tests for the phone-call + SMS scrapers (M4, ``app/scraper/call.py``). + +Drives the real page-number pagination + normalization logic through +``respx`` by wiring a real ``YunkeClient`` (primed with a fake session so the +login flow is skipped) against a mocked transport. + +Two endpoint suites: + +* ``POST /pc/callLog/list`` (phone calls) — envelope + ``data.data`` / ``data.totalCount`` / ``data.pageCount``. Maps to + :class:`PhoneCall` with ``channel="phone_call"``. ``incomingCall`` polarity + is ``1=incoming / 0=outgoing`` and is stored DIRECTLY (no inversion — unlike + the voice-call channel's ``isSend``). +* ``POST /pc/customer/smsLogsList`` (SMS) — envelope ``data.data`` / + ``data.dataCount`` (note: ``dataCount``, NOT ``totalCount``), no + ``pageCount``. Maps to :class:`Sms` with ``channel="sms"`` (the model + default). + +Both fixtures ship multiple pages so the tests exercise pagination across the +empty-page stop signal end-to-end, and assert the full field mapping incl. +the two distinct datetime string formats (``YYYY-MM-DD HH:MM:SS`` for calls, +``YYYY-MM-DD HH:MM`` for SMS) parsed to ms-epoch. +""" + +from __future__ import annotations + +import datetime as dt +import json as _json + +import httpx +import respx + +from app.scraper.call import scrape_phone_calls, scrape_sms +from app.session.yunke_client import YunkeClient +from tests.cassettes.callLog_fixture import ( + CALL_RECORDS, + TOTAL_COUNT, + make_call_log_response, +) +from tests.cassettes.callLog_fixture import ( + DEFAULT_PAGE_SIZE as CALL_PAGE_SIZE, +) +from tests.cassettes.smsLogs_fixture import ( + DATA_COUNT, + SMS_RECORDS, + make_sms_logs_response, +) +from tests.cassettes.smsLogs_fixture import ( + DEFAULT_PAGE_SIZE as SMS_PAGE_SIZE, +) + +BASE = "https://crm2.yunkecn.com" +CALL_LOG_URL = f"{BASE}/pc/callLog/list" +SMS_LOG_URL = f"{BASE}/pc/customer/smsLogsList" + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + + +def _ms_seconds(date_str: str) -> int: + """Expected ms-epoch for a ``YYYY-MM-DD HH:MM:SS`` string parsed as UTC. + + Mirrors the callLog scraper's documented parsing rule so the assertion is + an independent recomputation (not a self-reference). + """ + return int( + dt.datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S") + .replace(tzinfo=dt.UTC) + .timestamp() + * 1000 + ) + + +def _ms_minutes(date_str: str) -> int: + """Expected ms-epoch for a ``YYYY-MM-DD HH:MM`` string parsed as UTC. + + Mirrors the SMS scraper's documented parsing rule (seconds default to 00). + """ + return int( + dt.datetime.strptime(date_str, "%Y-%m-%d %H:%M") + .replace(tzinfo=dt.UTC) + .timestamp() + * 1000 + ) + + +# =========================================================================== +# Phone calls — POST /pc/callLog/list +# =========================================================================== + + +@respx.mock +async def test_scrape_phone_calls_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 4 records, 3 calls.""" + page1 = make_call_log_response( + CALL_RECORDS[:2], total_count=TOTAL_COUNT, page_num=1, page_size=CALL_PAGE_SIZE + ) + page2 = make_call_log_response( + CALL_RECORDS[2:], total_count=TOTAL_COUNT, page_num=2, page_size=CALL_PAGE_SIZE + ) + page3 = make_call_log_response( + [], total_count=TOTAL_COUNT, page_num=3, page_size=CALL_PAGE_SIZE + ) + + route = respx.post(CALL_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + httpx.Response(200, json=page3), + ] + ) + + client = _primed_client() + try: + results = [ + r async for r in scrape_phone_calls(client, page_size=CALL_PAGE_SIZE) + ] + finally: + await client.aclose() + + # Pagination: exactly 3 calls (page1, page2, empty-page stop). + assert route.call_count == 3 + + # Exactly the 4 fixture records (not duplicated, not dropped). + assert len(results) == 4 + assert {r["call_id"] for r in results} == {rec["id"] for rec in CALL_RECORDS} + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, CALL_RECORDS, strict=True): + # Identity + channel. + assert result["call_id"] == record["id"] + assert result["channel"] == "phone_call" + + # Staff side. + assert result["user_id"] == record["userId"] + assert result["user_id_name"] == record["userIdName"] + assert result["department_id"] == record["departmentId"] + + # Call metadata. + assert result["call_number"] == record["callNumber"] + assert result["sim_phone"] == record["simPhone"] + assert result["sim_number"] == record["simNumber"] + assert result["imei"] == record["imei"] + assert result["plan_customer_name"] == record["planCustomerName"] + + # Recording URL (may be None). + assert result["record_file"] == record["recordFile"] + + # Status + duration + start time coercions. + assert result["call_status"] == record["callStatus"] + # callSeconds preferred over callDuration string. + assert result["call_duration"] == record["callSeconds"] + assert result["start_call_time"] == _ms_seconds(record["startCallTime"]) + + # incomingCall polarity: 1=incoming/0=outgoing — stored DIRECTLY. + assert result["incoming_call"] == record["incomingCall"] + + # is_upload (bool field) carried; download_count + qc_status carried. + assert result["is_upload"] == record["isUpload"] + assert result["download_count"] == record["downloadCount"] + assert result["qc_status"] == record["qcStatus"] + assert result["ai_asr"] == record["aiAsr"] + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +@respx.mock +async def test_scrape_phone_calls_null_record_file_yields_none(): + """A record with ``recordFile=null`` -> ``record_file`` None.""" + rec = { + "id": "pc_null", + "startCallTime": "2026-01-15 10:00:00", + "callSeconds": 0, + "callStatus": 0, + "incomingCall": 1, # incoming + "userId": "staff_user_001", + "recordFile": None, + } + page1 = make_call_log_response([rec], total_count=1, page_num=1, page_size=CALL_PAGE_SIZE) + page2 = make_call_log_response([], total_count=1, page_num=2, page_size=CALL_PAGE_SIZE) + respx.post(CALL_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + ] + ) + + client = _primed_client() + try: + results = [ + r async for r in scrape_phone_calls(client, page_size=CALL_PAGE_SIZE) + ] + finally: + await client.aclose() + + assert len(results) == 1 + assert results[0]["record_file"] is None + assert results[0]["call_status"] == 0 + # incoming (incomingCall=1) -> incoming_call=1 (no inversion). + assert results[0]["incoming_call"] == 1 + + +@respx.mock +async def test_scrape_phone_calls_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_call_log_response( + CALL_RECORDS[:1], total_count=TOTAL_COUNT, page_num=1, page_size=1 + ) + page2 = make_call_log_response( + [], total_count=TOTAL_COUNT, page_num=2, page_size=1 + ) + route = respx.post(CALL_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + ] + ) + + client = _primed_client() + try: + results = [r async for r in scrape_phone_calls(client, page_size=1)] + finally: + await client.aclose() + + # One record on page 1, empty page 2 stops the loop. + assert len(results) == 1 + assert route.call_count == 2 + + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +@respx.mock +async def test_scrape_phone_calls_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_call_log_response([], total_count=0, page_num=1) + route = respx.post(CALL_LOG_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_phone_calls(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1 + + +# =========================================================================== +# SMS — POST /pc/customer/smsLogsList +# =========================================================================== + + +@respx.mock +async def test_scrape_sms_yields_records_and_paginates(): + """Two pages (2 + 1 records) + a terminating empty page -> 3 records.""" + page1 = make_sms_logs_response(SMS_RECORDS[:2], data_count=DATA_COUNT) + page2 = make_sms_logs_response(SMS_RECORDS[2:], data_count=DATA_COUNT) + page3 = make_sms_logs_response([], data_count=DATA_COUNT) + + route = respx.post(SMS_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + httpx.Response(200, json=page3), + ] + ) + + client = _primed_client() + try: + results = [r async for r in scrape_sms(client, page_size=SMS_PAGE_SIZE)] + finally: + await client.aclose() + + # Pagination: exactly 3 calls (page1, page2, empty-page stop). + assert route.call_count == 3 + + # Exactly the 3 fixture records. + assert len(results) == 3 + assert {r["sms_id"] for r in results} == {rec["id"] for rec in SMS_RECORDS} + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, SMS_RECORDS, strict=True): + # Identity + channel (the model's default 'sms'). + assert result["sms_id"] == record["id"] + assert result["channel"] == "sms" + + # Phone + text. + assert result["customer_phone"] == record["customerPhone"] + assert result["content"] == record["sms"] + # Empty preview string -> None (the scraper normalizes falsy -> None). + expected_preview = record["smsSub"] or None + assert result["content_preview"] == expected_preview + + # Parties. + assert result["sender"] == record["sender"] + assert result["direction"] == record["direction"] + assert result["user_id"] == record["userId"] + assert result["sender_department"] == record["senderDepartment"] + assert result["sender_department_id"] == record["senderDepartmentId"] + assert result["receive_name"] == record["receiveName"] + + # SIM + source. + assert result["sim_phone"] == record["simPhone"] + assert result["sim_number"] == record["simNumber"] + assert result["source"] == record["source"] + + # createTime "YYYY-MM-DD HH:MM" -> ms epoch (seconds default :00). + assert result["create_time"] == _ms_minutes(record["createTime"]) + + # raw_json preserves the original record. + assert result["raw_json"] == record + + +@respx.mock +async def test_scrape_sms_empty_preview_yields_none_content_preview(): + """An empty ``smsSub`` string -> ``content_preview`` None (full text kept).""" + rec = { + "id": "sms_empty_preview", + "customerPhone": "8613800000099", + "sms": "好的，收到。", + "smsSub": "", + "sender": "8613800000099", + "direction": 0, + "userId": "staff_user_001", + "createTime": "2026-01-16 09:00", + } + page1 = make_sms_logs_response([rec], data_count=1) + page2 = make_sms_logs_response([], data_count=1) + respx.post(SMS_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + ] + ) + + client = _primed_client() + try: + results = [r async for r in scrape_sms(client, page_size=SMS_PAGE_SIZE)] + finally: + await client.aclose() + + assert len(results) == 1 + assert results[0]["content"] == "好的，收到。" + assert results[0]["content_preview"] is None + + +@respx.mock +async def test_scrape_sms_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_sms_logs_response(SMS_RECORDS[:1], data_count=DATA_COUNT) + page2 = make_sms_logs_response([], data_count=DATA_COUNT) + route = respx.post(SMS_LOG_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + ] + ) + + client = _primed_client() + try: + results = [r async for r in scrape_sms(client, page_size=1)] + finally: + await client.aclose() + + assert len(results) == 1 + assert route.call_count == 2 + + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +@respx.mock +async def test_scrape_sms_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_sms_logs_response([], data_count=0) + route = respx.post(SMS_LOG_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_sms(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1