fix: wechat account scrape (form-data+dates) + frontend staff page

Backend (scrape_wechat_accounts):
- Send wxUser/list as form-urlencoded (not JSON) with the REQUIRED filters
  isOnJob / departmentId / startdate / enddate + empty userId/salesname/
  weChatNickName, matching the live request (empty fails with code -90000).
- scrape_wechat_accounts now takes department_id (the company-root dept id)
  plus optional start/end defaulting to a recent 7-day UTC window.
- paginate_by_page gains a form=True mode (data= + form content-type).
- Orchestrator sync_wechat_accounts resolves the root dept from the
  Department table (or accepts one) and skips with a warning when unknown.
- seed_accounts.py resolves the root via scrape_departments.

Frontend (yunkes):
- Add StaffList.vue (/staff) with is_on_job filter + pagination.
- Add /staff route and a 员工 nav item between 客户 and 群.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-01 17:11:17 +08:00
parent 4dcfc66741
commit 0df53c6b1a
10 changed files with 638 additions and 30 deletions
+36 -2
View File
@@ -25,6 +25,13 @@ over the response shape via three callables:
* ``build_body(page_num, page_size, **extra) -> dict`` — build the request
body for a given page.
By default the body is sent as JSON. A few yunke endpoints
(e.g. ``msg/chats``, ``wxUser/list``) bind ``@RequestParam`` from
**form-urlencoded** bodies instead — pass ``form=True`` to switch the paginator
to ``data=<body>`` with content-type
``application/x-www-form-urlencoded`` (mirroring the form-encoded
``msg/chats`` request the message scraper already makes directly).
``paginate_by_cursor`` is decoupled from the client entirely: it takes a
single ``fetch(cursor)`` coroutine that performs whatever request the caller
needs and returns ``(rows, next_cursor, has_more)``. The cursor-paginator's
@@ -45,7 +52,17 @@ if TYPE_CHECKING:
# :class:`RateLimiter` at runtime without forming an import cycle.
from app.session.yunke_client import YunkeClient
__all__ = ["RateLimiter", "paginate_by_cursor", "paginate_by_page"]
__all__ = [
"FORM_CONTENT_TYPE",
"RateLimiter",
"paginate_by_cursor",
"paginate_by_page",
]
# Content type for form-encoded bodies. Several yunke endpoints (msg/chats,
# wxUser/list) bind @RequestParam from form data rather than JSON, so the
# paginator + the manual message fetcher both send this content-type.
FORM_CONTENT_TYPE = "application/x-www-form-urlencoded"
# Informal type aliases for the response-handling callables. The yunke
# response shapes are dict-shaped JSON and do not benefit from finer typing.
@@ -65,6 +82,7 @@ async def paginate_by_page(
parse_rows: ParseRows,
build_body: BuildBody,
rate_limiter: RateLimiter | None = None,
form: bool = False,
**extra: Any,
) -> AsyncIterator[dict[str, Any]]:
"""Walk a page-numbered list endpoint, yielding one dict per record.
@@ -85,6 +103,15 @@ async def paginate_by_page(
``rate_limiter`` is an optional :class:`RateLimiter`; when supplied, it is
acquired before every upstream call.
``form`` selects the wire encoding of ``build_body``'s output. The default
(``False``) sends the body as JSON (``client.post(path, json=body)``) —
correct for most yunke list endpoints. Pass ``True`` for endpoints that
bind ``@RequestParam`` from a form-urlencoded body (``msg/chats``,
``wxUser/list``): the paginator then sends
``client.post(path, data=body, headers={"content-type": ...})`` and lets
httpx URL-encode the dict. Either way ``build_body`` returns the same
plain dict of field names → values.
"""
page_num = 1
total: int | None = None
@@ -95,7 +122,14 @@ async def paginate_by_page(
await rate_limiter.acquire()
body = build_body(page_num, page_size, **extra)
resp = await client.post(path, json=body)
if form:
resp = await client.post(
path,
data=body,
headers={"content-type": FORM_CONTENT_TYPE},
)
else:
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
+88 -7
View File
@@ -13,8 +13,12 @@ Endpoints covered today:
``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).
embedded ``userId``/``userName`` staff pair). The endpoint binds
``@RequestParam`` from a **form-urlencoded** body (NOT JSON) and REQUIRES
``isOnJob`` / ``departmentId`` (the company-root dept id) / ``startdate`` /
``enddate`` or it answers ``code -90000`` "缺少..."; a recent ~7-day
window works (the date affects per-account stats, not which accounts are
returned). Page-number pagination (Task 2.1).
* :func:`scrape_wechat_groups` — wraps ``POST /wechatweb/chatroom/getChatroomList``
and yields normalized :class:`app.db.models.ChatGroup` dicts, parsing the
semicolon-joined ``memberIds`` into a ``member_ids_list`` so the
@@ -58,6 +62,7 @@ disambiguations where the source has two fields for one column:
from __future__ import annotations
import datetime as dt
from collections.abc import AsyncIterator
from typing import Any
@@ -660,13 +665,57 @@ def _parse_wxuser_rows(resp: Any) -> list[dict[str, Any]]:
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.
def _build_wxuser_body(
page_num: int,
page_size: int,
*,
department_id: str,
is_on_job: str = "1",
start: str,
end: str,
) -> dict[str, Any]:
"""Build the form-urlencoded ``wxUser/list`` 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.
Phase-0 live probing established the real contract: the endpoint binds
Spring ``@RequestParam`` from a form body (NOT JSON) and REQUIRES the
filters below or it answers ``code -90000`` ("缺少..."). Empty
``userId`` / ``salesname`` / ``weChatNickName`` are sent explicitly because
the real client does, and the upstream treats missing keys differently
from empty strings for some filters.
"""
return {"pageNum": page_num, "pageSize": page_size}
return {
"pageNum": page_num,
"pageSize": page_size,
# REQUIRED — empty fails with code -90000. "1" => on-job accounts only
# (the 55 live rows). "0" would pull departed staff instead.
"isOnJob": is_on_job,
# REQUIRED — the company ROOT dept id (caller obtains it from
# scrape_departments). NOT hardcoded here.
"departmentId": department_id,
# REQUIRED — "缺少时间参数" if absent. YYYY-MM-DD strings. The window
# affects per-account stats, NOT which accounts are returned.
"startdate": start,
"enddate": end,
# Always-empty free-text filters (sent to mirror the real client).
"userId": "",
"salesname": "",
"weChatNickName": "",
}
def _default_date_window(now: dt.datetime | None = None) -> tuple[str, str]:
"""A recent 7-day window as ``(start, end)`` ``YYYY-MM-DD`` strings.
The ``wxUser/list`` stats window is cosmetic (it affects per-account
counters, not which accounts are returned), so any recent window works.
We default to the last 7 days measured from ``now`` (UTC by default to
stay deterministic and match the rest of the scraper layer's naive-UTC
convention). Both bounds are inclusive calendar days.
"""
now = now or dt.datetime.now(dt.UTC)
end_date = now.date()
start_date = end_date - dt.timedelta(days=7)
return start_date.isoformat(), end_date.isoformat()
def _normalize_wxuser_account(record: dict[str, Any]) -> dict[str, Any]:
@@ -702,7 +751,11 @@ def _normalize_wxuser_account(record: dict[str, Any]) -> dict[str, Any]:
async def scrape_wechat_accounts(
client: YunkeClient,
*,
department_id: str,
page_size: int = 50,
is_on_job: str = "1",
start: str | None = None,
end: str | None = None,
) -> AsyncIterator[dict[str, Any]]:
"""Yield normalized WeChat staff-account records from ``wxUser/list``.
@@ -711,6 +764,13 @@ async def scrape_wechat_accounts(
mapping each record to the ``WechatAccount`` column shape plus a verbatim
``raw_json`` copy.
The request is **form-urlencoded** (NOT JSON): the endpoint binds Spring
``@RequestParam`` from the form body and REQUIRES ``isOnJob`` /
``departmentId`` / ``startdate`` / ``enddate``. ``department_id`` is the
company-root dept id (the caller resolves it from
:func:`scrape_departments`); it is NOT hardcoded here so this scraper
stays account-agnostic.
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``).
@@ -720,10 +780,26 @@ async def scrape_wechat_accounts(
client:
Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are
handled by the client.
department_id:
Company-root department id (REQUIRED upstream filter). The caller
obtains it from :func:`scrape_departments` (the root node whose
``parent_id`` is falsy).
page_size:
Page size to request. Defaults to 50 (matches the live capture's
observed page size for this endpoint).
is_on_job:
``"1"`` (default) for on-job staff, ``"0"`` for departed. Required
upstream filter — empty fails with ``code -90000``.
start, end:
Optional ``YYYY-MM-DD`` bounds for the per-account stats window. The
date affects per-account counters, NOT which accounts are returned
(live total is 55 regardless), so any recent window works. Defaults
to a recent 7-day window computed from ``datetime.now(UTC)``.
"""
default_start, default_end = _default_date_window()
window_start = start or default_start
window_end = end or default_end
async for record in paginate_by_page(
client,
WXUSER_LIST_PATH,
@@ -731,6 +807,11 @@ async def scrape_wechat_accounts(
parse_total=_parse_wxuser_total,
parse_rows=_parse_wxuser_rows,
build_body=_build_wxuser_body,
form=True,
department_id=department_id,
is_on_job=is_on_job,
start=window_start,
end=window_end,
):
yield _normalize_wxuser_account(record)
+61 -4
View File
@@ -43,9 +43,10 @@ import logging
from collections.abc import AsyncIterator, Callable
from typing import Any
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import SyncRun
from app.db.models import Department, SyncRun
from app.scraper.call import scrape_phone_calls, scrape_sms
from app.scraper.media import download_message_media
from app.scraper.wechat import (
@@ -385,16 +386,72 @@ async def sync_departments(
)
async def _find_root_department_id(
session: AsyncSession,
) -> str | None:
"""Resolve the company-root department id from the mirrored ``Department`` table.
The ``wxUser/list`` form body REQUIRES ``departmentId`` (the company-root
dept id) — empty fails with ``code -90000``. The root is a node whose
``parent_id`` is falsy (``None`` / ``""`` / ``"0"`` — the yunke tree
convention). If several roots exist, the first by ``dept_id`` is returned
(with a warning) so a multi-root org still scrapes. Returns ``None`` when
no root is present so the caller can skip gracefully rather than send a
known-bad empty filter.
"""
stmt = (
sa.select(Department)
.where(
sa.or_(
Department.parent_id.is_(None),
Department.parent_id == "",
Department.parent_id == "0",
)
)
.order_by(Department.dept_id.asc())
.limit(1)
)
row = (await session.execute(stmt)).scalars().first()
return row.dept_id if row is not None else None
async def sync_wechat_accounts(
session: AsyncSession, client: YunkeClient
session: AsyncSession,
client: YunkeClient,
*,
department_id: str | None = None,
) -> SyncResult:
"""Full re-scrape of all WeChat staff accounts (+ embedded staff rows)."""
"""Full re-scrape of all WeChat staff accounts (+ embedded staff rows).
``department_id`` is the company-root dept id required by the
``wxUser/list`` form body. When omitted, the orchestrator resolves it from
the mirrored ``Department`` table (a root node whose ``parent_id`` is
falsy) — which is why :func:`sync_departments` must run first. If no root
can be resolved, the scrape is SKIPPED with a logged warning (returns a
``status='skipped'`` :class:`SyncResult` and records no ``SyncRun``) so a
missing org tree does not abort a multi-entity batch.
"""
resolved = department_id
if resolved is None:
resolved = await _find_root_department_id(session)
if not resolved:
logger.warning(
"sync_wechat_accounts: no root department_id resolved"
" (sync departments first?); skipping wxUser/list scrape"
)
return SyncResult(
entity=ENTITY_WECHAT_ACCOUNT,
status="skipped",
fetched=0,
upserted=0,
error="no root department_id available",
)
return await _sync_full_rescrape(
session,
entity_type=ENTITY_WECHAT_ACCOUNT,
channel=None,
scope_id=None,
scrape=lambda: scrape_wechat_accounts(client),
scrape=lambda: scrape_wechat_accounts(client, department_id=resolved),
upsert=upsert_wechat_account,
)