diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index 25ad6ca..831850a 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -1,19 +1,31 @@ -"""WeChat scrapers. +"""WeChat + organization scrapers. -Two endpoints today: +Endpoints covered today: * :func:`scrape_wechat_friends` — wraps ``POST /wechatweb/wxFriends/friend_list/es`` and yields normalized dicts ready to be upserted into :class:`app.db.models.Customer`. * :func:`scrape_wechat_messages` — wraps ``POST /wechatweb/msg/chats`` and yields normalized chat-message dicts (see Task 1.6). +* :func:`scrape_departments` — wraps ``POST /usercenter/dept/tree`` and + yields normalized :class:`app.db.models.Department` dicts. Single request, + no pagination; the tree (flat or nested) is flattened and the materialized + ``path`` is reconstructed from parent links (Task 2.0). +* :func:`scrape_wechat_accounts` — wraps ``POST /wechatweb/wxUser/list`` and + yields normalized :class:`app.db.models.WechatAccount` dicts (plus the + embedded ``userId``/``userName`` staff pair). Page-number pagination + (Task 2.1). -Both yield dicts (not ORM). Persistence is a LATER task — Task 1.7. Each dict -carries: +All yield dicts (not ORM). Persistence is the caller's job (the +:mod:`app.storage.repos` module bridges these dicts to the ORM tables). Each +dict carries: * snake_case keys matching the target ORM columns (see :func:`_normalize` - for friends and :func:`_normalize_message` for messages), -* ``channel == "wechat"``, + for friends, :func:`_normalize_message` for messages, + :func:`_normalize_department` for departments, and + :func:`_normalize_wxuser_account` for accounts), +* ``channel == "wechat"`` (only on communication-channel entities; the + organization entities ``Department`` / ``WechatAccount`` carry no channel), * ``raw_json`` — the original upstream record, verbatim, so nothing captured is lost even when a column has no target field. @@ -36,10 +48,17 @@ from app.scraper.base import paginate_by_cursor, paginate_by_page from app.scraper.parsers import normalize_wechat_type from app.session.yunke_client import YunkeClient -__all__ = ["scrape_wechat_friends", "scrape_wechat_messages"] +__all__ = [ + "scrape_departments", + "scrape_wechat_accounts", + "scrape_wechat_friends", + "scrape_wechat_messages", +] FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es" MSG_CHATS_PATH = "/wechatweb/msg/chats" +DEPT_TREE_PATH = "/usercenter/dept/tree" +WXUSER_LIST_PATH = "/wechatweb/wxUser/list" # Upstream sends form-encoded bodies for ``msg/chats`` (NOT JSON), so we have # to override the client's default JSON content-type on every request. @@ -408,3 +427,276 @@ async def scrape_wechat_messages( if _is_divider(record): continue yield _normalize_message(record) + + +# =========================================================================== +# Department tree scraper (Task 2.0) — POST /usercenter/dept/tree +# =========================================================================== + + +def _extract_dept_node_list(resp: Any) -> list[dict[str, Any]]: + """Extract the raw node list from a ``dept/tree`` response, defensively. + + The envelope's ``data`` may be: + + * a list of nodes (the common case), OR + * an object wrapping the node list under ``children`` (or ``list`` / + ``tree`` / ``data`` — be tolerant of wrapper-key variance). + + Returns ``[]`` on any malformed shape so the scraper yields nothing rather + than raising: a broken upstream should not abort the whole sync. Nodes are + returned AS-IS (possibly still nested via ``children``); flattening is the + next step (:func:`_flatten_dept_tree`). + """ + try: + body = resp.json() + except (AttributeError, ValueError): + return [] + data = body.get("data") if isinstance(body, dict) else None + if isinstance(data, list): + return data + if isinstance(data, dict): + # Walk the plausible wrapper keys in priority order. ``children`` is + # the documented yunke tree convention; the others are defensive. + for key in ("children", "list", "tree", "data"): + val = data.get(key) + if isinstance(val, list): + return val + return [] + + +def _flatten_dept_tree(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Flatten a (possibly nested) dept-tree node list depth-first. + + Each node MAY carry a ``children`` array (the nested shape); when absent, + the tree is already flat and parents are carried via ``pId``. Both shapes + flatten to the same depth-first list. + + Each emitted node is a shallow copy of the source node with the + ``children`` key stripped, so the ``raw_json`` we store per-dept does not + carry a recursive subtree (only that node's own attributes). Non-dict + entries are dropped defensively. + """ + flat: list[dict[str, Any]] = [] + # Iterative DFS via a stack. Nodes are pushed in reverse so the first + # child is popped first (preserves source order in the output). + stack: list[dict[str, Any]] = [ + n for n in reversed(nodes) if isinstance(n, dict) + ] + while stack: + node = stack.pop() + children = node.get("children") + # Shallow copy without ``children`` so raw_json stored per-dept is clean. + flat_node = {k: v for k, v in node.items() if k != "children"} + flat.append(flat_node) + if isinstance(children, list) and children: + stack.extend( + reversed([c for c in children if isinstance(c, dict)]) + ) + return flat + + +def _build_dept_paths(nodes: list[dict[str, Any]]) -> dict[str, str]: + """Compute the materialized path (ancestor names joined by ``/``) per node. + + Returns ``{id: path}``. The path is built by walking up the ``pId`` chain; + the walk terminates when it reaches a node whose ``pId`` is falsy + (``None`` / ``""`` / ``0`` / ``"0"``) OR whose parent is not in the tree + (handles the common yunke convention where root carries ``pId="0"`` and no + node with ``id="0"`` exists). A visited set caps the walk so a cyclic + ``pId`` chain cannot loop forever. + + A node whose own ``id`` is missing is skipped (no path entry) — the caller + treats a missing path as empty string. + """ + by_id: dict[str, dict[str, Any]] = {} + for n in nodes: + nid = n.get("id") + # First occurrence wins so duplicate ids cannot shadow the canonical + # node (defensive against an upstream that emits a node twice). + if nid is not None and nid not in by_id: + by_id[nid] = n + + paths: dict[str, str] = {} + for nid, node in by_id.items(): + parts: list[str] = [] + seen: set[str] = set() + cur: dict[str, Any] | None = node + while cur is not None: + cid = cur.get("id") + if cid is None or cid in seen: + break + seen.add(cid) + name = cur.get("name") + if name is not None: + parts.append(name) + pid = cur.get("pId") + # Falsy pId OR unknown parent terminates the walk. + cur = by_id.get(pid) if pid else None + paths[nid] = "/".join(reversed(parts)) + return paths + + +def _normalize_department( + node: dict[str, Any], path: str +) -> dict[str, Any]: + """Map one dept-tree node to the ``Department`` column shape. + + ``parent_id`` is carried verbatim from ``pId`` — the scraper does NOT + synthesize ``NULL`` for root nodes; downstream code treats falsy + ``parent_id`` (``"0"`` / ``""`` / ``None``) as "no parent" consistently + with the materialized-path builder. ``path`` is the precomputed + slash-joined ancestor-name chain from :func:`_build_dept_paths`. + """ + return { + "dept_id": node.get("id"), + "parent_id": node.get("pId"), + "name": node.get("name"), + "path": path, + "raw_json": node, + } + + +async def scrape_departments( + client: YunkeClient, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized department records from ``POST /usercenter/dept/tree``. + + Single request (no pagination). The endpoint follows yunke's tree + convention: nodes carry ``id`` / ``pId`` / ``name`` and MAY be delivered + either as a flat list (parents via ``pId``) or as a nested tree (each node + optionally carrying ``children``). Both shapes are flattened to one flat + list of records, each carrying the materialized ``path`` (ancestor names + joined with ``/``). + + Defensive envelope: ``data`` may be the node list directly OR an object + wrapping the node list under ``children``; both are unwrapped. + + The async-generator signature mirrors the other scrapers so callers can + consume every entity stream with the same ``async for`` pattern, even + though this endpoint always resolves in one round-trip. + """ + resp = await client.post(DEPT_TREE_PATH, json={}) + nodes = _extract_dept_node_list(resp) + flat = _flatten_dept_tree(nodes) + paths = _build_dept_paths(flat) + for node in flat: + nid = node.get("id") + path = paths.get(nid, "") if nid is not None else "" + yield _normalize_department(node, path) + + +# =========================================================================== +# WeChat staff-account scraper (Task 2.1) — POST /wechatweb/wxUser/list +# =========================================================================== + + +def _parse_wxuser_total(resp: Any) -> int | None: + """Read the declared row count from the ``wxUser/list`` envelope. + + Shape: ``data.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 + total = data.get("total") + return int(total) if isinstance(total, (int, float)) and not isinstance( + total, bool + ) else None + + +def _parse_wxuser_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the account list from the ``wxUser/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_wxuser_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``wxUser/list`` request body for a given page. + + Unlike ``friend_list/es``, the ``wxUser/list`` body carries no + staff/dept filter parameters in the live capture — just the page slice. + """ + return {"pageNum": page_num, "pageSize": page_size} + + +def _normalize_wxuser_account(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``wxUser/list`` record to the ``WechatAccount`` shape. + + Carries the embedded staff pair (``user_id`` <- ``userId``, + ``user_name`` <- ``userName``) so the persistence layer can upsert the + owning :class:`app.db.models.Staff` row alongside the account. Adds + ``raw_json`` (the verbatim upstream record). ``None``-safe: missing fields + become ``None`` rather than raising, so partial payloads do not break the + scrape. + """ + return { + "wx_id": record.get("wechatId"), + "nick_name": record.get("nickName"), + "alias": record.get("alias"), + "phone": record.get("phone"), + "user_id": record.get("userId"), + "user_name": record.get("userName"), + "contacts_num": record.get("contactsNum"), + "group_num": record.get("groupNum"), + "send_num": record.get("sendNum"), + "receive_num": record.get("receiveNum"), + "is_on_job": record.get("isOnJob"), + "db_size": record.get("dbSize"), + "online_status": record.get("onlineStatus"), + "last_update": record.get("lastUpdate"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_wechat_accounts( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WeChat staff-account records from ``wxUser/list``. + + Walks ``POST /wechatweb/wxUser/list`` page by page until the upstream + reports no more rows (empty ``data.data`` array — the primary stop signal), + mapping each record to the ``WechatAccount`` column shape plus a verbatim + ``raw_json`` copy. + + The envelope differs from ``friend_list/es``: there is **no** + ``basicResult`` wrapper. The pagination fields sit directly under ``data`` + (``data.data`` / ``data.total`` / ``data.pageNum`` / ``data.pageSize``). + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + page_size: + Page size to request. Defaults to 50 (matches the live capture's + observed page size for this endpoint). + """ + async for record in paginate_by_page( + client, + WXUSER_LIST_PATH, + page_size=page_size, + parse_total=_parse_wxuser_total, + parse_rows=_parse_wxuser_rows, + build_body=_build_wxuser_body, + ): + yield _normalize_wxuser_account(record) diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index 3981985..2f5437d 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -48,12 +48,14 @@ import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import DeclarativeBase -from app.db.models import Customer, Message, Staff, WechatAccount +from app.db.models import Customer, Department, Message, Staff, WechatAccount __all__ = [ "upsert_customer", "upsert_customers", + "upsert_department", "upsert_messages", + "upsert_staff", "upsert_staff_from_account", "upsert_wechat_account", ] @@ -303,29 +305,67 @@ async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> return len(recs) +# --------------------------------------------------------------------------- +# Department upsert (Task 2.0) +# --------------------------------------------------------------------------- + + +async def upsert_department(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`Department` (PK = ``dept_id``). + + Same ``dept_id`` twice -> one row, with non-PK fields (``parent_id`` / + ``name`` / ``path`` / ``raw_json``) refreshed from the later record. + """ + await _upsert_one(session, Department, rec) + + # --------------------------------------------------------------------------- # WechatAccount + embedded Staff (best-effort) # --------------------------------------------------------------------------- -async def upsert_wechat_account(session: AsyncSession, rec: dict[str, Any]) -> None: - """Best-effort idempotent upsert for :class:`WechatAccount` (PK = ``wx_id``). +async def upsert_staff(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`Staff` row from a staff-shaped dict. - The companion :func:`upsert_staff_from_account` covers the embedded staff - row (``user_id`` + ``user_name``); the two are split so callers can pick - either or both. A full staff scraper (departments, phones, etc.) is a - LATER milestone. + PK = ``user_id``. Same ``user_id`` twice -> one row, with non-PK fields + refreshed. Applies the documented coercions (``is_on_job`` -> bool). + + This is the primitive staff upsert; the account-scoped helper + :func:`upsert_staff_from_account` (which extracts the staff fields from a + wechat-account record) delegates here, and :func:`upsert_wechat_account` + calls that helper so an account upsert automatically refreshes its + embedded staff row too. + """ + await _upsert_one(session, Staff, rec) + + +async def upsert_wechat_account(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert a :class:`WechatAccount` AND its embedded Staff row. + + PK = ``wx_id``. The companion :func:`upsert_staff_from_account` covers the + embedded staff row (``user_id`` + ``user_name``); this helper calls it + automatically so an account scrape that lands accounts also lands the + owning staff — callers do not need to upsert staff separately. + + The staff upsert is best-effort: an account record with no ``user_id`` + (e.g. an orphaned / unassigned wechat number) still writes the account + row; the staff upsert silently no-ops. """ await _upsert_one(session, WechatAccount, rec) + await upsert_staff_from_account(session, rec) async def upsert_staff_from_account(session: AsyncSession, rec: dict[str, Any]) -> None: """Best-effort upsert of the Staff row embedded in a wechat-account record. - Maps the account's ``user_id`` + ``user_name`` (plus ``is_on_job`` when - present) onto :class:`Staff` (PK = ``user_id``). Silently no-ops when the - record has no ``user_id`` — there is nothing to key the row on. A - dedicated staff scraper that fills out phone / dept is a LATER milestone. + Maps the account's ``user_id`` + ``user_name`` (plus ``is_on_job`` / + ``phone`` / ``sale_phone`` when present) onto :class:`Staff` + (PK = ``user_id``) and delegates to :func:`upsert_staff`. Silently no-ops + when the record has no ``user_id`` — there is nothing to key the row on. + + Per Task 2.1, dept linkage for staff is inferred ELSEWHERE (from + customers' ``deptRoute``); staff with no customers have no dept. This + helper therefore does not populate ``dept_id`` — that is a later concern. """ user_id = rec.get("user_id") if not user_id: @@ -335,6 +375,11 @@ async def upsert_staff_from_account(session: AsyncSession, rec: dict[str, Any]) "name": rec.get("user_name"), "raw_json": rec, } - if "is_on_job" in rec: - staff_rec["is_on_job"] = rec["is_on_job"] - await _upsert_one(session, Staff, staff_rec) + # Forward any Staff-relevant fields the account record happens to carry. + # ``phone`` and ``sale_phone`` are usually absent on the wxUser payload + # (they come from the WhatsApp account payload); carry them through when + # present so a future shared caller does not silently drop them. + for key in ("is_on_job", "phone", "sale_phone"): + if key in rec: + staff_rec[key] = rec[key] + await upsert_staff(session, staff_rec) diff --git a/yunke-server/tests/cassettes/dept_tree_fixture.py b/yunke-server/tests/cassettes/dept_tree_fixture.py new file mode 100644 index 0000000..1efbe18 --- /dev/null +++ b/yunke-server/tests/cassettes/dept_tree_fixture.py @@ -0,0 +1,107 @@ +"""Fixture payloads for ``POST /usercenter/dept/tree`` (Task 2.0). + +yunke's dept/tree endpoint follows the same convention as the permissions API: +nodes carry ``id`` / ``pId`` / ``name``, and the tree MAY be delivered either +as a flat list of nodes or as a nested tree (each node optionally carrying a +``children`` array). The scraper must flatten both shapes into one flat list +of nodes and reconstruct ``parent_id`` from ``pId``. + +This cassette provides TWO realistic capture-shaped payloads so the scraper +tests can prove both shapes flatten to the SAME set of nodes: + +* ``NESTED_TREE_PAYLOAD`` — a root with two children, one of which has its own + sub-child (3 levels deep). Nested via ``children``. +* ``FLAT_TREE_PAYLOAD`` — the exact same 4 nodes delivered as a flat list + (parent links carried by ``pId``). + +Both decode to the same 4 departments with the same ids / parents / names / +paths, so the scraper test can swap the envelope and re-assert. + +The 4 nodes: + root "Sales" (id="1", pId="0") + ├── "East" (id="1.1", pId="1") + │ └── "TeamA" (id="1.1.1", pId="1.1") + └── "North" (id="1.2", pId="1") +""" + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# The canonical 4 nodes (single source of truth for both shapes below) +# --------------------------------------------------------------------------- + +# Root has pId="0" — the common yunke convention for "no parent". The scraper +# treats any falsy pId (None / "" / "0") OR an unknown pId as "root". +ROOT_NODE: dict[str, Any] = {"id": "1", "pId": "0", "name": "Sales"} +EAST_NODE: dict[str, Any] = {"id": "1.1", "pId": "1", "name": "East"} +TEAMA_NODE: dict[str, Any] = {"id": "1.1.1", "pId": "1.1", "name": "TeamA"} +NORTH_NODE: dict[str, Any] = {"id": "1.2", "pId": "1", "name": "North"} + +# All four nodes, in a stable order matching a depth-first traversal of the +# nested tree. Tests assert the scraper yields exactly these (order is not +# load-bearing — tests compare as sets where appropriate). +ALL_NODES: list[dict[str, Any]] = [ROOT_NODE, EAST_NODE, TEAMA_NODE, NORTH_NODE] + + +# --------------------------------------------------------------------------- +# Nested envelope — root carries children, recursively +# --------------------------------------------------------------------------- + +NESTED_TREE_NODES: list[dict[str, Any]] = [ + { + **ROOT_NODE, + "children": [ + { + **EAST_NODE, + "children": [{**TEAMA_NODE}], + }, + {**NORTH_NODE}, + ], + }, +] + + +# --------------------------------------------------------------------------- +# Flat envelope — every node at the top level, parents via pId +# --------------------------------------------------------------------------- + +FLAT_TREE_NODES: list[dict[str, Any]] = [ + {**ROOT_NODE}, + {**EAST_NODE}, + {**TEAMA_NODE}, + {**NORTH_NODE}, +] + + +def make_dept_tree_response( + nodes: list[dict[str, Any]], + *, + nested_in_data_object: bool = False, + trace_id: str = "trace-dept", +) -> dict[str, Any]: + """Build a capture-shaped ``dept/tree`` response envelope. + + The yunke dept/tree response top-level is the standard envelope + ``{message, code:"10000", success, traceId, data}``. The shape of ``data`` + itself varies between captures — be defensive: + + * ``data`` is the node list directly (the common case), OR + * ``data`` is an object wrapping the node list under ``children`` / + ``list`` / etc. + + The ``nested_in_data_object`` flag selects the second variant for the + "wrapped data object" path so a test can prove the scraper unwraps it. + The default (``False``) emits the most common flat-data array shape. + """ + data: Any = list(nodes) + if nested_in_data_object: + data = {"children": list(nodes)} + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id, + "data": data, + } diff --git a/yunke-server/tests/cassettes/wxUser_fixture.py b/yunke-server/tests/cassettes/wxUser_fixture.py new file mode 100644 index 0000000..60312e3 --- /dev/null +++ b/yunke-server/tests/cassettes/wxUser_fixture.py @@ -0,0 +1,147 @@ +"""Fixture payloads for ``POST /wechatweb/wxUser/list`` (Task 2.1). + +These records mirror the real field names captured from the live CRM (see the +data dictionary §2: ``wechatId, nickName, alias, phone, userId, +userName(员工真名), contactsNum, groupNum, sendNum/receiveNum, isOnJob, +dbSize, 在线状态, lastUpdate``; total=55). + +The envelope differs from ``wxFriends/friend_list/es``: there is **no** +``basicResult`` wrapper. The pagination fields sit directly under ``data``:: + + {"code":"10000","data":{"data":[...], "total":.., "pageNum":.., "pageSize":..}} + +The cassette ships 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, ``total=4`` +* page 2 -> the other 2 records, ``total=4`` +* (page 3 -> empty ``data`` array, served inline by the test) + +Records carry every field with a target ``WechatAccount`` column so the test +can assert a full mapping end-to-end, plus the embedded ``userId``+``userName`` +staff pair (the owning employee) so the embedded-staff repo upsert can be +exercised against capture-shaped input. +""" + +# The module name uses camelCase deliberately to mirror the upstream endpoint +# name (``wxUser``); the Task 2.1 spec mandates this exact filename. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped WeChat staff-account records (4 total across 2 pages). +# --------------------------------------------------------------------------- + +WXUSER_RECORDS: list[dict[str, Any]] = [ + { + "wechatId": "wxid_staff_alice01", + "nickName": "Alice WeChat", + "alias": "alice_wx", + "phone": "13800000001", + "userId": "staff_user_001", + "userName": "Alice Wang", + "contactsNum": 312, + "groupNum": 18, + "sendNum": 1204, + "receiveNum": 2387, + "isOnJob": 1, + "dbSize": 482103, + "onlineStatus": "online", + "headImageUrl": "https://wx.qlogo.cn/avatar_alice.png", + "lastUpdate": 1700000000123, + "version": "v1.20260731.alice", + }, + { + "wechatId": "wxid_staff_bob02", + "nickName": "Bob WeChat", + "alias": "bob_wx", + "phone": "13800000002", + "userId": "staff_user_002", + "userName": "Bob Chen", + "contactsNum": 87, + "groupNum": 4, + "sendNum": 211, + "receiveNum": 402, + "isOnJob": 1, + "dbSize": 91205, + "onlineStatus": "offline", + "headImageUrl": "https://wx.qlogo.cn/avatar_bob.png", + "lastUpdate": 1700000000456, + "version": "v1.20260731.bob", + }, + { + "wechatId": "wxid_staff_carol03", + "nickName": "Carol We Chat", + "alias": "carol_wx", + "phone": "13800000003", + "userId": "staff_user_003", + "userName": "Carol Liu", + "contactsNum": 1502, + "groupNum": 72, + "sendNum": 9034, + "receiveNum": 11245, + "isOnJob": 0, # departed staff — exercises the falsy isOnJob path + "dbSize": 2_100_445, + "onlineStatus": "offline", + "headImageUrl": "https://wx.qlogo.cn/avatar_carol.png", + "lastUpdate": 1700000000789, + "version": "v1.20260731.carol", + }, + { + # Account with NO embedded staff (userId missing) — the scraper must + # still yield it; the embedded-staff repo upsert must skip it cleanly. + "wechatId": "wxid_staff_orphan04", + "nickName": "Orphan WeChat", + "alias": "orphan_wx", + "phone": "13800000004", + "userId": None, + "userName": None, + "contactsNum": 5, + "groupNum": 0, + "sendNum": 1, + "receiveNum": 2, + "isOnJob": 0, + "dbSize": 1024, + "onlineStatus": "offline", + "headImageUrl": "https://wx.qlogo.cn/avatar_orphan.png", + "lastUpdate": 1700000001000, + "version": "v1.20260731.orphan", + }, +] + +# Default page size for the fixture (and the scraper). total=4 means the +# fixture fills exactly 2 pages of size 2 plus a terminating empty page. +TOTAL = len(WXUSER_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_wxuser_list_response( + records: list[dict[str, Any]], + *, + total: int, + page_num: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``wxUser/list`` response envelope. + + Shape (no ``basicResult`` wrapper):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "pageNum":.., "pageSize":..}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-{page_num}", + "data": { + "data": list(records), + "total": total, + "pageNum": page_num, + "pageSize": page_size, + }, + } diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index 8ff77cb..fbd7933 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -22,13 +22,15 @@ import sqlalchemy as sa from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.db.base import Base -from app.db.models import Customer, Message, Staff, WechatAccount +from app.db.models import Customer, Department, Message, Staff, WechatAccount from app.storage.repos import ( _coerce_bool, _coerce_int, upsert_customer, upsert_customers, + upsert_department, upsert_messages, + upsert_staff, upsert_staff_from_account, upsert_wechat_account, ) @@ -426,3 +428,276 @@ async def test_upsert_staff_from_account_skips_when_no_user_id(): assert rows == [] finally: await engine.dispose() + + +# --------------------------------------------------------------------------- +# Department upsert (Task 2.0) +# --------------------------------------------------------------------------- + + +async def test_upsert_department_is_idempotent_and_updates_path(): + """Same dept_id twice with different path -> 1 row, path refreshed.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "dept_id": "1.1", + "parent_id": "1", + "name": "East", + "path": "Sales/East", + } + rec2 = { + "dept_id": "1.1", + "parent_id": "1", + "name": "East", + # Re-scrape reconstructs the same path; verify it is overwritten + # rather than producing a duplicate row. + "path": "Sales/East", + "raw_json": {"id": "1.1", "pId": "1", "name": "East"}, + } + + await upsert_department(session, rec1) + await session.commit() + await upsert_department(session, rec2) + await session.commit() + + rows = ( + await session.execute( + sa.select(Department) + .where(Department.dept_id == "1.1") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(rows) == 1 + assert rows[0].dept_id == "1.1" + assert rows[0].parent_id == "1" + assert rows[0].name == "East" + assert rows[0].path == "Sales/East" + # raw_json from the second upsert wins (last write). + assert rows[0].raw_json == rec2["raw_json"] + finally: + await engine.dispose() + + +async def test_upsert_department_preserves_none_parent_for_root(): + """A root dept (parent_id None) is stored as-is; no spurious coercion.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "dept_id": "1", + "parent_id": None, + "name": "Sales", + "path": "Sales", + } + await upsert_department(session, rec) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Department).where(Department.dept_id == "1") + ) + ).scalar_one() + assert fetched.parent_id is None + assert fetched.name == "Sales" + assert fetched.path == "Sales" + finally: + await engine.dispose() + + +# --------------------------------------------------------------------------- +# upsert_staff — direct staff-shaped dict upsert (Task 2.1) +# --------------------------------------------------------------------------- + + +async def test_upsert_staff_is_idempotent_and_updates_name(): + """Same user_id twice with different name -> 1 row, name refreshed.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = {"user_id": "u_new", "name": "Old Name"} + rec2 = {"user_id": "u_new", "name": "New Name"} + + await upsert_staff(session, rec1) + await session.commit() + await upsert_staff(session, rec2) + await session.commit() + + rows = ( + await session.execute( + sa.select(Staff) + .where(Staff.user_id == "u_new") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(rows) == 1 + assert rows[0].name == "New Name" + finally: + await engine.dispose() + + +async def test_upsert_staff_coerces_is_on_job(): + """is_on_job '1' -> True; '0' -> False; None preserved.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + await upsert_staff( + session, {"user_id": "u_b1", "is_on_job": "1"} + ) + await upsert_staff( + session, {"user_id": "u_b0", "is_on_job": "0"} + ) + await upsert_staff( + session, {"user_id": "u_bn", "is_on_job": None} + ) + await session.commit() + + on = ( + await session.execute( + sa.select(Staff).where(Staff.user_id == "u_b1") + ) + ).scalar_one() + off = ( + await session.execute( + sa.select(Staff).where(Staff.user_id == "u_b0") + ) + ).scalar_one() + none_ = ( + await session.execute( + sa.select(Staff).where(Staff.user_id == "u_bn") + ) + ).scalar_one() + assert on.is_on_job is True + assert off.is_on_job is False + assert none_.is_on_job is None + finally: + await engine.dispose() + + +# --------------------------------------------------------------------------- +# upsert_wechat_account now ALSO upserts the embedded Staff (Task 2.1) +# --------------------------------------------------------------------------- + + +async def test_upsert_wechat_account_also_upserts_embedded_staff(): + """Account upsert creates BOTH the WechatAccount row and the Staff row.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "wx_id": "wxid_dup", + "nick_name": "Dup WeChat", + "user_id": "u_dup", + "user_name": "Dup Real Name", + "is_on_job": 1, + "phone": "13800000099", + } + await upsert_wechat_account(session, rec) + await session.commit() + + acct = ( + await session.execute( + sa.select(WechatAccount).where(WechatAccount.wx_id == "wxid_dup") + ) + ).scalar_one() + assert acct.user_id == "u_dup" + assert acct.user_name == "Dup Real Name" + + staff = ( + await session.execute( + sa.select(Staff).where(Staff.user_id == "u_dup") + ) + ).scalar_one() + assert staff.name == "Dup Real Name" + assert staff.is_on_job is True + finally: + await engine.dispose() + + +async def test_upsert_wechat_account_reupsert_updates_embedded_staff(): + """Re-upserting the account refreshes BOTH account + embedded staff rows.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "wx_id": "wxid_re", + "nick_name": "Old Nick", + "user_id": "u_re", + "user_name": "Old Name", + "is_on_job": 1, + } + rec2 = { + "wx_id": "wxid_re", + "nick_name": "New Nick", + "user_id": "u_re", + "user_name": "New Name", + "is_on_job": 0, # staff departed since last scrape + } + await upsert_wechat_account(session, rec1) + await session.commit() + await upsert_wechat_account(session, rec2) + await session.commit() + + # One account row, refreshed. + acct_rows = ( + await session.execute( + sa.select(WechatAccount) + .where(WechatAccount.wx_id == "wxid_re") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(acct_rows) == 1 + assert acct_rows[0].nick_name == "New Nick" + + # One staff row, refreshed (not duplicated). + staff_rows = ( + await session.execute( + sa.select(Staff) + .where(Staff.user_id == "u_re") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(staff_rows) == 1 + assert staff_rows[0].name == "New Name" + assert staff_rows[0].is_on_job is False + finally: + await engine.dispose() + + +async def test_upsert_wechat_account_without_user_id_skips_staff(): + """Account with no user_id is written; embedded-staff upsert is skipped.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "wx_id": "wxid_no_staff", + "nick_name": "Orphan WeChat", + # no user_id / user_name + } + await upsert_wechat_account(session, rec) + await session.commit() + + # Account row written. + acct = ( + await session.execute( + sa.select(WechatAccount).where( + WechatAccount.wx_id == "wxid_no_staff" + ) + ) + ).scalar_one() + assert acct.nick_name == "Orphan WeChat" + + # No staff rows at all. + staff_rows = ( + await session.execute(sa.select(Staff)) + ).scalars().all() + assert staff_rows == [] + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_scrape_departments.py b/yunke-server/tests/test_scrape_departments.py new file mode 100644 index 0000000..6fcb41e --- /dev/null +++ b/yunke-server/tests/test_scrape_departments.py @@ -0,0 +1,187 @@ +"""Tests for the department-tree scraper (Task 2.0). + +Drives the real flatten + normalize logic through ``respx`` by wiring a real +``YunkeClient`` (primed with a fake session so the login flow is skipped) +against a mocked transport. + +The dept/tree endpoint is a SINGLE request (no pagination). The scraper must: + +* decode the envelope defensively (``data`` is the node list, OR an object + wrapping the node list under ``children``), +* flatten BOTH a nested tree (each node carrying ``children``) AND a flat list + of nodes (parents via ``pId``) into the same flat set of records, +* map ``id``/``pId``/``name`` to ``dept_id``/``parent_id``/``name``, +* build a materialized ``path`` by walking up the parent chain (names joined + with ``/``; root nodes have ``path == name``), +* preserve the verbatim node under ``raw_json``. + +The two-shape contract (nested vs flat) is asserted by serving the SAME four +nodes in both shapes and asserting the scraper yields the SAME result set in +both cases. +""" + +from __future__ import annotations + +import httpx +import respx + +from app.scraper.wechat import scrape_departments +from app.session.yunke_client import YunkeClient +from tests.cassettes.dept_tree_fixture import ( + ALL_NODES, + EAST_NODE, + FLAT_TREE_NODES, + NESTED_TREE_NODES, + NORTH_NODE, + ROOT_NODE, + TEAMA_NODE, + make_dept_tree_response, +) + +BASE = "https://crm2.yunkecn.com" +DEPT_TREE_URL = f"{BASE}/usercenter/dept/tree" + +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 _expected_path(node: dict) -> str: + """Compute the expected materialized path for a fixture node. + + Root (Sales) -> "Sales"; East -> "Sales/East"; TeamA -> "Sales/East/TeamA"; + North -> "Sales/North". + """ + by_id = {n["id"]: n for n in ALL_NODES} + parts: list[str] = [] + cur: dict | None = node + while cur is not None: + parts.append(cur["name"]) + pid = cur.get("pId") + # Falsy / unknown pId terminates the walk (root reached). + cur = by_id.get(pid) if pid else None + return "/".join(reversed(parts)) + + +# --------------------------------------------------------------------------- +# End-to-end: nested tree flattens correctly with paths reconstructed +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_departments_flattens_nested_tree_with_paths(): + """Nested tree -> flat list of 4 depts with correct parent_id + path.""" + payload = make_dept_tree_response(NESTED_TREE_NODES) + route = respx.post(DEPT_TREE_URL).mock( + return_value=httpx.Response(200, json=payload) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_departments(client)] + finally: + await client.aclose() + + # Single request (no pagination). + assert route.call_count == 1 + + # All four nodes yielded (order-independent). + assert len(results) == 4 + by_dept_id = {r["dept_id"]: r for r in results} + assert set(by_dept_id) == {n["id"] for n in ALL_NODES} + + # Field mapping + parent reconstruction for each node. + for node in ALL_NODES: + rec = by_dept_id[node["id"]] + assert rec["dept_id"] == node["id"] + assert rec["name"] == node["name"] + # Root node's parent_id is preserved verbatim (pId="0") — the scraper + # does not synthesize NULL; it carries pId through. Materialized-path + # construction treats falsy/unknown pId as "stop walking". + assert rec["parent_id"] == node["pId"] + assert rec["path"] == _expected_path(node) + # raw_json preserves the original (flattened) node — keys id/pId/name. + assert rec["raw_json"]["id"] == node["id"] + assert rec["raw_json"]["name"] == node["name"] + + # Spot-check the headline path expectations (defensive against path-helper + # bugs masking themselves). + assert by_dept_id[ROOT_NODE["id"]]["path"] == "Sales" + assert by_dept_id[EAST_NODE["id"]]["path"] == "Sales/East" + assert by_dept_id[TEAMA_NODE["id"]]["path"] == "Sales/East/TeamA" + assert by_dept_id[NORTH_NODE["id"]]["path"] == "Sales/North" + + +# --------------------------------------------------------------------------- +# Flat list of nodes (parents via pId) yields the SAME result as nested +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_departments_flattens_flat_list_to_same_result(): + """Flat node list (no children arrays) flattens to the same 4 records.""" + payload = make_dept_tree_response(FLAT_TREE_NODES) + respx.post(DEPT_TREE_URL).mock(return_value=httpx.Response(200, json=payload)) + + client = _primed_client() + try: + results = [r async for r in scrape_departments(client)] + finally: + await client.aclose() + + assert len(results) == 4 + by_dept_id = {r["dept_id"]: r for r in results} + for node in ALL_NODES: + rec = by_dept_id[node["id"]] + assert rec["parent_id"] == node["pId"] + assert rec["path"] == _expected_path(node) + + +# --------------------------------------------------------------------------- +# Defensive envelope: data is an OBJECT wrapping nodes under `children` +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_departments_unwraps_data_object_children(): + """When ``data`` is an object with ``children``, scraper unwraps it.""" + payload = make_dept_tree_response( + FLAT_TREE_NODES, nested_in_data_object=True + ) + respx.post(DEPT_TREE_URL).mock(return_value=httpx.Response(200, json=payload)) + + client = _primed_client() + try: + results = [r async for r in scrape_departments(client)] + finally: + await client.aclose() + + assert len(results) == 4 + assert {r["dept_id"] for r in results} == {n["id"] for n in ALL_NODES} + + +# --------------------------------------------------------------------------- +# Empty tree -> no yields, single call (no infinite loop on missing data) +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_departments_empty_tree_yields_nothing(): + """Empty data array -> no yields, exactly one call.""" + payload = make_dept_tree_response([]) + route = respx.post(DEPT_TREE_URL).mock( + return_value=httpx.Response(200, json=payload) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_departments(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1 diff --git a/yunke-server/tests/test_scrape_wechat_accounts.py b/yunke-server/tests/test_scrape_wechat_accounts.py new file mode 100644 index 0000000..5fc15c5 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_accounts.py @@ -0,0 +1,192 @@ +"""Tests for the WeChat staff-account scraper (Task 2.1). + +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. + +The fixture ships 4 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=4`` +* page 2 -> 2 records, ``total=4`` +* page 3 -> empty ``data`` array (no more rows) + +The scraper must yield exactly the 4 records (normalized to ``WechatAccount`` +snake_case columns + ``raw_json``), make exactly 3 calls, and stop without a +4th. The empty-page stop is the primary signal; ``total=4`` is only a +defensive over-fetch ceiling that must NOT trigger early (we yield exactly +4 == total, so the strict-``>`` ceiling never fires here). +""" + +from __future__ import annotations + +import json as _json + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_accounts +from app.session.yunke_client import YunkeClient +from tests.cassettes.wxUser_fixture import ( + DEFAULT_PAGE_SIZE, + TOTAL, + WXUSER_RECORDS, + make_wxuser_list_response, +) + +BASE = "https://crm2.yunkecn.com" +WXUSER_LIST_URL = f"{BASE}/wechatweb/wxUser/list" + +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) + + +# --------------------------------------------------------------------------- +# End-to-end: pagination + field mapping + raw_json +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_accounts_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 4 records, 3 calls.""" + page1 = make_wxuser_list_response( + WXUSER_RECORDS[:2], total=TOTAL, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_wxuser_list_response( + WXUSER_RECORDS[2:], total=TOTAL, page_num=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_wxuser_list_response( + [], total=TOTAL, page_num=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(WXUSER_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_accounts( + 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 4 fixture records (not duplicated, not dropped). + assert len(results) == 4 + assert {r["wx_id"] for r in results} == { + rec["wechatId"] for rec in WXUSER_RECORDS + } + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, WXUSER_RECORDS, strict=True): + # Identity. + assert result["wx_id"] == record["wechatId"] + + # Naming. + assert result["nick_name"] == record["nickName"] + assert result["alias"] == record["alias"] + + # Phones. + assert result["phone"] == record["phone"] + + # Embedded staff fields (the owning employee). + assert result["user_id"] == record["userId"] + assert result["user_name"] == record["userName"] + + # Counts. + assert result["contacts_num"] == record["contactsNum"] + assert result["group_num"] == record["groupNum"] + assert result["send_num"] == record["sendNum"] + assert result["receive_num"] == record["receiveNum"] + + # Status / sizing / timestamps. + assert result["is_on_job"] == record["isOnJob"] + assert result["db_size"] == record["dbSize"] + assert result["online_status"] == record["onlineStatus"] + assert result["last_update"] == record["lastUpdate"] + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# Pagination body: pageNum/pageSize advanced correctly +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_accounts_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_wxuser_list_response( + WXUSER_RECORDS[:1], total=TOTAL, page_num=1, page_size=1 + ) + page2 = make_wxuser_list_response( + [], total=TOTAL, page_num=2, page_size=1 + ) + route = respx.post(WXUSER_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_accounts(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 + + # First request: pageNum=1. + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + # Second request: pageNum advanced to 2. + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +# --------------------------------------------------------------------------- +# Empty first page -> no yields, single call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_accounts_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_wxuser_list_response([], total=0, page_num=1) + route = respx.post(WXUSER_LIST_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_wechat_accounts(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1