diff --git a/alembic/versions/cps_wx_user.py b/alembic/versions/cps_wx_user.py new file mode 100644 index 0000000..0e0db37 --- /dev/null +++ b/alembic/versions/cps_wx_user.py @@ -0,0 +1,48 @@ +"""cps 微信用户表 cps_wx_user + cps_click 加 openid + +CPS 落地页微信网页授权拿 openid/昵称头像,做用户级群统计。 + +Revision ID: cps_wx_user +Revises: comparison_debug_fields +Create Date: 2026-06-19 12:00:00.000000 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "cps_wx_user" +down_revision: Union[str, Sequence[str], None] = "comparison_debug_fields" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "cps_wx_user", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("openid", sa.String(64), nullable=False), + sa.Column("unionid", sa.String(64), nullable=True), + sa.Column("nickname", sa.String(128), nullable=True), + sa.Column("headimgurl", sa.String(512), nullable=True), + sa.Column("first_code", sa.String(16), nullable=True), + sa.Column("first_group_id", sa.Integer(), nullable=True), + sa.Column("first_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("last_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + op.create_index("ix_cps_wx_user_openid", "cps_wx_user", ["openid"], unique=True) + op.create_index("ix_cps_wx_user_unionid", "cps_wx_user", ["unionid"]) + op.create_index("ix_cps_wx_user_first_group_id", "cps_wx_user", ["first_group_id"]) + + op.add_column("cps_click", sa.Column("openid", sa.String(64), nullable=True)) + op.create_index("ix_cps_click_openid", "cps_click", ["openid"]) + + +def downgrade() -> None: + op.drop_index("ix_cps_click_openid", table_name="cps_click") + op.drop_column("cps_click", "openid") + op.drop_index("ix_cps_wx_user_first_group_id", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_unionid", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_openid", table_name="cps_wx_user") + op.drop_table("cps_wx_user") diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py index 3271a74..4b1c7c3 100644 --- a/app/admin/repositories/cps.py +++ b/app/admin/repositories/cps.py @@ -9,7 +9,7 @@ from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from uuid import uuid4 -from sqlalchemy import desc, select +from sqlalchemy import desc, func, select from sqlalchemy.orm import Session from app.admin.repositories.queries import _as_utc, offset_paginate @@ -19,6 +19,7 @@ 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 +from app.models.cps_wx_user import CpsWxUser # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账 _INVALID_STATUS = {"4", "5"} @@ -470,3 +471,47 @@ def group_order_daily( "settled_commission_cents": sum(o.commission_cents or 0 for o in settled), } return out + + +def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]: + """群内微信用户(首次来源 first_group_id=group_id)+ 每人在该群的 visit/copy 次数(领券画像)。 + + 按首次进入倒序。copy 次数 = 该用户在该群点了几次「复制口令」(领券意向),visit = 进落地页次数。 + """ + users = ( + db.execute( + select(CpsWxUser) + .where(CpsWxUser.first_group_id == group_id) + .order_by(desc(CpsWxUser.first_seen)) + .limit(limit) + ) + .scalars() + .all() + ) + if not users: + return [] + openids = [u.openid for u in users] + stat: dict[str, dict] = {} + rows = db.execute( + select(CpsClick.openid, CpsClick.event_type, func.count()) + .where(CpsClick.group_id == group_id) + .where(CpsClick.openid.in_(openids)) + .group_by(CpsClick.openid, CpsClick.event_type) + ).all() + for openid, event_type, cnt in rows: + s = stat.setdefault(openid, {"visit": 0, "copy": 0}) + if event_type == "copy": + s["copy"] = cnt + else: + s["visit"] = cnt + return [ + { + "openid": u.openid, + "nickname": u.nickname, + "headimgurl": u.headimgurl, + "first_seen": u.first_seen, + "visit_count": stat.get(u.openid, {}).get("visit", 0), + "copy_count": stat.get(u.openid, {}).get("copy", 0), + } + for u in users + ] diff --git a/app/admin/routers/cps.py b/app/admin/routers/cps.py index 6f892dd..a1fd267 100644 --- a/app/admin/routers/cps.py +++ b/app/admin/routers/cps.py @@ -488,3 +488,11 @@ def group_daily( "days": days, "rows": rows, } + + +@router.get("/groups/{group_id}/wx-users", summary="群内微信用户(领券画像:头像/昵称/领券次数)") +def group_wx_users(group_id: int, db: AdminDb) -> dict: + group = cps_repo.get_group(db, group_id) + if group is None: + raise HTTPException(status_code=404, detail="群不存在") + return {"users": cps_repo.group_wx_users(db, group_id=group_id)} diff --git a/app/api/v1/cps_redirect.py b/app/api/v1/cps_redirect.py index 49a061d..5270586 100644 --- a/app/api/v1/cps_redirect.py +++ b/app/api/v1/cps_redirect.py @@ -8,16 +8,22 @@ from __future__ import annotations import html import json +import logging from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse from sqlalchemy.orm import Session from app.core import media +from app.core.config import settings from app.db.session import get_db +from app.integrations import wx_oauth from app.models.cps_activity import CpsActivity from app.models.cps_link import CpsLink from app.repositories import cps_link as cps_link_repo +from app.repositories import cps_wx_user as cps_wx_user_repo + +logger = logging.getLogger("shagua.cps_redirect") router = APIRouter(tags=["cps-redirect"]) @@ -27,6 +33,16 @@ _FALLBACK_URL = "https://www.meituan.com/" # 淘宝活动未设图时的兜底主视觉(存量已回填,基本只在老 link/异常时触发) _DEFAULT_TAOBAO_IMAGE = "/media/taobao_landing.jpg" +# 微信网页授权拿到的用户标识 cookie(种在 coupon 域,30 天免重复授权) +_WX_OPENID_COOKIE = "wx_openid" +_WX_UINFO_COOKIE = "wx_uinfo" # "1" = 已拿过昵称头像(userinfo),点领券不再跳授权 +_COOKIE_MAX_AGE = 30 * 86400 + + +def _is_wechat(request: Request) -> bool: + """是否微信内置浏览器(网页授权只在微信内有意义,外部浏览器不跳授权)。""" + return "micromessenger" in (request.headers.get("user-agent", "").lower()) + def _client_ip(request: Request) -> str | None: xff = request.headers.get("x-forwarded-for") @@ -35,11 +51,13 @@ def _client_ip(request: Request) -> str | None: return request.client.host if request.client else None -def _record(db: Session, link: CpsLink, request: Request, event_type: str) -> None: +def _record( + db: Session, link: CpsLink, request: Request, event_type: str, openid: str | None = None +) -> None: try: cps_link_repo.record_click( db, link=link, ip=_client_ip(request), - ua=request.headers.get("user-agent"), event_type=event_type, + ua=request.headers.get("user-agent"), event_type=event_type, openid=openid, ) except Exception: pass # 记点击失败不阻断 @@ -53,38 +71,89 @@ def wx_mp_domain_verify() -> PlainTextResponse: return PlainTextResponse("F7wnRQ7xPbhVOWC8") -@router.get("/c/{code}", summary="群发短链落地(记点击 + 跳转/淘宝落地页)") +@router.get("/c/{code}", summary="群发短链落地(微信授权拿 openid + 记点击 + 跳转/淘宝落地页)") def cps_landing(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) - _record(db, link, request, "visit") + openid = request.cookies.get(_WX_OPENID_COOKIE) + # 微信内 + 还没拿到 openid + 配了服务号 → 先 base 静默授权拿 openid,再 302 回本页。 + # 非微信 / 已有 cookie / 未配服务号 → 直接走原逻辑(openid 可能为 None,绝不阻断领券)。 + if openid is None and settings.wx_mp_configured and _is_wechat(request): + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}") + return RedirectResponse(auth_url, status_code=302) + _record(db, link, request, "visit", openid=openid) if link.platform == "taobao": activity = db.get(CpsActivity, link.activity_id) image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE - return HTMLResponse(_taobao_landing_html(link.target_url, image_url)) + has_uinfo = request.cookies.get(_WX_UINFO_COOKIE) == "1" + return HTMLResponse( + _taobao_landing_html(link.target_url, image_url, code, openid, has_uinfo) + ) # 美团短链 / 京东链接:直接 302 跳 return RedirectResponse(link.target_url, status_code=302) -@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件)") +@router.get("/wx/oauth/cb", include_in_schema=False) +def wx_oauth_cb(code: str, state: str, db: Session = Depends(get_db)): + """微信网页授权回调。state='base:{原code}'(静默拿 openid) 或 'uinfo:{原code}'(补昵称头像)。 + 换 openid(+userinfo)→ upsert 用户 → 种 cookie → 302 回落地页。任何失败兜底回落地页, + 绝不阻断用户领券。""" + kind, _, orig_code = state.partition(":") + try: + token = wx_oauth.exchange_code(code) # {openid, access_token, scope, ...} + openid = token["openid"] + link = cps_link_repo.get_by_code(db, orig_code) + group_id = link.group_id if link else None + nickname = headimgurl = unionid = None + if kind == "uinfo" and "userinfo" in (token.get("scope") or ""): + info = wx_oauth.get_userinfo(token["access_token"], openid) + nickname, headimgurl, unionid = ( + info.get("nickname"), info.get("headimgurl"), info.get("unionid"), + ) + cps_wx_user_repo.upsert( + db, openid=openid, code=orig_code, group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + except Exception: + logger.exception("[wx_oauth] callback failed state=%s", state) + return RedirectResponse(f"/c/{orig_code}" if orig_code else _FALLBACK_URL, status_code=302) + # userinfo 授权回来带 ?authed=1,落地页据此自动触发复制(把"点领券→授权→复制"衔接为一步) + target = f"/c/{orig_code}" + ("?authed=1" if kind == "uinfo" else "") + resp = RedirectResponse(target, status_code=302) + resp.set_cookie(_WX_OPENID_COOKIE, openid, max_age=_COOKIE_MAX_AGE, httponly=True, samesite="lax") + if nickname: + resp.set_cookie(_WX_UINFO_COOKIE, "1", max_age=_COOKIE_MAX_AGE, samesite="lax") + return resp + + +@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件,带 openid)") def cps_copy(code: str, request: Request, db: Session = Depends(get_db)) -> dict: link = cps_link_repo.get_by_code(db, code) if link is not None: - _record(db, link, request, "copy") + _record(db, link, request, "copy", openid=request.cookies.get(_WX_OPENID_COOKIE)) return {"ok": True} -def _taobao_landing_html(token: str, image_url: str) -> str: +def _taobao_landing_html( + token: str, image_url: str, code: str, openid: str | None, has_uinfo: bool +) -> str: """淘宝落地页 H5(主视觉图按活动传入,复制淘口令按钮在 75% 屏高)。 - image_url 转绝对(落地页 coupon 域,相对也可达,绝对更稳)后经 html.escape 嵌入 - (防属性注入);token 经 json.dumps 安全嵌入 JS。 + image_url 经 html.escape 嵌入 (防注入);token 经 json.dumps 安全嵌入 JS。 + uinfo_url:已有 openid 但还没拿过昵称头像时,生成 userinfo 授权链接 —— 用户点「领券」 + 时先跳它(交互触发,避免微信快照页),授权回来自动复制。已拿过 / 无 openid 则为空,直接复制。 """ safe_img = html.escape(media.to_abs_media_url(image_url) or image_url, quote=True) + uinfo_url = "" + if openid and not has_uinfo and settings.wx_mp_configured: + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + uinfo_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_userinfo", f"uinfo:{code}") return ( _TAOBAO_HTML .replace("__IMAGE_URL__", safe_img) + .replace("__UINFO_URL__", json.dumps(uinfo_url)) .replace("__TOKEN_JS__", json.dumps(token)) ) @@ -112,6 +181,7 @@ body{font-family:-apple-system,"PingFang SC",sans-serif;background:#fff0ef;color
""" diff --git a/app/core/config.py b/app/core/config.py index bfc4139..8862476 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -95,6 +95,17 @@ class Settings(BaseSettings): """美团 CPS 凭证齐全(缺则接口返空,而非 502)。""" return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET) + # ===== 微信服务号(网页授权) ===== + # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。 + # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 + WX_MP_APPID: str = "" + WX_MP_SECRET: str = "" + + @property + def wx_mp_configured(self) -> bool: + """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" + return bool(self.WX_MP_APPID and self.WX_MP_SECRET) + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 diff --git a/app/integrations/wx_oauth.py b/app/integrations/wx_oauth.py new file mode 100644 index 0000000..0af50d6 --- /dev/null +++ b/app/integrations/wx_oauth.py @@ -0,0 +1,75 @@ +"""微信服务号网页授权:构造授权链接 + code 换 openid + 拉 userinfo。 + +用于 CPS 落地页(coupon.shaguabijia.com)在微信内拿用户身份。流程(见微信文档): + 授权链接(snsapi_base|snsapi_userinfo) → 回调带 code → sns/oauth2/access_token 换 openid + → (userinfo 时)sns/userinfo 拿昵称头像。 + +注意: + - 这里的 access_token 是【网页授权 token】,与基础 access_token(cgi-bin/token)不同, + 走 sns/* 接口,不需要 IP 白名单。 + - secret 只在服务器,不下发客户端。 + - 用 WX_MP_APPID/SECRET(已认证服务号),区别于 WECHAT_APP_ID(App 移动应用)。 +""" +from __future__ import annotations + +from urllib.parse import quote + +import httpx + +from app.core.config import settings + +_API = "https://api.weixin.qq.com" +_AUTHORIZE = "https://open.weixin.qq.com/connect/oauth2/authorize" +_TIMEOUT = 10 + + +class WxOauthError(Exception): + """网页授权调用失败(凭证/网络/code 失效等)。调用方兜底为无 openid。""" + + +def build_authorize_url(redirect_uri: str, scope: str, state: str) -> str: + """构造网页授权跳转链接。scope: 'snsapi_base'(静默,仅 openid) | 'snsapi_userinfo'(昵称头像)。 + + 微信要求 redirect_uri urlEncode、参数顺序固定、结尾 #wechat_redirect。 + """ + return ( + f"{_AUTHORIZE}?appid={settings.WX_MP_APPID}" + f"&redirect_uri={quote(redirect_uri, safe='')}" + f"&response_type=code&scope={scope}&state={quote(state, safe='')}" + f"#wechat_redirect" + ) + + +def exchange_code(code: str) -> dict: + """code 换网页授权 access_token + openid。返回含 openid/access_token/scope(/unionid)。 + + code 5 分钟内一次性有效。失败(errcode)抛 WxOauthError。 + """ + resp = httpx.get( + f"{_API}/sns/oauth2/access_token", + params={ + "appid": settings.WX_MP_APPID, + "secret": settings.WX_MP_SECRET, + "code": code, + "grant_type": "authorization_code", + }, + timeout=_TIMEOUT, + ) + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"exchange_code failed: {data}") + return data + + +def get_userinfo(access_token: str, openid: str) -> dict: + """拉用户信息(nickname/headimgurl/unionid)。需 scope=snsapi_userinfo 的 access_token。""" + resp = httpx.get( + f"{_API}/sns/userinfo", + params={"access_token": access_token, "openid": openid, "lang": "zh_CN"}, + timeout=_TIMEOUT, + ) + resp.encoding = "utf-8" # 昵称含中文/emoji,强制 utf-8 避免乱码 + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"get_userinfo failed: {data}") + return data diff --git a/app/models/__init__.py b/app/models/__init__.py index 1955e31..768d192 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -10,6 +10,7 @@ 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.cps_wx_user import CpsWxUser # noqa: F401 from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401 from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, diff --git a/app/models/cps_link.py b/app/models/cps_link.py index 2702c8a..bb88e6b 100644 --- a/app/models/cps_link.py +++ b/app/models/cps_link.py @@ -49,6 +49,8 @@ class CpsClick(Base): ) ip: Mapped[str | None] = mapped_column(String(64), nullable=True) ua: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计 + openid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) clicked_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True, nullable=False ) diff --git a/app/models/cps_wx_user.py b/app/models/cps_wx_user.py new file mode 100644 index 0000000..669b26c --- /dev/null +++ b/app/models/cps_wx_user.py @@ -0,0 +1,42 @@ +"""CPS 落地页微信用户(cps_wx_user)。 + +用户在微信内打开群发短链落地页 /c/{code},经服务号网页授权: + - base 静默:拿 openid(唯一标识,统计主力) + - userinfo(点领券触发):补 nickname/headimgurl/unionid + +按 openid 唯一,记录首次来源群(first_group_id),用于群内用户级统计(谁领了券)。 +下单归因到人需 user-level sid(另一期),本表只承载身份 + 领券/点击侧。 +""" +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 CpsWxUser(Base): + __tablename__ = "cps_wx_user" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + # 服务号下用户唯一标识(网页授权拿到) + openid: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + # 开放平台 unionid(服务号绑开放平台 + scope=userinfo 才有);跨 App/服务号统一用户 + unionid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + # 昵称/头像:userinfo 授权后才有(base 阶段为空) + nickname: Mapped[str | None] = mapped_column(String(128), nullable=True) + headimgurl: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 首次进入来源(从哪个群的链接授权进来) + first_code: Mapped[str | None] = mapped_column(String(16), nullable=True) + first_group_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + first_seen: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + last_seen: 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"" diff --git a/app/repositories/cps_link.py b/app/repositories/cps_link.py index 2e34970..00b9c1f 100644 --- a/app/repositories/cps_link.py +++ b/app/repositories/cps_link.py @@ -52,12 +52,13 @@ def create_link( def record_click( db: Session, *, link: CpsLink, ip: str | None = None, ua: str | None = None, - event_type: str = "visit", + event_type: str = "visit", openid: str | None = None, ) -> None: - """记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。""" + """记一条点击。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), + ip=ip, ua=(ua[:500] if ua else None), openid=openid, )) db.commit() diff --git a/app/repositories/cps_wx_user.py b/app/repositories/cps_wx_user.py new file mode 100644 index 0000000..9fde31e --- /dev/null +++ b/app/repositories/cps_wx_user.py @@ -0,0 +1,43 @@ +"""cps_wx_user 数据访问:按 openid upsert。""" +from __future__ import annotations + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models.cps_wx_user import CpsWxUser + + +def get_by_openid(db: Session, openid: str) -> CpsWxUser | None: + return db.execute( + select(CpsWxUser).where(CpsWxUser.openid == openid) + ).scalar_one_or_none() + + +def upsert( + db: Session, *, openid: str, code: str | None = None, group_id: int | None = None, + nickname: str | None = None, headimgurl: str | None = None, unionid: str | None = None, +) -> CpsWxUser: + """按 openid upsert。 + + - 首次:建行,记来源 code/group。 + - 已存在:仅在传入非 None 时更新昵称/头像/unionid(base 阶段为 None,不覆盖已有画像); + 并刷新 last_seen(显式 set,因 onupdate 仅在字段有变更时触发)。 + """ + u = get_by_openid(db, openid) + if u is None: + u = CpsWxUser( + openid=openid, first_code=code, first_group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + db.add(u) + else: + if nickname is not None: + u.nickname = nickname + if headimgurl is not None: + u.headimgurl = headimgurl + if unionid is not None: + u.unionid = unionid + u.last_seen = func.now() + db.commit() + db.refresh(u) + return u