diff --git a/yunke-server/app/scraper/__init__.py b/yunke-server/app/scraper/__init__.py new file mode 100644 index 0000000..2afc8ca --- /dev/null +++ b/yunke-server/app/scraper/__init__.py @@ -0,0 +1,12 @@ +"""Scraper layer for the yunke data-mirror system. + +Scrapers sit between the session layer (:mod:`app.session.yunke_client`) and +the persistence layer (a LATER task). Each scraper is an async generator that +walks one upstream list endpoint and yields **normalized dicts** — +snake_case keys matching the target ORM model columns, plus a ``raw_json`` key +holding the original upstream record verbatim. + +Persistence (upsert into ORM tables) is intentionally out of scope here: it is +Task 1.7. Keeping scrapers pure producers lets us test ingestion logic against +``respx`` fixtures without a database. +""" diff --git a/yunke-server/app/scraper/base.py b/yunke-server/app/scraper/base.py new file mode 100644 index 0000000..cf151b1 --- /dev/null +++ b/yunke-server/app/scraper/base.py @@ -0,0 +1,137 @@ +"""Reusable building blocks for list scrapers. + +This module establishes the pagination pattern every list scraper in the +project reuses (see design doc §6.2 拉取层). It is intentionally small: + +* :func:`paginate_by_page` — a generic async generator that walks page-numbered + endpoints until a page comes back empty (the primary stop signal used by the + yunke CRM), with a defensive over-fetch ceiling keyed off the declared total. +* :class:`RateLimiter` — a tiny asyncio limiter that enforces a minimum + interval between successive calls. Scrapers that need to be polite to the + upstream construct one and ``await`` :meth:`acquire` before each call. + +The paginator talks to :class:`app.session.yunke_client.YunkeClient` (which +already injects auth headers + handles 401 re-login). It is generic over the +response shape via three callables: + +* ``parse_total(resp) -> int | None`` — read the declared row count from the + envelope (used only for the over-fetch safety; ``None`` disables it). +* ``parse_rows(resp) -> list[dict]`` — extract the list of records. +* ``build_body(page_num, page_size, **extra) -> dict`` — build the request + body for a given page. +""" + +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator, Callable +from typing import Any + +from app.session.yunke_client import YunkeClient + +__all__ = ["RateLimiter", "paginate_by_page"] + +# Informal type aliases for the three response-handling callables. The yunke +# response shapes are dict-shaped JSON and do not benefit from finer typing. +ParseTotal = Callable[[Any], "int | None"] +ParseRows = Callable[[Any], "list[dict[str, Any]]"] +BuildBody = Callable[..., "dict[str, Any]"] + + +async def paginate_by_page( + client: YunkeClient, + path: str, + *, + page_size: int, + parse_total: ParseTotal, + parse_rows: ParseRows, + build_body: BuildBody, + rate_limiter: RateLimiter | None = None, + **extra: Any, +) -> AsyncIterator[dict[str, Any]]: + """Walk a page-numbered list endpoint, yielding one dict per record. + + Pagination strategy + ------------------- + Start at ``pageNum=1`` and advance while pages are non-empty. The yunke + CRM signals "no more rows" by returning an empty ``data`` array on the + page *after* the last real record; that empty page is the **primary stop + signal**. As a defensive ceiling against a misbehaving upstream that never + sends an empty page, we also stop once we have yielded strictly more rows + than the declared ``total`` (strict ``>`` so an honest source yields + exactly ``total`` rows). + + ``total`` is parsed lazily from the first response and may be ``None`` if + the envelope does not carry it — in that case only the empty-page signal + terminates the loop. + + ``rate_limiter`` is an optional :class:`RateLimiter`; when supplied, it is + acquired before every upstream call. + """ + page_num = 1 + total: int | None = None + yielded = 0 + + while True: + if rate_limiter is not None: + await rate_limiter.acquire() + + body = build_body(page_num, page_size, **extra) + resp = await client.post(path, json=body) + + # Read the declared total once from the first response. The yunke + # envelopes keep it on every page, but reading it once is enough for + # the over-fetch ceiling and avoids trusting stale updates. + if total is None: + total = parse_total(resp) + + rows = parse_rows(resp) + if not rows: + # Empty page == the upstream's "no more rows" signal. This is the + # normal, primary termination path. + return + + for row in rows: + yield row + yielded += 1 + + # Over-fetch safety: a well-behaved source sends an empty page next, + # but if it never does (or sends duplicates), cap at the declared total. + # Strict ``>`` so we yield exactly ``total`` when the count is honest. + if total is not None and yielded > total: + return + + page_num += 1 + + +class RateLimiter: + """Minimal asyncio limiter enforcing a minimum interval between calls. + + Call ``await limiter.acquire()`` before each upstream request. The first + call never waits. Successive calls wait just long enough that the gap + between the previous call's dispatch and this one is at least + ``min_interval`` seconds. + + This is deliberately tiny: no tokens, no concurrency cap, no skew handling + — just a floor on inter-call spacing, which is all the yunke scrapers need + to stay polite to the CRM. If you need a semaphore or token bucket, that + belongs in a future helper, not here. + """ + + def __init__(self, min_interval: float = 0.0) -> None: + self._min_interval = float(min_interval) + self._last: float = 0.0 + + async def acquire(self) -> None: + """Block until at least ``min_interval`` has elapsed since the last call.""" + if self._min_interval <= 0: + self._last = time.monotonic() + return + + now = time.monotonic() + wait = self._min_interval - (now - self._last) + if wait > 0: + await asyncio.sleep(wait) + + self._last = time.monotonic() diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py new file mode 100644 index 0000000..56ab654 --- /dev/null +++ b/yunke-server/app/scraper/wechat.py @@ -0,0 +1,171 @@ +"""WeChat friend-list scraper. + +Wraps ``POST /wechatweb/wxFriends/friend_list/es`` and yields normalized dicts +ready to be upserted into :class:`app.db.models.Customer` (persistence is a +LATER task — Task 1.7). Each dict carries: + +* snake_case keys matching the ``Customer`` columns (see :func:`_normalize`), +* ``channel == "wechat"``, +* ``raw_json`` — the original upstream record, verbatim, so nothing captured + is lost even when a column has no target field. + +Field names are mapped from the real capture (see Task 1.5 spec). The mapping +is a straight ``camelCase -> snake_case`` translation with a couple of +disambiguations where the source has two fields for one column: + +* ``phone`` <- ``phone`` +* ``wechat_phone`` <- ``wechatPhone`` +* ``dept_name`` <- ``deptName`` (falls back to ``deptRouteName``) +* ``head_image_url`` <- ``friendHeadImageUrl`` (falls back to ``headImg``) +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +from app.scraper.base import paginate_by_page +from app.session.yunke_client import YunkeClient + +__all__ = ["scrape_wechat_friends"] + +FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es" + + +def _parse_total(resp: Any) -> int | None: + """Read the declared row count from the friend_list envelope. + + Shape: ``data.basicResult.total`` (an int). Returns ``None`` if the path + is absent so the paginator falls back to the empty-page stop signal. + """ + 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 + basic = data.get("basicResult") + if not isinstance(basic, dict): + return None + total = basic.get("total") + return int(total) if isinstance(total, (int, float)) else None + + +def _parse_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the list of friend records from the envelope. + + Shape: ``data.basicResult.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 [] + basic = data.get("basicResult") + if not isinstance(basic, dict): + return [] + rows = basic.get("data") + return rows if isinstance(rows, list) else [] + + +def _build_body( + page_num: int, + page_size: int, + *, + department_id: str = "", + user_id: str = "", +) -> dict[str, Any]: + """Build the friend_list request body for a given page. + + The real client sends ``departmentId`` / ``userId`` filters; we accept + them as optional kwargs (default empty) so per-staff / per-department + scrapes are a drop-in without changing the paginator contract. + """ + body: dict[str, Any] = {"pageNum": page_num, "pageSize": page_size} + if department_id: + body["departmentId"] = department_id + if user_id: + body["userId"] = user_id + return body + + +def _normalize(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream friend record to the ``Customer`` column shape. + + Adds ``channel="wechat"`` and preserves the original record under + ``raw_json``. ``None``-safe: missing fields become ``None`` rather than + raising, so partial payloads do not break the whole scrape. + """ + return { + "friend_id": record.get("friendWxId"), + "channel": "wechat", + "owner_account_id": record.get("wxId"), + "friend_nickname": record.get("friendWxNickname"), + "friend_remark": record.get("friendWxRemark"), + "real_name": record.get("realName"), + "phone": record.get("phone"), + "wechat_phone": record.get("wechatPhone"), + "gender": record.get("gender"), + "region": record.get("region"), + "description": record.get("description"), + "sale_name": record.get("saleName"), + "user_id": record.get("userId"), + "dept_route": record.get("deptRoute"), + # Prefer the canonical deptName; fall back to the route-display name. + "dept_name": record.get("deptName") or record.get("deptRouteName"), + # Prefer friendHeadImageUrl; fall back to the legacy headImg field. + "head_image_url": record.get("friendHeadImageUrl") or record.get("headImg"), + "last_chat_time": record.get("lastChatTime"), + "send_num": record.get("sendNum"), + "receive_num": record.get("receiveNum"), + "message_count": record.get("messageCount"), + "delete_flag": record.get("deleteFlag"), + "is_block": record.get("isBlock"), + "friend_state": record.get("friendState"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_wechat_friends( + client: YunkeClient, + *, + department_id: str = "", + user_id: str = "", + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WeChat friend records from the CRM. + + Walks ``POST /wechatweb/wxFriends/friend_list/es`` page by page until the + upstream reports no more rows, mapping each record to the ``Customer`` + column shape (snake_case) plus a verbatim ``raw_json`` copy. + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client; this scraper only concerns itself with the + friend-list endpoint contract. + department_id, user_id: + Optional upstream filters (passed through to the request body as + ``departmentId`` / ``userId``). Empty by default — scrape everything + the logged-in account can see. + 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, + FRIEND_LIST_PATH, + page_size=page_size, + parse_total=_parse_total, + parse_rows=_parse_rows, + build_body=_build_body, + department_id=department_id, + user_id=user_id, + ): + yield _normalize(record) diff --git a/yunke-server/tests/cassettes/__init__.py b/yunke-server/tests/cassettes/__init__.py new file mode 100644 index 0000000..089345b --- /dev/null +++ b/yunke-server/tests/cassettes/__init__.py @@ -0,0 +1,6 @@ +"""Test fixtures for yunke scrapers. + +Each module in this package holds realistic, capture-shaped payloads for one +upstream endpoint so the scraper test suite can drive the real pagination / +normalization code through ``respx`` without touching the live CRM. +""" diff --git a/yunke-server/tests/cassettes/wxFriends_fixture.py b/yunke-server/tests/cassettes/wxFriends_fixture.py new file mode 100644 index 0000000..58c3d64 --- /dev/null +++ b/yunke-server/tests/cassettes/wxFriends_fixture.py @@ -0,0 +1,122 @@ +"""Fixture payloads for ``POST /wechatweb/wxFriends/friend_list/es``. + +These records mirror the real field names captured from the live CRM (see the +Task 1.5 spec). Two realistic friend records are provided so a test can drive +the paginator across two pages: page 1 returns both records, page 2 returns an +empty ``data`` array to signal "no more" (the documented primary stop signal). +""" +# The module name uses camelCase deliberately: it mirrors the upstream endpoint +# name (``wxFriends``) and the Task 1.5 spec mandates this exact filename. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# Realistic, capture-shaped friend records. Field names are EXACTLY what the +# live endpoint returns — the scraper maps these to ``Customer`` snake_case +# columns. Every field that has a target column is populated so the test can +# assert a full mapping end-to-end. +FRIEND_RECORDS: list[dict[str, Any]] = [ + { + "companyCode": "8m0atd", + "wxId": "wxid_staff_alice01", + "friendWxId": "wxid_friend_0001", + "friendWxAlias": "alice_alias", + "friendWxNickname": "Alice Wang", + "friendWxRemark": "Alice-VIP", + "realName": "Wang Ai Li", + "friendWxPhone": "13800000001", + "phone": "13800000001", + "wechatPhone": "13800000011", + "gender": 2, + "region": "Shanghai", + "description": "VIP client, prefers wechat", + "saleName": "Tom Sale", + "userId": "staff_user_001", + "deptRoute": "1.2.3", + "deptRouteName": "Sales/East/TeamA", + "deptName": "Sales-East-TeamA", + "friendHeadImageUrl": "https://cdn.example.com/avatar_alice.png", + "headImg": "https://cdn.example.com/avatar_alice_legacy.png", + "lastChatTime": 1700000000123, + "sendNum": 12, + "receiveNum": 34, + "messageCount": 46, + "deleteFlag": 0, + "isBlock": 0, + "friendState": 1, + "version": "v1.20260731.alice", + "addTime": 1690000000000, + "createTime": 1690000000000, + "fromType": "qrcode", + "addType": "scan", + }, + { + "companyCode": "8m0atd", + "wxId": "wxid_staff_bob02", + "friendWxId": "wxid_friend_0002", + "friendWxAlias": "bob_alias", + "friendWxNickname": "Bob Chen", + "friendWxRemark": "Bob-Lead", + "realName": "Chen Bo", + "friendWxPhone": "13800000002", + "phone": "13800000002", + "wechatPhone": "13800000022", + "gender": 1, + "region": "Beijing", + "description": "New lead from expo", + "saleName": "Jerry Sale", + "userId": "staff_user_002", + "deptRoute": "1.2.4", + "deptRouteName": "Sales/North/TeamB", + "deptName": "Sales-North-TeamB", + "friendHeadImageUrl": "https://cdn.example.com/avatar_bob.png", + "headImg": "https://cdn.example.com/avatar_bob_legacy.png", + "lastChatTime": 1700000000456, + "sendNum": 3, + "receiveNum": 9, + "messageCount": 12, + "deleteFlag": 0, + "isBlock": 1, + "friendState": 2, + "version": "v1.20260731.bob", + "addTime": 1691000000000, + "createTime": 1691000000000, + "fromType": "phone", + "addType": "import", + }, +] + + +def make_friend_list_response( + records: list[dict[str, Any]], + *, + total: int, + page_num: int = 1, + page_size: int = 50, +) -> dict[str, Any]: + """Build a capture-shaped ``friend_list/es`` response envelope. + + The shape matches the live API exactly: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"displayStatus":"0", + "basicResult":{"pageNum":..,"pageSize":..,"total":.., + "data":[ {}, ... ]}}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": f"trace-{page_num}", + "data": { + "displayStatus": "0", + "basicResult": { + "pageNum": page_num, + "pageSize": page_size, + "total": total, + "data": list(records), + }, + }, + } diff --git a/yunke-server/tests/test_scrape_base.py b/yunke-server/tests/test_scrape_base.py new file mode 100644 index 0000000..c711cb7 --- /dev/null +++ b/yunke-server/tests/test_scrape_base.py @@ -0,0 +1,48 @@ +"""Smoke tests for the reusable scraper building blocks in :mod:`app.scraper.base`. + +The generic paginator is exercised end-to-end by the WeChat friend-scraper +tests (it is the only caller today). Here we only sanity-check the +:class:`RateLimiter`'s spacing behaviour so the helper is not shipped blind. +""" + +from __future__ import annotations + +import asyncio +import time + +from app.scraper.base import RateLimiter + + +async def test_rate_limiter_no_interval_never_waits(): + """``min_interval=0`` is a passthrough — acquire returns immediately.""" + limiter = RateLimiter(min_interval=0.0) + start = time.monotonic() + for _ in range(5): + await limiter.acquire() + elapsed = time.monotonic() - start + # 5 acquires with no spacing should complete well under a real interval. + assert elapsed < 0.1 + + +async def test_rate_limiter_enforces_min_interval(): + """Two acquires are spaced by at least ``min_interval`` seconds.""" + limiter = RateLimiter(min_interval=0.05) + await limiter.acquire() # first call never waits; primes ``_last`` + start = time.monotonic() + await limiter.acquire() # must wait >= 0.05s since the prior call + elapsed = time.monotonic() - start + assert elapsed >= 0.04 # allow a small scheduler slack margin + + +async def test_rate_limiter_concurrent_acquire_is_safe(): + """Concurrent acquires do not raise; each just waits its turn.""" + limiter = RateLimiter(min_interval=0.01) + await asyncio.gather(*(limiter.acquire() for _ in range(8))) + # No assertion on timing here — the contract is "no crash, polite spacing". + # We only assert the gather completed without raising. + + +def test_rate_limiter_is_instantiable_with_negative_interval(): + """Negative intervals collapse to the passthrough branch (no surprising raise).""" + limiter = RateLimiter(min_interval=-1.0) + assert limiter._min_interval <= 0 # sanity check the guard diff --git a/yunke-server/tests/test_scrape_wechat_friends.py b/yunke-server/tests/test_scrape_wechat_friends.py new file mode 100644 index 0000000..d2eab65 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_friends.py @@ -0,0 +1,154 @@ +"""Tests for the WeChat friend-list scraper (Task 1.5). + +Drives the *real* 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 pages are served: + +* page 1 -> total=2, both records (i.e. the full result set in one page), +* page 2 -> total=2, empty ``data`` array (the "no more rows" signal). + +The paginator must yield exactly the 2 records (normalized), make exactly 2 +calls, then stop without a 3rd call. +""" + +from __future__ import annotations + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_friends +from app.session.yunke_client import YunkeClient +from tests.cassettes.wxFriends_fixture import ( + FRIEND_RECORDS, + make_friend_list_response, +) + +BASE = "https://crm2.yunkecn.com" +FRIEND_LIST_URL = f"{BASE}/wechatweb/wxFriends/friend_list/es" + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + c = YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + return c + + +@respx.mock +async def test_scrape_wechat_friends_yields_normalized_records_and_paginates(): + """Two pages served; scraper yields 2 mapped dicts and makes exactly 2 calls.""" + # Page 1: both records, total=2. Page 2: empty data, total still 2. + page1 = make_friend_list_response(FRIEND_RECORDS, total=len(FRIEND_RECORDS), page_num=1) + page2 = make_friend_list_response([], total=len(FRIEND_RECORDS), page_num=2) + + 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_friends(client)] + finally: + await client.aclose() + + # --- Pagination behaviour: exactly 2 calls, no speculative 3rd. --- + assert route.call_count == 2 + + # --- Yielded exactly the 2 fixture records (not duplicated, not dropped). --- + assert len(results) == 2 + assert {r["friend_id"] for r in results} == { + rec["friendWxId"] for rec in FRIEND_RECORDS + } + + # --- Field mapping for each record (compare against the source record). --- + for result, record in zip(results, FRIEND_RECORDS, strict=True): + # Identity / ownership / channel. + assert result["friend_id"] == record["friendWxId"] + assert result["channel"] == "wechat" + assert result["owner_account_id"] == record["wxId"] + assert result["user_id"] == record["userId"] + + # Naming. + assert result["friend_nickname"] == record["friendWxNickname"] + assert result["friend_remark"] == record["friendWxRemark"] + assert result["real_name"] == record["realName"] + + # Phones: `phone` <- "phone", `wechat_phone` <- "wechatPhone". + assert result["phone"] == record["phone"] + assert result["wechat_phone"] == record["wechatPhone"] + + # Misc mapped fields. + assert result["gender"] == record["gender"] + assert result["region"] == record["region"] + assert result["description"] == record["description"] + assert result["sale_name"] == record["saleName"] + assert result["dept_route"] == record["deptRoute"] + assert result["dept_name"] == record["deptName"] + assert result["head_image_url"] == record["friendHeadImageUrl"] + assert result["last_chat_time"] == record["lastChatTime"] + assert result["send_num"] == record["sendNum"] + assert result["receive_num"] == record["receiveNum"] + assert result["message_count"] == record["messageCount"] + assert result["delete_flag"] == record["deleteFlag"] + assert result["is_block"] == record["isBlock"] + assert result["friend_state"] == record["friendState"] + + # 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_wechat_friends_sends_pagination_body(): + """Outgoing request carries pageNum/pageSize and optional filters.""" + page1 = make_friend_list_response([], total=0, page_num=1) + route = respx.post(FRIEND_LIST_URL).mock( + return_value=httpx.Response(200, json=page1) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_wechat_friends( + client, department_id="dept-9", user_id="staff-1", page_size=7 + ) + ] + finally: + await client.aclose() + + # Empty upstream -> no rows, exactly one call (the empty-page stop). + assert results == [] + assert route.call_count == 1 + + sent = route.calls.last.request + import json as _json + + body = _json.loads(sent.content.decode()) + assert body["pageNum"] == 1 + assert body["pageSize"] == 7 + assert body["departmentId"] == "dept-9" + assert body["userId"] == "staff-1" + + +@respx.mock +async def test_scrape_wechat_friends_stops_on_empty_first_page(): + """If the very first page is empty, the 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_friends(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1