# CPS 群详情 · 每日明细「按天 · 按用户」领券下钻 — 实现计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 在 CPS 群详情页每日明细表每行加「明细」按钮,点开弹窗展示该群该天以用户为单位的领券/点击,并在「点击次数」上悬停展示该用户当天点过的优惠券(券名 ×次数)。 **Architecture:** 后端 `shaguabijia-app-server` 新增只读聚合接口 `/admin/api/cps/groups/{id}/day-users?date=YYYY-MM-DD`(券明细内联返回),并把既有 `/daily` 的 `date` 由 `MM-DD` 改全为 `YYYY-MM-DD` 供下钻定位整天。前端 `shaguabijia-admin-web` 加明细列 + Modal + Tooltip,一次请求渲染。纯读侧聚合,**不动数据库**。 **Tech Stack:** 后端 FastAPI + SQLAlchemy + pytest(`.venv`);前端 Next.js 15 + Ant Design 5 + TypeScript(无测试框架,门槛为 `npm run build`)。 **配套 spec:** `docs/superpowers/specs/2026-06-26-cps-day-user-drilldown-design.md` --- ## 仓库与分支 - **前端** `e:\project\shaguabijia-admin-web`:已在分支 `feat/cps-day-user-drilldown`。 - **后端** `e:\project\shaguabijia-app-server`:独立 git 仓,需自建同名分支(Task 1 Step 0)。 后端命令统一**从后端仓根目录运行**;git 用 `git -C e:\project\shaguabijia-app-server ...` 避免切目录。 ## File Structure | 文件 | 改动 | 职责 | |---|---|---| | `app/admin/routers/cps.py`(后端) | 改 1 行 + 加 1 个 endpoint | `/daily` 的 `date` 改全日期;新增 `/day-users` 路由(鉴权、日期解析、窗口计算) | | `app/admin/repositories/cps.py`(后端) | 加 1 个函数 + 补 1 处 import | `group_day_users(...)`:按 openid 聚合领券/点击 + 券明细(Python 侧) | | `tests/test_cps_admin.py`(后端) | 新建 | admin CPS 端点测试:`/daily` 日期格式 + `/day-users` 聚合与边界 | | `src/app/(main)/cps/groups/[id]/page.tsx`(前端) | 改 | 日期列加宽、明细列、Modal、Tooltip、按用户表 | --- ## Task 1: 后端 `/daily` 的 `date` 改为 `YYYY-MM-DD` **Files:** - Modify: `app/admin/routers/cps.py:472`(`group_daily` 内构行处) - Test: `tests/test_cps_admin.py`(新建) - [ ] **Step 0: 在后端仓建分支** 先看工作区状态,避免把无关改动带入: Run: ``` git -C e:\project\shaguabijia-app-server status --short git -C e:\project\shaguabijia-app-server checkout -b feat/cps-day-user-drilldown ``` Expected: 切到新分支 `feat/cps-day-user-drilldown`。 - [ ] **Step 1: 写失败测试(新建测试文件,含共用 fixtures + 种子)** 新建 `tests/test_cps_admin.py`,内容如下(fixtures 仿 `tests/test_admin_read.py` 自包含;种子函数后续 Task 复用): ```python """admin CPS 端点测试:每日明细 date 格式 + 按天按用户领券下钻(/day-users)。""" from __future__ import annotations from datetime import date, datetime, timedelta, timezone import pytest from fastapi.testclient import TestClient from app.admin.main import admin_app from app.admin.repositories import admin_user as admin_repo from app.admin.repositories import cps as cps_repo from app.db.session import SessionLocal from app.models.cps_link import CpsClick from app.models.cps_wx_user import CpsWxUser from app.repositories import cps_link as cps_link_repo _BJ = timezone(timedelta(hours=8)) @pytest.fixture() def admin_client() -> TestClient: return TestClient(admin_app) @pytest.fixture() def admin_token() -> str: db = SessionLocal() try: if admin_repo.get_by_username(db, "cps_test_admin") is None: admin_repo.create_admin( db, username="cps_test_admin", password="cps-pass", role="super_admin" ) finally: db.close() c = TestClient(admin_app) r = c.post("/admin/api/auth/login", json={"username": "cps_test_admin", "password": "cps-pass"}) return r.json()["access_token"] def _auth(token: str) -> dict: return {"Authorization": f"Bearer {token}"} def _seed_cps_day(day_bj: date) -> tuple[int, str]: """造淘宝群 + 2 个活动(券) + 2 条 link + day_bj 当天点击(含匿名 + 次日各 1) + 授权画像。 o_user(本群唯一):券A visit×3、券B visit×2、copy×2 → visit_count=5、copy_count=2。 另造 1 条匿名 visit(openid=None)与 1 条次日 visit,均不应计入当天该用户聚合。 clicked_at 统一存 tz-aware UTC(SQLite DateTime 按字段渲染、忽略 tzinfo; 转 UTC 后字段即 UTC 墙钟,与 repo 侧 _as_utc 比较口径一致)。返回 (group_id, openid)。 """ db = SessionLocal() try: gid = cps_repo.create_group(db, name="下钻测试群", platforms=["taobao"]).id openid = f"od_user_{gid}" # 按群唯一,规避 CpsWxUser.openid 唯一约束跨用例冲突 a1 = cps_repo.create_activity(db, name="618神券", platform="taobao", payload="tkl-1") a2 = cps_repo.create_activity(db, name="买一送一", platform="taobao", payload="tkl-2") link1 = cps_link_repo.create_link( db, group_id=gid, activity_id=a1.id, sid=None, target_url="t1", platform="taobao" ) link2 = cps_link_repo.create_link( db, group_id=gid, activity_id=a2.id, sid=None, target_url="t2", platform="taobao" ) db.add(CpsWxUser(openid=openid, nickname="张三", headimgurl="https://h/1")) db.commit() def at(hour: int) -> datetime: return datetime(day_bj.year, day_bj.month, day_bj.day, hour, tzinfo=_BJ).astimezone( timezone.utc ) for _ in range(3): db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=at(10))) for _ in range(2): db.add(CpsClick(link_id=link2.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=at(11))) for _ in range(2): db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="copy", openid=openid, clicked_at=at(12))) # 匿名点击 — 不计入 db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=None, clicked_at=at(13))) # 次日点击 — 验证时间窗,不计入当天 nxt = (datetime(day_bj.year, day_bj.month, day_bj.day, 10, tzinfo=_BJ) + timedelta(days=1)).astimezone(timezone.utc) db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=nxt)) db.commit() return gid, openid finally: db.close() def test_daily_date_is_full_iso(admin_client: TestClient, admin_token: str) -> None: """/daily 每行 date 改为 YYYY-MM-DD(10 字符、两个连字符),不再是 MM-DD。""" gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/daily", params={"days": 3}, headers=_auth(admin_token) ) assert r.status_code == 200, r.text rows = r.json()["rows"] assert rows, "应有按天补零行" for row in rows: assert len(row["date"]) == 10 and row["date"].count("-") == 2, row["date"] ``` - [ ] **Step 2: 跑测试确认失败** Run(从后端仓根目录 `e:\project\shaguabijia-app-server`): ``` .\.venv\Scripts\pytest.exe tests\test_cps_admin.py::test_daily_date_is_full_iso -v ``` Expected: FAIL —— `date` 当前为 `"%m-%d"`(长度 5),断言 `len==10` 不通过。 - [ ] **Step 3: 改 router 的日期格式** `app/admin/routers/cps.py` 的 `group_daily` 内(约 472 行),把构行的日期字段改为全日期: ```python row = { "date": cur.strftime("%Y-%m-%d"), "click_pv": cp["click_pv"] if cp else 0, "click_uv": cp["click_uv"] if cp else 0, "copy_pv": cp["copy_pv"] if cp else 0, } ``` (仅把原 `cur.strftime("%m-%d")` 改成 `cur.strftime("%Y-%m-%d")`,其余不动。) - [ ] **Step 4: 跑测试确认通过** Run: ``` .\.venv\Scripts\pytest.exe tests\test_cps_admin.py::test_daily_date_is_full_iso -v ``` Expected: PASS。 - [ ] **Step 5: 提交(后端仓)** ``` git -C e:\project\shaguabijia-app-server add app/admin/routers/cps.py tests/test_cps_admin.py git -C e:\project\shaguabijia-app-server commit -m "feat(cps): /daily 的 date 改为 YYYY-MM-DD(下钻定位整天用)" ``` --- ## Task 2: 后端新增 `group_day_users` 聚合 + `/day-users` 接口 **Files:** - Modify: `app/admin/repositories/cps.py:20`(import 补 `CpsLink`)、文件末尾加函数 - Modify: `app/admin/routers/cps.py`(`group_wx_users` 路由后追加新路由) - Test: `tests/test_cps_admin.py`(追加) - [ ] **Step 1: 写失败测试(聚合 happy path,含匿名/跨天排除、券倒序、合计自洽)** 在 `tests/test_cps_admin.py` 末尾追加: ```python def test_day_users_aggregates(admin_client: TestClient, admin_token: str) -> None: """当天该群:仅授权用户;领券=copy、点击=visit;coupons=visit 券×次数倒序、合计=点击次数。""" gid, openid = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026-06-25"}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text body = r.json() assert body["group_id"] == gid assert body["date"] == "2026-06-25" users = body["users"] assert len(users) == 1 # 匿名不计、次日不计 u = users[0] assert u["openid"] == openid assert u["nickname"] == "张三" assert u["headimgurl"] == "https://h/1" assert u["copy_count"] == 2 assert u["visit_count"] == 5 assert [c["name"] for c in u["coupons"]] == ["618神券", "买一送一"] assert [c["count"] for c in u["coupons"]] == [3, 2] assert sum(c["count"] for c in u["coupons"]) == u["visit_count"] ``` - [ ] **Step 2: 跑测试确认失败** Run: ``` .\.venv\Scripts\pytest.exe tests\test_cps_admin.py::test_day_users_aggregates -v ``` Expected: FAIL —— 404/路由不存在(`/day-users` 尚未实现)。 - [ ] **Step 3: 补 repo import** `app/admin/repositories/cps.py` 第 20 行,给 `cps_link` 的 import 补上 `CpsLink`: ```python from app.models.cps_link import CpsClick, CpsLink ``` - [ ] **Step 4: 加 repo 聚合函数** 在 `app/admin/repositories/cps.py` 末尾(`group_wx_users` 之后)追加: ```python def group_day_users( db: Session, *, group_id: int, start: datetime, end: datetime, limit: int = 200, ) -> list[dict]: """该群某天(北京)以用户为单位的领券/点击 + 每人 visit 过的券。 时间窗为半开区间 [start, end)(end=次日 00:00),避免午夜双计。只统计 openid 非空 (可归属到人)的点击 —— 匿名点击(美团/京东 302 多为匿名)不计入。券名 = 该点击 link 对应活动名;活动被硬删则兜底 活动#{id}。copy=领券次数、visit=点击次数;coupons 仅 取 visit 事件按活动分组、按次数倒序(合计 = visit_count)。排序:领券 desc、再点击 desc。 与 group_wx_users 同风格(Python 侧聚合,跨 PG/SQLite 无方言坑)。 注:每日明细行的 click_pv/copy_pv 计全部点击(含匿名、UV 按 ip,ua);本函数只计 openid 用户,故各用户求和 <= 当天行总数,二者口径不同、不必相等。 """ rows = db.execute( select(CpsClick.openid, CpsClick.event_type, CpsClick.link_id) .where(CpsClick.group_id == group_id) .where(CpsClick.clicked_at >= _as_utc(start)) .where(CpsClick.clicked_at < _as_utc(end)) .where(CpsClick.openid.is_not(None)) ).all() if not rows: return [] # link_id -> activity_id -> 券名(活动名) link_ids = {r.link_id for r in rows} link_to_act = dict( db.execute( select(CpsLink.id, CpsLink.activity_id).where(CpsLink.id.in_(link_ids)) ).all() ) act_ids = {aid for aid in link_to_act.values() if aid is not None} act_name = ( dict( db.execute( select(CpsActivity.id, CpsActivity.name).where(CpsActivity.id.in_(act_ids)) ).all() ) if act_ids else {} ) def _coupon_name(link_id: int) -> str: aid = link_to_act.get(link_id) if aid is None: return f"链接#{link_id}" return act_name.get(aid) or f"活动#{aid}" stat: dict[str, dict] = {} for openid, event_type, link_id in rows: s = stat.setdefault(openid, {"copy": 0, "visit": 0, "coupons": {}}) if event_type == "copy": s["copy"] += 1 else: s["visit"] += 1 name = _coupon_name(link_id) s["coupons"][name] = s["coupons"].get(name, 0) + 1 openids = list(stat.keys()) users = { u.openid: u for u in db.execute( select(CpsWxUser).where(CpsWxUser.openid.in_(openids)) ).scalars().all() } result = [ { "openid": openid, "nickname": users[openid].nickname if openid in users else None, "headimgurl": users[openid].headimgurl if openid in users else None, "copy_count": s["copy"], "visit_count": s["visit"], "coupons": [ {"name": name, "count": cnt} for name, cnt in sorted( s["coupons"].items(), key=lambda kv: kv[1], reverse=True ) ], } for openid, s in stat.items() ] result.sort(key=lambda x: (x["copy_count"], x["visit_count"]), reverse=True) return result[:limit] ``` - [ ] **Step 5: 加 router 端点** `app/admin/routers/cps.py`,在 `group_wx_users` 路由(文件末尾那个)之后追加: ```python @router.get("/groups/{group_id}/day-users", summary="某天该群按用户的领券/点击 + 每人点过的券") def group_day_users( group_id: int, db: AdminDb, date: Annotated[str, Query(description="北京日期 YYYY-MM-DD")], ) -> dict: group = cps_repo.get_group(db, group_id) if group is None: raise HTTPException(status_code=404, detail="群不存在") bj = timezone(timedelta(hours=8)) try: day0 = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=bj) except ValueError as e: raise HTTPException(status_code=400, detail="date 格式应为 YYYY-MM-DD") from e start = day0.replace(hour=0, minute=0, second=0, microsecond=0) end = start + timedelta(days=1) users = cps_repo.group_day_users(db, group_id=group_id, start=start, end=end) return { "group_id": group.id, "group_name": group.name, "date": date, "users": users, } ``` (`datetime`/`timedelta`/`timezone`、`Annotated`、`Query`、`HTTPException` 该文件均已 import;无需新增。) - [ ] **Step 6: 跑测试确认通过** Run: ``` .\.venv\Scripts\pytest.exe tests\test_cps_admin.py::test_day_users_aggregates -v ``` Expected: PASS。 - [ ] **Step 7: 提交(后端仓)** ``` git -C e:\project\shaguabijia-app-server add app/admin/repositories/cps.py app/admin/routers/cps.py tests/test_cps_admin.py git -C e:\project\shaguabijia-app-server commit -m "feat(cps): 新增 /day-users 按天按用户领券下钻接口" ``` --- ## Task 3: 后端 `/day-users` 边界用例(404 / 400 / 401 / 空天 / 跨年) **Files:** - Test: `tests/test_cps_admin.py`(追加) > 这些断言验证 Task 2 实现已覆盖的边界。多数应直接通过;若某条失败,回到 Task 2 对应分支修实现。 - [ ] **Step 1: 追加边界测试** 在 `tests/test_cps_admin.py` 末尾追加: ```python def test_day_users_group_not_found(admin_client: TestClient, admin_token: str) -> None: r = admin_client.get( "/admin/api/cps/groups/999999/day-users", params={"date": "2026-06-25"}, headers=_auth(admin_token), ) assert r.status_code == 404 def test_day_users_bad_date(admin_client: TestClient, admin_token: str) -> None: gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026/06/25"}, # 非 YYYY-MM-DD headers=_auth(admin_token), ) assert r.status_code == 400 def test_day_users_requires_auth(admin_client: TestClient) -> None: r = admin_client.get( "/admin/api/cps/groups/1/day-users", params={"date": "2026-06-25"} ) assert r.status_code == 401 def test_day_users_empty_when_no_clicks(admin_client: TestClient, admin_token: str) -> None: gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026-06-20"}, # 该群当天无任何点击 headers=_auth(admin_token), ) assert r.status_code == 200 assert r.json()["users"] == [] def test_day_users_cross_year(admin_client: TestClient, admin_token: str) -> None: """跨年:YYYY-MM-DD 才能精确定位 12-31(MM-DD 会丢年份);次日(次年 01-01)不计入。""" gid, openid = _seed_cps_day(date(2025, 12, 31)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2025-12-31"}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text users = r.json()["users"] assert len(users) == 1 assert users[0]["openid"] == openid assert users[0]["visit_count"] == 5 # 次年 01-01 那条被时间窗排除 ``` - [ ] **Step 2: 跑整个测试文件确认全绿** Run: ``` .\.venv\Scripts\pytest.exe tests\test_cps_admin.py -v ``` Expected: 全部 PASS(共 7 个用例:daily 日期、aggregates、404、bad_date、auth、empty、cross_year)。 - [ ] **Step 3: 跑后端全量回归(确认没碰坏既有)** Run: ``` .\.venv\Scripts\pytest.exe -q ``` Expected: 全绿(与改动前同样的通过数 + 新增 7 条)。 - [ ] **Step 4: 提交(后端仓)** ``` git -C e:\project\shaguabijia-app-server add tests/test_cps_admin.py git -C e:\project\shaguabijia-app-server commit -m "test(cps): /day-users 边界用例(404/400/401/空天/跨年)" ``` --- ## Task 4: 前端日期列显示全日期并加宽 **Files:** - Modify: `src/app/(main)/cps/groups/[id]/page.tsx` 前端无测试框架,门槛为类型/构建通过(`npm run build`)。 - [ ] **Step 1: 加宽日期列** `dailyColumns` 第一列(约 174 行)`width: 64` → `width: 96`: ```tsx { title: '日期', dataIndex: 'date', width: 96, fixed: 'left' }, ``` (`date` 现由后端给全 `YYYY-MM-DD`,直接显示即可;`DailyRow.date` 类型仍是 `string`,无需改接口。) - [ ] **Step 2: 构建确认通过** Run(前端仓根目录): ``` npm run build ``` Expected: 构建成功,无类型错误。 - [ ] **Step 3: 提交(前端仓)** ``` git add src/app/(main)/cps/groups/[id]/page.tsx git commit -m "feat(cps): 每日明细日期列改显示全日期并加宽" ``` --- ## Task 5: 前端下钻 — 明细列 + Modal + Tooltip **Files:** - Modify: `src/app/(main)/cps/groups/[id]/page.tsx` - [ ] **Step 1: antd 引入 Modal、Tooltip** 第 9 行 import 增加 `Modal, Tooltip`(保持字母序即可): ```tsx import { Button, Card, Col, Empty, Modal, Row, Segmented, Space, Spin, Statistic, Table, Tooltip, message } from 'antd'; ``` - [ ] **Step 2: 加按天用户类型** 在 `WxUser` 接口定义之后(约 58 行后)追加: ```tsx // 某天该群按用户领券下钻(/day-users) interface DayCoupon { name: string; count: number; } interface DayUser { openid: string; nickname: string | null; headimgurl: string | null; copy_count: number; visit_count: number; coupons: DayCoupon[]; } ``` - [ ] **Step 3: 加弹窗状态 + 拉取 handler** 在组件内 `wxUsers` 相关 `useEffect` 之后(约 140 行后)追加。注意须在 `dailyColumns` 定义之前,供其按钮引用: ```tsx // 某天领券下钻弹窗 const [dayDetail, setDayDetail] = useState<{ open: boolean; date: string; loading: boolean; users: DayUser[]; }>({ open: false, date: '', loading: false, users: [] }); const openDayDetail = useCallback( async (date: string) => { setDayDetail({ open: true, date, loading: true, users: [] }); try { const r = await api.get<{ users: DayUser[] }>( `/admin/api/cps/groups/${id}/day-users?date=${date}`, ); setDayDetail((s) => ({ ...s, loading: false, users: r.data.users || [] })); } catch (e) { message.error(errMsg(e, '明细加载失败')); setDayDetail((s) => ({ ...s, loading: false })); } }, [id], ); ``` - [ ] **Step 4: 每日明细表加「明细」列 + 调大 scroll.x** 在 `dailyColumns` 数组末尾(`结算佣金` 列之后,约 226 行)追加一列: ```tsx { title: '明细', key: 'drill', width: 64, fixed: 'right', render: (_, r) => ( ), }, ``` 同时,每日明细 `