marco
7fd70a264d
- 服务号网页授权(WX_MP_APPID/SECRET,区别于 App 的 WECHAT_APP_ID):
落地页 base 静默拿 openid;点「领券」补 userinfo 拿昵称头像(交互触发避快照页)
- 新表 cps_wx_user + cps_click 加 openid + 迁移;integrations/wx_oauth
- /c/{code} 授权链路 + 回调 /wx/oauth/cb + cookie 免重复授权;click 带 openid
- admin 群详情加「群内微信用户」(领券画像)端点
- 下单归因到人需 user-level sid,本期只做身份+领券侧
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""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 | None, 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,
|
|
event_type: str = "visit", openid: str | None = None,
|
|
) -> None:
|
|
"""记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。
|
|
openid: 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计。"""
|
|
db.add(CpsClick(
|
|
link_id=link.id, group_id=link.group_id, sid=link.sid, event_type=event_type,
|
|
ip=ip, ua=(ua[:500] if ua else None), openid=openid,
|
|
))
|
|
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, uv, copy_pv, copy_uv}}。
|
|
visit 计入 pv/uv,copy(淘宝复制口令) 计入 copy_pv/copy_uv;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_visit: dict[int, set] = {}
|
|
seen_copy: dict[int, set] = {}
|
|
for c in clicks:
|
|
a = agg.setdefault(c.group_id, {"pv": 0, "uv": 0, "copy_pv": 0, "copy_uv": 0})
|
|
key = (c.ip, c.ua)
|
|
if c.event_type == "copy":
|
|
a["copy_pv"] += 1
|
|
s = seen_copy.setdefault(c.group_id, set())
|
|
if key not in s:
|
|
s.add(key)
|
|
a["copy_uv"] += 1
|
|
else:
|
|
a["pv"] += 1
|
|
s = seen_visit.setdefault(c.group_id, set())
|
|
if key not in s:
|
|
s.add(key)
|
|
a["uv"] += 1
|
|
return agg
|