yunke/yunke-server/app/scraper/call.py

"""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)