feat(cps): 群点击时序聚合端点(天/小时级 PV/UV/复制,补零连续桶)

- repo group_click_timeseries:北京时区分桶,visit/copy 的 PV/UV,无数据补 0
- GET /cps/groups/{id}/timeseries?granularity=day&days=N | granularity=hour

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/admin/repositories/cps.py

@@ -5,7 +5,7 @@
 """
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from uuid import uuid4
 
@@ -17,12 +17,16 @@ from app.integrations import meituan
 from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
+from app.models.cps_link import CpsClick
 from app.models.cps_order import CpsOrder
 
 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
 _SETTLED_STATUS = "6"
 
+# CPS 点击时序按北京时区分桶(运营看的是北京时间)
+_BJ_TZ = timezone(timedelta(hours=8))
+
 
 # ───────────── 单位换算 ─────────────
 def _yuan_to_cents(v: object) -> int | None:
@@ -370,3 +374,54 @@ def group_stats(
     # 佣金降序(None 当 0),次按点击
     rows.sort(key=lambda x: ((x["est_commission_cents"] or 0), x["click_pv"]), reverse=True)
     return rows
+
+
+def group_click_timeseries(
+    db: Session, *, group_id: int, granularity: str,
+    start: datetime, end: datetime,
+) -> list[dict]:
+    """该群点击时序(北京时区分桶,granularity=day|hour)。
+
+    每桶:click_pv/uv(visit)+copy_pv/uv;UV=distinct(ip,ua)。**生成连续桶(无数据补 0)**,
+    让折线图 x 轴按所选范围铺满、不断裂(CPS 数据稀疏,补零是关键)。
+    """
+    rows = db.execute(
+        select(CpsClick.clicked_at, CpsClick.event_type, CpsClick.ip, CpsClick.ua)
+        .where(CpsClick.group_id == group_id)
+        .where(CpsClick.clicked_at >= _as_utc(start))
+        .where(CpsClick.clicked_at <= _as_utc(end))
+    ).all()
+
+    agg: dict = {}  # bucket_key -> {vp, vu(set), cp, cu(set)}
+    for clicked_at, event_type, ip, ua in rows:
+        bj = clicked_at.astimezone(_BJ_TZ)
+        key = bj.strftime("%Y-%m-%d") if granularity == "day" else bj.hour
+        b = agg.setdefault(key, {"vp": 0, "vu": set(), "cp": 0, "cu": set()})
+        uid = f"{ip or ''}|{ua or ''}"
+        if event_type == "copy":
+            b["cp"] += 1
+            b["cu"].add(uid)
+        else:
+            b["vp"] += 1
+            b["vu"].add(uid)
+
+    def _point(label: str, b: dict | None) -> dict:
+        return {
+            "label": label,
+            "click_pv": b["vp"] if b else 0,
+            "click_uv": len(b["vu"]) if b else 0,
+            "copy_pv": b["cp"] if b else 0,
+            "copy_uv": len(b["cu"]) if b else 0,
+        }
+
+    points: list[dict] = []
+    if granularity == "day":
+        cur = start.astimezone(_BJ_TZ).date()
+        last = end.astimezone(_BJ_TZ).date()
+        while cur <= last:
+            points.append(_point(cur.strftime("%m-%d"), agg.get(cur.strftime("%Y-%m-%d"))))
+            cur += timedelta(days=1)
+    else:
+        for h in range(24):
+            points.append(_point(f"{h:02d}:00", agg.get(h)))
+    return points

app/admin/routers/cps.py

@@ -396,3 +396,40 @@ def get_stats(
         total_est_commission_cents=sum(s.est_commission_cents or 0 for s in stats),
         total_settled_commission_cents=sum(s.settled_commission_cents or 0 for s in stats),
     )
+
+
+@router.get("/groups/{group_id}/timeseries", summary="群点击时序(折线图:天/小时级 PV/UV/复制)")
+def group_timeseries(
+    group_id: int,
+    db: AdminDb,
+    granularity: Annotated[str, Query(pattern="^(day|hour)$")] = "day",
+    days: Annotated[int, Query(ge=1, le=30)] = 7,
+    date: Annotated[str | None, Query()] = None,  # hour 粒度看哪天(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))
+    now_bj = datetime.now(bj)
+    if granularity == "hour":
+        if date:
+            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
+        else:
+            day0 = now_bj
+        start = day0.replace(hour=0, minute=0, second=0, microsecond=0)
+        end = start + timedelta(days=1) - timedelta(microseconds=1)
+    else:
+        start = (now_bj - timedelta(days=days - 1)).replace(hour=0, minute=0, second=0, microsecond=0)
+        end = now_bj
+    points = cps_repo.group_click_timeseries(
+        db, group_id=group_id, granularity=granularity, start=start, end=end
+    )
+    return {
+        "group_id": group.id,
+        "group_name": group.name,
+        "granularity": granularity,
+        "points": points,
+    }