yunke/yunke-server/tests/test_scrape_wechat_groups.py

"""Tests for the WeChat chat-group scraper (M2).

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.

Groups reuse the SAME upstream endpoint as friends
(``POST /wechatweb/wxFriends/friend_list/es``); the ``friendTypeYunke="2"``
body parameter selects groups (``"1"`` selects friends). The response
envelope is identical to the friend scraper's, so this test reuses
:func:`tests.cassettes.wxFriends_fixture.make_friend_list_response` to build
capture-shaped envelopes.

The fixture ships 3 records across 2 pages of size 2, plus a terminating
empty page (the documented primary stop signal for page-number endpoints):

* page 1 -> 2 records, ``total=3``
* page 2 -> 1 record, ``total=3``
* page 3 -> empty ``data`` array (no more rows)

The scraper must yield exactly the 3 records (normalized to ``ChatGroup``
snake_case columns + ``raw_json``), make exactly 3 calls, and stop without a
4th. Member details are NOT surfaced by ``friend_list/es`` — ``member_ids``
is ``None`` and ``member_ids_list`` is ``[]`` on every yielded record.
"""

from __future__ import annotations

import json as _json
from typing import Any

import httpx
import respx

from app.scraper.wechat import _normalize_chat_group, scrape_wechat_groups
from app.session.yunke_client import YunkeClient
from tests.cassettes.wxFriends_fixture import make_friend_list_response

BASE = "https://crm2.yunkecn.com"
FRIEND_LIST_URL = f"{BASE}/wechatweb/wxFriends/friend_list/es"

USERID = "a" * 32
COMPANY = "8m0atd"

# Realistic, capture-shaped group records. Field names are EXACTLY what the
# live ``friend_list/es`` endpoint returns for ``friendTypeYunke="2"``
# (verified 2026-08-10 Playwright capture). Every field with a target
# ``ChatGroup`` column is populated so the test asserts a full mapping
# end-to-end. (Member ids are NOT in this payload — the endpoint surfaces
# only the ``friendNum`` count — so ``member_ids`` / ``member_ids_list`` are
# exercised as None / [] separately.)
GROUP_RECORDS: list[dict[str, Any]] = [
    {
        "friendWxId": "12345678901@chatroom",
        "friendWxNickname": "VIP客户交流群",
        "friendNum": 3,
        "friendHeadImageUrl": "https://wx.qlogo.cn/room_12345678901.png",
        "wxId": "wxid_staff_alice01",
        "ownerName": "Alice WeChat",
        "userId": "staff_user_001",
        "phone": "13800000001",
        "wechatPhone": "13800000011",
        "messageCount": 4521,
        "sendNum": 1204,
        "lastChatTime": 1700000000123,
        "version": "v1.20260731.room1",
    },
    {
        "friendWxId": "98765432109@chatroom",
        "friendWxNickname": "华东销售大群",
        "friendNum": 2,
        "friendHeadImageUrl": "https://wx.qlogo.cn/room_98765432109.png",
        "wxId": "wxid_staff_bob02",
        "ownerName": "Bob WeChat",
        "userId": "staff_user_002",
        "phone": "13800000002",
        "wechatPhone": "13800000022",
        "messageCount": 9821,
        "sendNum": 211,
        "lastChatTime": 1700000000456,
        "version": "v1.20260731.room2",
    },
    {
        # A group whose owner has no resolved staff id (userId None) — the
        # scraper must still yield it; None-safe fields stay None.
        "friendWxId": "55555555555@chatroom",
        "friendWxNickname": "External Coordination",
        "friendNum": 1,
        "friendHeadImageUrl": "https://wx.qlogo.cn/room_55555555555.png",
        "wxId": "wxid_staff_orphan04",
        "ownerName": "Orphan WeChat",
        "userId": None,
        "phone": None,
        "wechatPhone": None,
        "messageCount": 12,
        "sendNum": 0,
        "lastChatTime": 1700000000789,
        "version": "v1.20260731.room3",
    },
]

TOTAL = len(GROUP_RECORDS)
DEFAULT_PAGE_SIZE = 2


def _primed_client() -> YunkeClient:
    """A YunkeClient already in a logged-in state (login flow skipped)."""
    return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY)


# ---------------------------------------------------------------------------
# End-to-end: pagination + field mapping
# ---------------------------------------------------------------------------


@respx.mock
async def test_scrape_wechat_groups_yields_records_and_paginates():
    """Two full pages + a terminating empty page -> 3 records, 3 calls."""
    page1 = make_friend_list_response(
        GROUP_RECORDS[:2], total=TOTAL, page_num=1, page_size=DEFAULT_PAGE_SIZE
    )
    page2 = make_friend_list_response(
        GROUP_RECORDS[2:], total=TOTAL, page_num=2, page_size=DEFAULT_PAGE_SIZE
    )
    page3 = make_friend_list_response(
        [], total=TOTAL, page_num=3, page_size=DEFAULT_PAGE_SIZE
    )

    route = respx.post(FRIEND_LIST_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_wechat_groups(client, page_size=DEFAULT_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 (not duplicated, not dropped).
    assert len(results) == 3
    assert {r["room_id"] for r in results} == {
        rec["friendWxId"] for rec in GROUP_RECORDS
    }

    # Field mapping for each record (compare against the source record).
    for result, record in zip(results, GROUP_RECORDS, strict=True):
        # Identity + channel.
        assert result["room_id"] == record["friendWxId"]
        assert result["channel"] == "wechat"

        # Naming + ownership.
        assert result["room_name"] == record["friendWxNickname"]
        assert result["owner_account_id"] == record["wxId"]
        assert result["owner_account_nickname"] == record["ownerName"]
        assert result["owner_user_id"] == record["userId"]
        # phone preferred, wechatPhone fallback.
        assert result["owner_user_phone"] == (
            record["phone"] or record["wechatPhone"]
        )

        # Counts.
        assert result["member_count"] == record["friendNum"]
        assert result["message_count"] == record["messageCount"]
        assert result["own_message_count"] == record["sendNum"]

        # Avatar + timestamps.
        assert result["head_image_url"] == record["friendHeadImageUrl"]
        assert result["last_chat_time"] == record["lastChatTime"]

        # Member details NOT surfaced by friend_list/es -> None / [].
        assert result["member_ids"] is None
        assert result["member_ids_list"] == []
        # Fields the new endpoint does not carry.
        assert result["loss_members"] is None
        assert result["add_members"] is None
        assert result["update_time"] is None

        # Version carried verbatim; raw_json preserves the original record.
        assert result["version"] == record["version"]
        assert result["raw_json"] == record


# ---------------------------------------------------------------------------
# member_ids behaviour: always None / [] (not surfaced by friend_list/es)
# ---------------------------------------------------------------------------


@respx.mock
async def test_scrape_wechat_groups_emits_empty_member_ids():
    """``friend_list/es`` does not surface member ids; they are always
    ``None`` / ``[]`` on yielded records (member mirroring deferred)."""
    record = dict(GROUP_RECORDS[0])
    page1 = make_friend_list_response([record], total=1, page_num=1)
    page2 = make_friend_list_response([], total=1, page_num=2)
    respx.post(FRIEND_LIST_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_wechat_groups(client, page_size=DEFAULT_PAGE_SIZE)
        ]
    finally:
        await client.aclose()

    assert len(results) == 1
    assert results[0]["member_ids"] is None
    assert results[0]["member_ids_list"] == []


# ---------------------------------------------------------------------------
# Pagination body: pageNum/pageSize advanced + friendTypeYunke="2"
# ---------------------------------------------------------------------------


@respx.mock
async def test_scrape_wechat_groups_sends_pagination_body():
    """Outgoing body carries pageNum/pageSize and ``friendTypeYunke="2"``."""
    page1 = make_friend_list_response(
        [GROUP_RECORDS[0]], total=TOTAL, page_num=1, page_size=1
    )
    page2 = make_friend_list_response([], total=TOTAL, page_num=2, page_size=1)
    route = respx.post(FRIEND_LIST_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_wechat_groups(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
    # The crucial switch: friendTypeYunke="2" selects GROUPS ("1" = friends).
    assert body1["friendTypeYunke"] == "2"

    body2 = _json.loads(route.calls[1].request.content.decode())
    assert body2["pageNum"] == 2
    assert body2["pageSize"] == 1
    assert body2["friendTypeYunke"] == "2"


# ---------------------------------------------------------------------------
# Empty first page -> no yields, single call
# ---------------------------------------------------------------------------


@respx.mock
async def test_scrape_wechat_groups_stops_on_empty_first_page():
    """If the very first page is empty, scraper stops without over-fetching."""
    empty = make_friend_list_response([], total=0, page_num=1)
    route = respx.post(FRIEND_LIST_URL).mock(
        return_value=httpx.Response(200, json=empty)
    )

    client = _primed_client()
    try:
        results = [r async for r in scrape_wechat_groups(client)]
    finally:
        await client.aclose()

    assert results == []
    assert route.call_count == 1


# ---------------------------------------------------------------------------
# 群主/建群补充列提取(2026-08-25 对齐云客员工聊天窗口·微信群 tab)
# raw_json 键: ownerName/ownerWchatId/ownerAlias/ownerImage/friendCount/createTime
# smoke.db 实测: createTime 全部为 'YYYY-MM-DD HH:mm:ss' 字符串;ms int 直通。
# ---------------------------------------------------------------------------


def test_normalize_chat_group_extracts_owner_fields():
    """6 个补充键提取到同名列;date-str createTime 归一为 ms epoch。"""
    record = {
        "friendWxId": "123@chatroom",
        "friendWxNickname": "测试群",
        "friendNum": 25,
        "wxId": "wxid_x",
        "ownerName": "Q**e",
        "ownerWchatId": "RL3*****108",
        "ownerAlias": "Fr******er",
        "ownerImage": "https://wx.qlogo.cn/o/0",
        "friendCount": 22,
        "createTime": "2025-10-27 13:30:55",
    }
    out = _normalize_chat_group(record)
    assert out["owner_name"] == "Q**e"
    assert out["owner_wxid"] == "RL3*****108"
    assert out["owner_alias"] == "Fr******er"
    assert out["owner_avatar"] == "https://wx.qlogo.cn/o/0"
    assert out["friend_count"] == 22
    # 'YYYY-MM-DD HH:mm:ss' -> ms epoch(naive UTC,同 addTime 约定)
    assert out["create_time"] == 1761571855000


def test_normalize_chat_group_owner_fields_none_safe():
    """缺失全部 6 个键 -> 全 None(老 capture 不炸)。"""
    out = _normalize_chat_group({"friendWxId": "r@chatroom"})
    for key in ("owner_name", "owner_wxid", "owner_alias",
                "owner_avatar", "friend_count", "create_time"):
        assert out[key] is None


def test_normalize_chat_group_create_time_ms_passthrough():
    """ms-epoch int 直通(不做日期解析)。"""
    out = _normalize_chat_group(
        {"friendWxId": "r@chatroom", "createTime": 1740000000000}
    )
    assert out["create_time"] == 1740000000000