feat(cps): 群发券分发与对账 + 短链点击追踪

群(sid)/活动池/转链(带sid)/美团 query_order 按 sid 对账/按群统计;
短链 /c/{code} 记点击(PV/UV)→302 跳美团,点击→下单漏斗。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-17 10:00:29 +08:00
parent 9d93b70b9b
commit 277f9b16a2
16 changed files with 1187 additions and 2 deletions
+61
View File
@@ -0,0 +1,61 @@
"""cps_link / cps_click tables (群发短链 + 点击统计)
Revision ID: cps_link_tables
Revises: cps_tables
Create Date: 2026-06-16 13:20:00.000000
群发券链接套一层 /c/{code} 做点击统计:cps_link(短码→美团目标) + cps_click(点击事件)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "cps_link_tables"
down_revision: Union[str, Sequence[str], None] = "cps_tables"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
_NOW = sa.text("CURRENT_TIMESTAMP")
def upgrade() -> None:
op.create_table(
"cps_link",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("code", sa.String(length=16), nullable=False),
sa.Column("group_id", sa.Integer(), nullable=False),
sa.Column("activity_id", sa.Integer(), nullable=False),
sa.Column("sid", sa.String(length=64), nullable=False),
sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan"),
sa.Column("target_url", sa.String(length=1024), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("code", name="uq_cps_link_code"),
)
op.create_index("ix_cps_link_code", "cps_link", ["code"])
op.create_index("ix_cps_link_group_id", "cps_link", ["group_id"])
op.create_index("ix_cps_link_activity_id", "cps_link", ["activity_id"])
op.create_index("ix_cps_link_sid", "cps_link", ["sid"])
op.create_table(
"cps_click",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("link_id", sa.Integer(), nullable=False),
sa.Column("group_id", sa.Integer(), nullable=False),
sa.Column("sid", sa.String(length=64), nullable=False),
sa.Column("ip", sa.String(length=64), nullable=True),
sa.Column("ua", sa.String(length=512), nullable=True),
sa.Column("clicked_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_cps_click_link_id", "cps_click", ["link_id"])
op.create_index("ix_cps_click_group_id", "cps_click", ["group_id"])
op.create_index("ix_cps_click_sid", "cps_click", ["sid"])
op.create_index("ix_cps_click_clicked_at", "cps_click", ["clicked_at"])
def downgrade() -> None:
op.drop_table("cps_click")
op.drop_table("cps_link")
+92
View File
@@ -0,0 +1,92 @@
"""cps_group / cps_activity / cps_order tables (CPS 分发与对账)
Revision ID: cps_tables
Revises: b3f1a2c4d5e6
Create Date: 2026-06-16 12:40:00.000000
群发 CPS 优惠券的分发与对账:cps_group(群=sid) + cps_activity(可推广活动池) +
cps_order(美团 query_order 按 sid 拉回的订单明细)。金额统一存「分」。
order_id 全局唯一(reconcile 按它 upsert,订单状态会变则更新)。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = "cps_tables"
down_revision: Union[str, Sequence[str], None] = "b3f1a2c4d5e6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
_NOW = sa.text("CURRENT_TIMESTAMP")
def upgrade() -> None:
op.create_table(
"cps_group",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("sid", sa.String(length=64), nullable=False),
sa.Column("name", sa.String(length=128), nullable=False),
sa.Column("member_count", sa.Integer(), nullable=True),
sa.Column("status", sa.String(length=20), nullable=False, server_default="active"),
sa.Column("remark", sa.String(length=256), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("sid", name="uq_cps_group_sid"),
)
op.create_index("ix_cps_group_sid", "cps_group", ["sid"])
op.create_table(
"cps_activity",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan"),
sa.Column("name", sa.String(length=128), nullable=False),
sa.Column("act_id", sa.String(length=64), nullable=True),
sa.Column("product_view_sign", sa.String(length=128), nullable=True),
sa.Column("status", sa.String(length=20), nullable=False, server_default="active"),
sa.Column("remark", sa.String(length=256), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.PrimaryKeyConstraint("id"),
)
op.create_table(
"cps_order",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("order_id", sa.String(length=64), nullable=False),
sa.Column("sid", sa.String(length=64), nullable=True),
sa.Column("act_id", sa.String(length=64), nullable=True),
sa.Column("biz_line", sa.Integer(), nullable=True),
sa.Column("trade_type", sa.Integer(), nullable=True),
sa.Column("pay_price_cents", sa.Integer(), nullable=True),
sa.Column("commission_cents", sa.Integer(), nullable=True),
sa.Column("commission_rate", sa.String(length=16), nullable=True),
sa.Column("refund_price_cents", sa.Integer(), nullable=True),
sa.Column("refund_profit_cents", sa.Integer(), nullable=True),
sa.Column("mt_status", sa.String(length=8), nullable=True),
sa.Column("invalid_reason", sa.String(length=128), nullable=True),
sa.Column("product_name", sa.String(length=512), nullable=True),
sa.Column("pay_time", sa.DateTime(timezone=True), nullable=True),
sa.Column("mt_update_time", sa.DateTime(timezone=True), nullable=True),
sa.Column(
"raw",
sa.JSON().with_variant(postgresql.JSONB(), "postgresql"),
nullable=False,
),
sa.Column("first_seen", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("order_id", name="uq_cps_order_order_id"),
)
op.create_index("ix_cps_order_order_id", "cps_order", ["order_id"])
op.create_index("ix_cps_order_sid", "cps_order", ["sid"])
op.create_index("ix_cps_order_act_id", "cps_order", ["act_id"])
op.create_index("ix_cps_order_mt_status", "cps_order", ["mt_status"])
op.create_index("ix_cps_order_pay_time", "cps_order", ["pay_time"])
def downgrade() -> None:
op.drop_table("cps_order")
op.drop_table("cps_activity")
op.drop_table("cps_group")
+2
View File
@@ -19,6 +19,7 @@ from app.admin.routers.admins import router as admins_router
from app.admin.routers.audit import router as audit_router
from app.admin.routers.auth import router as auth_router
from app.admin.routers.config import router as config_router
from app.admin.routers.cps import router as cps_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
from app.admin.routers.feedback import router as feedback_router
@@ -90,5 +91,6 @@ admin_app.include_router(feedback_router)
admin_app.include_router(admins_router)
admin_app.include_router(audit_router)
admin_app.include_router(config_router)
admin_app.include_router(cps_router)
admin_app.include_router(ad_audit_router)
admin_app.include_router(ad_revenue_router)
+287
View File
@@ -0,0 +1,287 @@
"""admin CPS 数据访问:群/活动 CRUD + 美团订单拉单入库(对账) + 按群统计聚合。
美团 query_order 返回的金额是「元」字符串、时间是秒级时间戳,这里统一转「分」+ tz-aware。
统计在 Python 侧聚合(订单量级小、admin 低频),逻辑清晰、跨 PG/SQLite 无 SQL 方言坑。
"""
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from uuid import uuid4
from sqlalchemy import desc, select
from sqlalchemy.orm import Session
from app.admin.repositories.queries import _as_utc, offset_paginate
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_order import CpsOrder
# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
_INVALID_STATUS = {"4", "5"}
_SETTLED_STATUS = "6"
# ───────────── 单位换算 ─────────────
def _yuan_to_cents(v: object) -> int | None:
"""「元」字符串/数 → 分。None / 空 / 字面 "null" → None。"""
if v is None:
return None
s = str(v).strip()
if not s or s.lower() == "null":
return None
try:
return int((Decimal(s) * 100).to_integral_value())
except (InvalidOperation, ValueError):
return None
def _ts_to_dt(ts: object) -> datetime | None:
"""秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
if not ts:
return None
try:
return datetime.fromtimestamp(int(ts), tz=timezone.utc)
except (ValueError, OSError, TypeError):
return None
# ───────────── 群 ─────────────
def get_group(db: Session, group_id: int) -> CpsGroup | None:
return db.get(CpsGroup, group_id)
def get_group_by_sid(db: Session, sid: str) -> CpsGroup | None:
return db.execute(select(CpsGroup).where(CpsGroup.sid == sid)).scalar_one_or_none()
def list_groups(
db: Session, *, keyword: str | None = None, status: str | None = None,
limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsGroup], int | None, int]:
stmt = select(CpsGroup)
if keyword and keyword.strip():
kw = f"%{keyword.strip()}%"
stmt = stmt.where(CpsGroup.name.ilike(kw) | CpsGroup.sid.ilike(kw))
if status:
stmt = stmt.where(CpsGroup.status == status)
return offset_paginate(db, stmt, (desc(CpsGroup.id),), limit=limit, cursor=cursor)
def create_group(
db: Session, *, name: str, sid: str | None = None,
member_count: int | None = None, remark: str | None = None, commit: bool = True,
) -> CpsGroup:
"""建群。sid 留空 → 自动 g<id>(先 flush 拿 id 再回填)。"""
group = CpsGroup(
name=name,
sid=sid or f"tmp{uuid4().hex[:20]}", # 临时唯一占位,留空时下面回填 g<id>
member_count=member_count,
remark=remark,
)
db.add(group)
db.flush()
if not sid:
group.sid = f"g{group.id}"
db.flush()
if commit:
db.commit()
db.refresh(group)
return group
def update_group(
db: Session, group: CpsGroup, *, name: str | None = None,
member_count: int | None = None, status: str | None = None,
remark: str | None = None, commit: bool = True,
) -> CpsGroup:
if name is not None:
group.name = name
if member_count is not None:
group.member_count = member_count
if status is not None:
group.status = status
if remark is not None:
group.remark = remark
if commit:
db.commit()
db.refresh(group)
else:
db.flush()
return group
# ───────────── 活动 ─────────────
def get_activity(db: Session, activity_id: int) -> CpsActivity | None:
return db.get(CpsActivity, activity_id)
def list_activities(
db: Session, *, platform: str | None = None, status: str | None = None,
limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsActivity], int | None, int]:
stmt = select(CpsActivity)
if platform:
stmt = stmt.where(CpsActivity.platform == platform)
if status:
stmt = stmt.where(CpsActivity.status == status)
return offset_paginate(db, stmt, (desc(CpsActivity.id),), limit=limit, cursor=cursor)
def create_activity(
db: Session, *, name: str, platform: str = "meituan", act_id: str | None = None,
product_view_sign: str | None = None, remark: str | None = None, commit: bool = True,
) -> CpsActivity:
activity = CpsActivity(
name=name, platform=platform, act_id=act_id,
product_view_sign=product_view_sign, remark=remark,
)
db.add(activity)
if commit:
db.commit()
db.refresh(activity)
else:
db.flush()
return activity
# ───────────── 订单(对账) ─────────────
def _map_order_fields(r: dict) -> dict:
"""美团 query_order 单条 dataList → CpsOrder 字段(金额转分、时间转 datetime)。"""
pn = r.get("productName")
if pn and len(pn) > 500:
pn = pn[:500]
return {
"sid": (r.get("sid") or None),
"act_id": str(r["actId"]) if r.get("actId") is not None else None,
"biz_line": r.get("businessLine"),
"trade_type": r.get("tradeType"),
"pay_price_cents": _yuan_to_cents(r.get("payPrice")),
"commission_cents": _yuan_to_cents(r.get("profit")),
"commission_rate": str(r["commissionRate"]) if r.get("commissionRate") is not None else None,
"refund_price_cents": _yuan_to_cents(r.get("refundPrice")),
"refund_profit_cents": _yuan_to_cents(r.get("refundProfit")),
"mt_status": str(r["status"]) if r.get("status") is not None else None,
"invalid_reason": (r.get("invalidReason") or None),
"product_name": pn or None,
"pay_time": _ts_to_dt(r.get("payTime")),
"mt_update_time": _ts_to_dt(r.get("updateTime")),
"raw": r,
}
def reconcile_orders(
db: Session, *, start_time: int, end_time: int,
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
) -> dict:
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
"""
fetched = inserted = updated = pages = 0
page = 1
while page <= max_pages:
resp = meituan.query_order(
sid=sid, start_time=start_time, end_time=end_time,
query_time_type=query_time_type, page=page, limit=100,
)
rows = ((resp.get("data") or {}).get("dataList")) or []
if not rows:
break
pages += 1
for r in rows:
order_id = str(r.get("orderId") or "").strip()
if not order_id:
continue
fetched += 1
fields = _map_order_fields(r)
existing = db.execute(
select(CpsOrder).where(CpsOrder.order_id == order_id)
).scalar_one_or_none()
if existing is None:
db.add(CpsOrder(order_id=order_id, **fields))
inserted += 1
else:
for k, v in fields.items():
setattr(existing, k, v)
updated += 1
if len(rows) < 100:
break
page += 1
db.commit()
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
def list_orders(
db: Session, *, sid: str | None = None, mt_status: str | None = None,
limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsOrder], int | None, int]:
stmt = select(CpsOrder)
if sid:
stmt = stmt.where(CpsOrder.sid == sid)
if mt_status:
stmt = stmt.where(CpsOrder.mt_status == mt_status)
return offset_paginate(db, stmt, (desc(CpsOrder.id),), limit=limit, cursor=cursor)
# ───────────── 统计(按群聚合) ─────────────
def group_stats(
db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
) -> list[dict]:
"""按 sid 聚合订单 + join 群信息。已建但本期无单的活跃群也列出(全 0)。
未归群的 sid(历史/其它来源)单独成行 group_id=None。按预估佣金降序。
"""
stmt = select(CpsOrder)
if date_from is not None:
stmt = stmt.where(CpsOrder.pay_time >= _as_utc(date_from))
if date_to is not None:
stmt = stmt.where(CpsOrder.pay_time <= _as_utc(date_to))
orders = list(db.execute(stmt).scalars().all())
groups = {g.sid: g for g in db.execute(select(CpsGroup)).scalars().all()}
buckets: dict[str | None, list[CpsOrder]] = {}
for o in orders:
buckets.setdefault(o.sid, []).append(o)
rows: list[dict] = []
for sid, items in buckets.items():
g = groups.get(sid) if sid else None
valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
rows.append({
"group_id": g.id if g else None,
"sid": sid,
"name": g.name if g else (sid or "(无 sid 归属)"),
"member_count": g.member_count if g else None,
"order_count": len(valid),
"settled_count": len(settled),
"canceled_count": len(canceled),
"gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
"est_commission_cents": sum(o.commission_cents or 0 for o in valid),
"settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
})
# 已建的活跃群但本期无单 → 补 0 行,让运营看到全部群
seen = set(buckets.keys())
for sid, g in groups.items():
if sid not in seen and g.status == "active":
rows.append({
"group_id": g.id, "sid": sid, "name": g.name,
"member_count": g.member_count, "order_count": 0,
"settled_count": 0, "canceled_count": 0, "gmv_cents": 0,
"est_commission_cents": 0, "settled_commission_cents": 0,
})
# 合并点击数据(按 group_id):点击只对生成过我们短链的群有,未归群行(group_id=None)恒 0
clicks = cps_link_repo.click_stats_by_group(db, date_from=date_from, date_to=date_to)
for r in rows:
c = clicks.get(r["group_id"]) if r["group_id"] is not None else None
r["click_pv"] = c["pv"] if c else 0
r["click_uv"] = c["uv"] if c else 0
rows.sort(key=lambda x: x["est_commission_cents"], reverse=True)
return rows
+268
View File
@@ -0,0 +1,268 @@
"""admin CPS 分发与对账:群(sid)管理 + 活动池 + 生成带 sid 券链接 + 美团订单对账 + 统计。
当前仅接美团(get_referral_link / query_order)。淘宝/京东待各自联盟凭证。
群/活动管理 = operator;订单对账(涉佣金) = finance;只读列表/统计 = 登录即可。
"""
from __future__ import annotations
import time
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import cps as cps_repo
from app.admin.schemas.common import CursorPage
from app.core.config import settings
from app.repositories import cps_link as cps_link_repo
from app.admin.schemas.cps import (
CpsActivityCreate,
CpsActivityOut,
CpsGroupCreate,
CpsGroupOut,
CpsGroupStat,
CpsGroupUpdate,
CpsOrderOut,
CpsReconcileResult,
CpsReferralLinkOut,
CpsReferralLinkRequest,
CpsStatsOut,
)
from app.integrations import meituan
from app.integrations.meituan import MeituanCpsError
from app.models.admin import AdminUser
router = APIRouter(
prefix="/admin/api/cps",
tags=["admin-cps"],
dependencies=[Depends(get_current_admin)],
)
# ───────────── 群(sid) ─────────────
@router.get("/groups", response_model=CursorPage[CpsGroupOut], summary="群(sid)列表")
def list_groups(
db: AdminDb,
keyword: Annotated[str | None, Query(max_length=100)] = None,
status: Annotated[str | None, Query(pattern="^(active|archived)$")] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CpsGroupOut]:
items, next_cursor, total = cps_repo.list_groups(
db, keyword=keyword, status=status, limit=limit, cursor=cursor
)
return CursorPage(
items=[CpsGroupOut.model_validate(g) for g in items],
next_cursor=next_cursor,
total=total,
)
@router.post("/groups", response_model=CpsGroupOut, summary="新建群(分配 sid)")
def create_group(
body: CpsGroupCreate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> CpsGroupOut:
if body.sid and cps_repo.get_group_by_sid(db, body.sid) is not None:
raise HTTPException(status_code=409, detail="sid 已存在")
group = cps_repo.create_group(
db, name=body.name, sid=body.sid, member_count=body.member_count,
remark=body.remark, commit=False,
)
write_audit(
db, admin, action="cps.group.create", target_type="cps_group", target_id=group.id,
detail={"sid": group.sid, "name": group.name}, ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(group)
return CpsGroupOut.model_validate(group)
@router.patch("/groups/{group_id}", response_model=CpsGroupOut, summary="编辑群")
def update_group(
group_id: int,
body: CpsGroupUpdate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> CpsGroupOut:
group = cps_repo.get_group(db, group_id)
if group is None:
raise HTTPException(status_code=404, detail="群不存在")
cps_repo.update_group(
db, group, name=body.name, member_count=body.member_count,
status=body.status, remark=body.remark, commit=False,
)
write_audit(
db, admin, action="cps.group.update", target_type="cps_group", target_id=group_id,
detail=body.model_dump(exclude_none=True), ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(group)
return CpsGroupOut.model_validate(group)
# ───────────── 活动 ─────────────
@router.get("/activities", response_model=CursorPage[CpsActivityOut], summary="活动列表")
def list_activities(
db: AdminDb,
platform: Annotated[str | None, Query(max_length=20)] = None,
status: Annotated[str | None, Query(pattern="^(active|archived)$")] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CpsActivityOut]:
items, next_cursor, total = cps_repo.list_activities(
db, platform=platform, status=status, limit=limit, cursor=cursor
)
return CursorPage(
items=[CpsActivityOut.model_validate(a) for a in items],
next_cursor=next_cursor,
total=total,
)
@router.post("/activities", response_model=CpsActivityOut, summary="新建活动")
def create_activity(
body: CpsActivityCreate,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> CpsActivityOut:
if not body.act_id and not body.product_view_sign:
raise HTTPException(status_code=400, detail="act_id 或 product_view_sign 必填其一")
activity = cps_repo.create_activity(
db, name=body.name, platform=body.platform, act_id=body.act_id,
product_view_sign=body.product_view_sign, remark=body.remark, commit=False,
)
write_audit(
db, admin, action="cps.activity.create", target_type="cps_activity", target_id=activity.id,
detail={"name": activity.name, "act_id": activity.act_id, "platform": activity.platform},
ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(activity)
return CpsActivityOut.model_validate(activity)
# ───────────── 转链 ─────────────
@router.post("/referral-link", response_model=CpsReferralLinkOut, summary="生成群发短链(带 sid + 点击统计)")
def generate_referral_link(
body: CpsReferralLinkRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> CpsReferralLinkOut:
group = cps_repo.get_group(db, body.group_id)
if group is None:
raise HTTPException(status_code=404, detail="群不存在")
activity = cps_repo.get_activity(db, body.activity_id)
if activity is None:
raise HTTPException(status_code=404, detail="活动不存在")
if activity.platform != "meituan":
raise HTTPException(status_code=400, detail=f"当前仅支持美团转链(活动平台={activity.platform})")
try:
# 全要:短链作跳转目标(微信可打开),长链/deeplink 作参考/备用
resp = meituan.get_referral_link(
act_id=activity.act_id,
product_view_sign=activity.product_view_sign,
sid=group.sid,
link_type_list=[1, 2, 3],
)
except MeituanCpsError as e:
raise HTTPException(status_code=502, detail=f"美团转链失败: {e}") from e
link_map = {str(k): v for k, v in (resp.get("referralLinkMap") or {}).items()}
if not link_map and resp.get("data"):
link_map = {"1": resp["data"]}
# 跳转目标优先短链(2,微信可打开)→ 长链(1)→ deeplink(3)
target_url = link_map.get("2") or link_map.get("1") or link_map.get("3")
if not target_url:
raise HTTPException(status_code=502, detail="美团未返回有效链接(检查 actId 是否在推广有效期)")
# 存我们的短链:发群用 /c/{code},用户点 → 我们记点击 → 302 跳 target_url
link = cps_link_repo.create_link(
db, group_id=group.id, activity_id=activity.id, sid=group.sid,
target_url=target_url, platform=activity.platform, commit=False,
)
write_audit(
db, admin, action="cps.referral_link.generate", target_type="cps_link", target_id=link.id,
detail={"sid": group.sid, "activity_id": activity.id, "code": link.code},
ip=get_client_ip(request), commit=False,
)
db.commit()
db.refresh(link)
base = settings.CPS_REDIRECT_BASE.rstrip("/")
redirect_url = f"{base}/c/{link.code}" if base else f"/c/{link.code}"
return CpsReferralLinkOut(
redirect_url=redirect_url,
code=link.code,
sid=group.sid,
group_name=group.name,
activity_name=activity.name,
target_url=target_url,
link_map=link_map,
)
# ───────────── 订单对账 ─────────────
@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
def reconcile_orders(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
days: Annotated[int, Query(ge=1, le=90)] = 7,
sid: Annotated[str | None, Query(max_length=64)] = None,
) -> CpsReconcileResult:
now = int(time.time())
try:
result = cps_repo.reconcile_orders(
db, start_time=now - days * 86400, end_time=now, sid=sid,
)
except MeituanCpsError as e:
raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
write_audit(
db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
detail={"days": days, "sid": sid, **result}, ip=get_client_ip(request), commit=True,
)
return CpsReconcileResult(**result)
@router.get("/orders", response_model=CursorPage[CpsOrderOut], summary="订单明细")
def list_orders(
db: AdminDb,
sid: Annotated[str | None, Query(max_length=64)] = None,
mt_status: Annotated[str | None, Query(max_length=8)] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CpsOrderOut]:
items, next_cursor, total = cps_repo.list_orders(
db, sid=sid, mt_status=mt_status, limit=limit, cursor=cursor
)
return CursorPage(
items=[CpsOrderOut.model_validate(o) for o in items],
next_cursor=next_cursor,
total=total,
)
# ───────────── 统计 ─────────────
@router.get("/stats", response_model=CpsStatsOut, summary="按群对账统计")
def get_stats(
db: AdminDb,
days: Annotated[int, Query(ge=1, le=90)] = 30,
) -> CpsStatsOut:
date_to = datetime.now(timezone.utc)
date_from = date_to - timedelta(days=days)
rows = cps_repo.group_stats(db, date_from=date_from, date_to=date_to)
stats = [CpsGroupStat(**r) for r in rows]
return CpsStatsOut(
groups=stats,
total_order_count=sum(s.order_count for s in stats),
total_est_commission_cents=sum(s.est_commission_cents for s in stats),
total_settled_commission_cents=sum(s.settled_commission_cents for s in stats),
)
+122
View File
@@ -0,0 +1,122 @@
"""admin CPS 分发与对账 schemas。金额统一「分」(cents),前端 yuan() 展示。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# ───────────── 群(sid) ─────────────
class CpsGroupOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
sid: str
name: str
member_count: int | None = None
status: str
remark: str | None = None
created_at: datetime
class CpsGroupCreate(BaseModel):
name: str = Field(min_length=1, max_length=128)
# 留空 → 后端自动生成 g<id>;填写则必须字母+数字(美团 sid 限制)。
sid: str | None = Field(default=None, max_length=64, pattern=r"^[A-Za-z0-9]+$")
member_count: int | None = Field(default=None, ge=0)
remark: str | None = Field(default=None, max_length=256)
class CpsGroupUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=128)
member_count: int | None = Field(default=None, ge=0)
status: str | None = Field(default=None, pattern=r"^(active|archived)$")
remark: str | None = Field(default=None, max_length=256)
# ───────────── 活动(actId) ─────────────
class CpsActivityOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
platform: str
name: str
act_id: str | None = None
product_view_sign: str | None = None
status: str
remark: str | None = None
created_at: datetime
class CpsActivityCreate(BaseModel):
name: str = Field(min_length=1, max_length=128)
platform: str = Field(default="meituan", pattern=r"^(meituan|taobao|jd)$")
act_id: str | None = Field(default=None, max_length=64)
product_view_sign: str | None = Field(default=None, max_length=128)
remark: str | None = Field(default=None, max_length=256)
# ───────────── 转链 ─────────────
class CpsReferralLinkRequest(BaseModel):
group_id: int
activity_id: int
# 链接类型:1长链 2短链 3deeplink。默认短链(微信可打开)+deeplink。
link_types: list[int] = Field(default_factory=lambda: [2, 3])
class CpsReferralLinkOut(BaseModel):
redirect_url: str # ★ 我们的群发短链(/c/{code}),发群就用这个,点击经我们统计
code: str
sid: str
group_name: str
activity_name: str
target_url: str # 实际 302 跳转的美团短链
link_map: dict[str, str] # 美团原始 1长链/2短链/3deeplink(参考/备用)
# ───────────── 对账拉单 ─────────────
class CpsReconcileResult(BaseModel):
fetched: int # 本次美团返回的订单条数
inserted: int # 新入库
updated: int # 已存在→更新状态/佣金
pages: int # 翻了几页
# ───────────── 订单明细 ─────────────
class CpsOrderOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
order_id: str
sid: str | None = None
act_id: str | None = None
pay_price_cents: int | None = None
commission_cents: int | None = None
commission_rate: str | None = None
mt_status: str | None = None
invalid_reason: str | None = None
product_name: str | None = None
pay_time: datetime | None = None
# ───────────── 统计 ─────────────
class CpsGroupStat(BaseModel):
group_id: int | None = None # None = 未归群的 sid(历史/其它来源)
sid: str | None = None
name: str
member_count: int | None = None
click_pv: int = 0 # 点击总次数
click_uv: int = 0 # 独立点击(按 ip+ua 近似去重)
order_count: int # 有效订单数(不含取消/风控)
settled_count: int # 已结算单数
canceled_count: int # 取消+风控单数
gmv_cents: int # 成交额(有效单 payPrice 合计)
est_commission_cents: int # 预估佣金(有效单 profit 合计)
settled_commission_cents: int # 已结算佣金
class CpsStatsOut(BaseModel):
groups: list[CpsGroupStat]
total_order_count: int
total_est_commission_cents: int
total_settled_commission_cents: int
+39
View File
@@ -0,0 +1,39 @@
"""CPS 群发短链跳转:用户点 /c/{code} → 记一条点击 → 302 跳美团。
公网、无鉴权(群里任何人点都要能跳)。记点击失败绝不影响跳转(用户体验优先)。
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, Request
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.repositories import cps_link as cps_link_repo
router = APIRouter(tags=["cps-redirect"])
# code 不存在/失效时的兜底落地(避免用户看到报错)
_FALLBACK_URL = "https://www.meituan.com/"
def _client_ip(request: Request) -> str | None:
xff = request.headers.get("x-forwarded-for")
if xff:
return xff.split(",")[0].strip()
return request.client.host if request.client else None
@router.get("/c/{code}", summary="群发短链跳转(记点击 + 302)")
def cps_redirect(code: str, request: Request, db: Session = Depends(get_db)):
link = cps_link_repo.get_by_code(db, code)
if link is None:
return RedirectResponse(_FALLBACK_URL, status_code=302)
try:
cps_link_repo.record_click(
db, link=link, ip=_client_ip(request),
ua=request.headers.get("user-agent"),
)
except Exception:
pass # 记点击失败不阻断跳转
return RedirectResponse(link.target_url, status_code=302)
+7
View File
@@ -190,6 +190,13 @@ class Settings(BaseSettings):
# MVP 落地页就放 app-server 的 /media 静态目录下(dl.html)。
INVITE_LANDING_URL: str = "https://app-api.shaguabijia.com/media/dl.html"
# ===== CPS 群发短链跳转 =====
# 群发券链接套一层我们的短链 /c/{code} 做点击统计后 302 跳美团。这里是写进 admin
# 生成链接里的对外跳转域名前缀。本地填 http://localhost:8770;生产**强烈建议用独立
# 域名**(大量群发会被微信封,别用主 API 域名连累整个 App)。留空 → 跳转端点用请求
# 的 Host 兜底拼绝对地址。
CPS_REDIRECT_BASE: str = ""
# ===== CORS =====
CORS_ALLOW_ORIGINS: str = ""
+39 -2
View File
@@ -126,13 +126,23 @@ def query_coupon(
def get_referral_link(
*,
product_view_sign: str,
product_view_sign: str | None = None,
act_id: str | None = None,
platform: int = 1,
biz_line: int | None = None,
sid: str | None = None,
link_type_list: list[int] | None = None,
) -> dict[str, Any]:
body: dict[str, Any] = {"productViewSign": product_view_sign}
"""换推广链接。actId(活动物料)与 productViewSign(商品券)二选一。
用 actId 入参时,出参 data 为 null,链接全在 referralLinkMap(实测)。
"""
if not act_id and not product_view_sign:
raise MeituanCpsError("act_id 或 product_view_sign 必填其一")
body: dict[str, Any] = {}
if act_id:
body["actId"] = act_id
else:
body["productViewSign"] = product_view_sign
if platform == 2:
body["platform"] = 2
if biz_line is not None:
@@ -142,3 +152,30 @@ def get_referral_link(
body["linkTypeList"] = link_type_list or [1, 3]
return _call("/cps_open/common/api/v1/get_referral_link", body)
def query_order(
*,
sid: str | None = None,
start_time: int,
end_time: int,
query_time_type: int = 1,
page: int = 1,
limit: int = 100,
) -> dict[str, Any]:
"""按时间窗(+可选 sid)拉 CPS 订单做对账。实测要点:
- 分页是 page / limit(不是 pageNo/pageSize);startTime/endTime 是 10 位秒级时间戳
- query_time_type: 1 按支付时间 2 按更新时间
- 响应 data.dataList[],每条含 orderId / sid / payPrice(元) / profit(元) /
commissionRate / status(2付款 3完成 4取消 5风控 6结算) / payTime(秒) / actId 等
"""
body: dict[str, Any] = {
"queryTimeType": query_time_type,
"startTime": start_time,
"endTime": end_time,
"page": page,
"limit": limit,
}
if sid:
body["sid"] = sid
return _call("/cps_open/common/api/v1/query_order", body)
+3
View File
@@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router
from app.api.v1.compare_milestone import router as compare_milestone_router
from app.api.v1.compare_record import router as compare_record_router
from app.api.v1.coupon import router as coupon_router
from app.api.v1.cps_redirect import router as cps_redirect_router
from app.api.internal.price import router as internal_price_router
from app.api.internal.store import router as internal_store_router
from app.api.v1.feedback import router as feedback_router
@@ -107,6 +108,8 @@ app.include_router(report_router)
app.include_router(internal_price_router)
app.include_router(internal_store_router)
app.include_router(platform_router)
# CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团)
app.include_router(cps_redirect_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+4
View File
@@ -6,6 +6,10 @@ from app.models.ad_watch_log import AdWatchLog # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.app_config import AppConfig # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.cps_activity import CpsActivity # noqa: F401
from app.models.cps_group import CpsGroup # noqa: F401
from app.models.cps_link import CpsClick, CpsLink # noqa: F401
from app.models.cps_order import CpsOrder # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.coupon_state import ( # noqa: F401
CouponClaimRecord,
+34
View File
@@ -0,0 +1,34 @@
"""CPS 可推广活动池(cps_activity)。
运营预存常推的活动(),生成链接时选它,省得每次记 actId当前只接美团:
`act_id` = 美团联盟我要推广-活动推广第一列的物料 ID`platform` 预留,
淘宝/京东时复用本表(各自的活动标识)
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class CpsActivity(Base):
__tablename__ = "cps_activity"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
platform: Mapped[str] = mapped_column(String(20), nullable=False, default="meituan") # meituan|taobao|jd
name: Mapped[str] = mapped_column(String(128), nullable=False)
# 美团活动物料 ID(actId);淘宝/京东接入后存各自活动标识。
act_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 美团商品券 productViewSign(与 act_id 二选一转链;多数活动用 act_id)。
product_view_sign: Mapped[str | None] = mapped_column(String(128), nullable=True)
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active") # active | archived
remark: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<CpsActivity id={self.id} platform={self.platform} name={self.name!r}>"
+32
View File
@@ -0,0 +1,32 @@
"""CPS 推广群(cps_group)。
每个微信群一条记录,`sid` 是美团联盟二级渠道追踪位(get_referral_link / query_order
共用):发券时塞进推广链接,订单回来按 sid 归群对账美团限制 sid 仅字母+数字 64
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class CpsGroup(Base):
__tablename__ = "cps_group"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# 每群唯一的渠道标识,建群时分配(留空则自动生成 g<id>)。字母+数字,≤64。
sid: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
name: Mapped[str] = mapped_column(String(128), nullable=False)
# 群人数,运营填,作转化率分母(可空)。
member_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active") # active | archived
remark: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<CpsGroup id={self.id} sid={self.sid!r} name={self.name!r}>"
+52
View File
@@ -0,0 +1,52 @@
"""CPS 群发短链(cps_link)+ 点击事件(cps_click)。
群发券链接套一层我们自己的短链 `/c/{code}`:用户点 记一条 cps_click 302
target_url(美团短链,微信可打开)据此统计点击 PV/UV(按群/活动/时段),配合 cps_order
的下单/佣金做"点击→下单→佣金"漏斗group_id/sid cps_click 里冗余一份,统计直接按
群聚合点击不必 join cps_link
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class CpsLink(Base):
__tablename__ = "cps_link"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# 短码:发群链接 /c/{code} 的 code,全局唯一。
code: Mapped[str] = mapped_column(String(16), unique=True, index=True, nullable=False)
group_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
activity_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
sid: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
platform: Mapped[str] = mapped_column(String(20), nullable=False, default="meituan")
# 302 跳转目标(美团短链,微信内可打开并唤起 App)。
target_url: Mapped[str] = mapped_column(String(1024), nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<CpsLink id={self.id} code={self.code!r} sid={self.sid!r}>"
class CpsClick(Base):
__tablename__ = "cps_click"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
link_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
group_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False) # 冗余,按群聚合快
sid: Mapped[str] = mapped_column(String(64), index=True, nullable=False) # 冗余
ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
ua: Mapped[str | None] = mapped_column(String(512), nullable=True)
clicked_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<CpsClick id={self.id} link_id={self.link_id} group_id={self.group_id}>"
+63
View File
@@ -0,0 +1,63 @@
"""CPS 对账订单(cps_order)。
从美团联盟 query_order 按时间窗拉回 sid 归群的订单明细字段对齐 query_order
实测返回:
- payPrice / profit 字符串 入库统一转(与全站口径一致)
- payTime / updateTime 是秒级时间戳 入库转 tz-aware datetime
- status: 2付款 3完成 4取消 5风控 6结算(取消/风控不计佣金)
order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class CpsOrder(Base):
__tablename__ = "cps_order"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# 美团订单号(加密串),全局唯一,upsert 幂等键。
order_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
# 渠道追踪位 = 群 sid(历史无 sid 订单为空)。按它归群聚合。
sid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
act_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
biz_line: Mapped[int | None] = mapped_column(Integer, nullable=True) # 1=外卖
trade_type: Mapped[int | None] = mapped_column(Integer, nullable=True) # 1=cps 2=cpa
pay_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True) # 预估佣金(profit)
commission_rate: Mapped[str | None] = mapped_column(String(16), nullable=True) # "300"=3% "10"=0.1%
refund_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
refund_profit_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 美团订单状态: 2付款 3完成 4取消 5风控 6结算
mt_status: Mapped[str | None] = mapped_column(String(8), index=True, nullable=True)
invalid_reason: Mapped[str | None] = mapped_column(String(128), nullable=True)
product_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
pay_time: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), index=True, nullable=True
)
mt_update_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
raw: Mapped[dict] = mapped_column(
JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=dict
)
first_seen: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<CpsOrder id={self.id} order_id={self.order_id!r} "
f"sid={self.sid!r} status={self.mt_status} profit_cents={self.commission_cents}>"
)
+82
View File
@@ -0,0 +1,82 @@
"""CPS 群发短链 + 点击 数据访问(用户侧跳转端点 + admin 生成/统计 共用)。
短码生成去掉易混字符;点击按 (ip, ua) 近似去重 UV点击量级小admin 低频,统计在
Python 侧聚合, PG/SQLite 无方言坑( admin cps repo 同风格)
"""
from __future__ import annotations
import secrets
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.cps_link import CpsClick, CpsLink
# 去掉易混字符 0/O/1/l/I,运营/用户肉眼复制不易错
_CODE_ALPHABET = "abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
def _gen_code(n: int = 8) -> str:
return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(n))
def get_by_code(db: Session, code: str) -> CpsLink | None:
return db.execute(select(CpsLink).where(CpsLink.code == code)).scalar_one_or_none()
def create_link(
db: Session, *, group_id: int, activity_id: int, sid: str, target_url: str,
platform: str = "meituan", commit: bool = True,
) -> CpsLink:
"""生成唯一短码并存库。冲突重试 5 次,仍冲突用 12 位兜底(概率近 0)。"""
code = _gen_code()
for _ in range(5):
if get_by_code(db, code) is None:
break
code = _gen_code()
else:
code = _gen_code(12)
link = CpsLink(
code=code, group_id=group_id, activity_id=activity_id, sid=sid,
target_url=target_url, platform=platform,
)
db.add(link)
if commit:
db.commit()
db.refresh(link)
else:
db.flush()
return link
def record_click(db: Session, *, link: CpsLink, ip: str | None = None, ua: str | None = None) -> None:
db.add(CpsClick(
link_id=link.id, group_id=link.group_id, sid=link.sid,
ip=ip, ua=(ua[:500] if ua else None),
))
db.commit()
def click_stats_by_group(
db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
) -> dict[int, dict]:
"""按 group_id 聚合点击 → {group_id: {"pv": N, "uv": M}}。uv 按 (ip, ua) 近似去重。"""
stmt = select(CpsClick)
if date_from is not None:
stmt = stmt.where(CpsClick.clicked_at >= date_from)
if date_to is not None:
stmt = stmt.where(CpsClick.clicked_at <= date_to)
clicks = list(db.execute(stmt).scalars().all())
agg: dict[int, dict] = {}
seen: dict[int, set] = {}
for c in clicks:
a = agg.setdefault(c.group_id, {"pv": 0, "uv": 0})
a["pv"] += 1
s = seen.setdefault(c.group_id, set())
key = (c.ip, c.ua)
if key not in s:
s.add(key)
a["uv"] += 1
return agg