Compare commits

..

12 Commits

Author SHA1 Message Date
guke eb15018322 test(welfare): 不活跃测试加 autouse 隔离夹具(清零余额+清活跃事件/审计),去掉临时 band-aid
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 19:13:25 +08:00
guke a6cfef64a8 feat(welfare): 不活跃清零-选取与逐用户清零(三桶归零+审计+流水) 2026-07-16 18:54:40 +08:00
guke 7802aeb163 feat(welfare): 可插拔不活跃预警通知器 + LogNotifier 占位
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-16 18:47:50 +08:00
guke f843e68d0f feat(welfare): INACTIVITY_* 配置项 + warn stages 解析
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-16 18:47:02 +08:00
guke 108c175142 feat(welfare): 活跃口径共享模块-子查询与 last_active_expr 2026-07-16 18:40:36 +08:00
guke e6f467a935 feat(welfare): 活跃口径共享模块-常量与北京0点cutoff助手 2026-07-16 18:39:51 +08:00
guke 708e454959 feat(welfare): 迁移新增 inactivity_reset_log / inactivity_notification_log 两表
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-16 18:34:07 +08:00
guke 0ec6b2bc56 feat(welfare): inactivity_reset_log + inactivity_notification_log 模型 2026-07-16 18:23:20 +08:00
guke 23e4671312 docs(welfare): 15天不活跃清零 实现计划(11 任务, TDD 分解)
从已定稿 spec 拆出可执行计划:模型/迁移/活跃口径共享模块/配置/通知器/清零+预警业务逻辑/
每日 worker/admin 重构/表字典。每任务含失败测试→实现→通过→提交的 bite-sized 步骤,含完整代码。
活跃口径 = created_at 基线 + home_view/比价/领券(不含 last_login_at);home_view 事件名以
HOME_VIEW_EVENT 常量占位(前端明天定名);总闸默认关。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 18:16:47 +08:00
guke b3ed66d4eb docs(welfare): 活跃口径移除 last_login_at 改用 created_at 基线;home_view 事件名待定
按评审结论定稿:
- ① 彻底移除 last_login_at:活跃口径 = max(home_view, 比价, 领券);登录/re-login 不再计为活跃
- 连带:用 User.created_at 接替"恒非空基线"位(注册即第1日活跃,只有真实使用才推进)。
  无任何信号的用户 = 注册满 15 天即清,新用户有 15 天宽限
- ② home_view 事件名待前端明天加埋点时定,后端以 HOME_VIEW_EVENT 常量占位
- 连带影响:共享口径 admin 侧同步(移除 last_login_at + created_at 基线 + 纳入 home_view),
  admin 最近活跃/DAU 口径随之变化,回归测试按新口径更新预期
- 同步 §3/§4/§9/§10/§11/§12/§13/§14

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 17:58:45 +08:00
guke c9c35a4350 docs(welfare): 明确不活跃清零的时间边界(北京自然日, 第16日0点)
补充清零日界定义:末次活跃日记为第1日,从末次活跃起「第16日0点」(北京)清零
= 北京 00:00 of (last_active_date + 15天),按自然日 0 点对齐,非从末次活跃时刻滚动 15×24h。
- 应清零 ⟺ (cn_today() − last_active_date).days ≥ RESET_DAYS
- SQL cutoff = 北京 00:00 of (cn_today() − (RESET_DAYS−1)),新增 activity.reset_cutoff() 辅助
- 资格边界(第16日0点)与 worker 执行点 RUN_HOUR 解耦;inactive_days 改按北京自然日
- 同步更新 §4/§6/§8/§10

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 17:50:03 +08:00
guke 5f1d1a5e5f docs(welfare): 15天不活跃清零金币/现金 设计文档(spec)
连续15天不活跃自动清零金币+折算现金+邀请现金,清零前可配置节奏预警,全程留审计以备纠纷排查。

关键决策:
- 活跃口径复用"用户管理"并抽共享模块 activity.py(admin 一并重构);新增 home_view 埋点代表"进首页",last_login_at 仅作灰度期兜底
- 审计双写:inactivity_reset_log 专表(记清零前三桶余额)+ 钱包流水 biz_type=inactivity_reset
- 预警:可插拔通知器 + 日志占位(v1),后续接 JPush/短信
- 触发:进程内每日 worker(仿 daily_exchange_worker),总闸默认关

仅设计文档,尚未实现。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 17:39:10 +08:00
21 changed files with 2274 additions and 1195 deletions
@@ -0,0 +1,68 @@
"""add inactivity tables
Revision ID: 135e79414fd0
Revises: comparison_llm_cost
Create Date: 2026-07-16 18:31:02.105929
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '135e79414fd0'
down_revision: Union[str, Sequence[str], None] = 'comparison_llm_cost'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"inactivity_reset_log",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("user_id", sa.Integer(), nullable=False),
sa.Column("coin_balance_before", sa.Integer(), nullable=False),
sa.Column("cash_balance_cents_before", sa.Integer(), nullable=False),
sa.Column("invite_cash_balance_cents_before", sa.Integer(), nullable=False),
sa.Column("last_active_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("inactive_days", sa.Integer(), nullable=False),
sa.Column("reason", sa.String(length=32), nullable=False),
sa.Column("reset_at", sa.DateTime(timezone=True),
server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
with op.batch_alter_table("inactivity_reset_log", schema=None) as batch_op:
batch_op.create_index(batch_op.f("ix_inactivity_reset_log_user_id"), ["user_id"], unique=False)
batch_op.create_index(batch_op.f("ix_inactivity_reset_log_reset_at"), ["reset_at"], unique=False)
op.create_table(
"inactivity_notification_log",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("user_id", sa.Integer(), nullable=False),
sa.Column("stage", sa.Integer(), nullable=False),
sa.Column("inactive_days", sa.Integer(), nullable=False),
sa.Column("coin_balance", sa.Integer(), nullable=False),
sa.Column("cash_balance_cents", sa.Integer(), nullable=False),
sa.Column("invite_cash_balance_cents", sa.Integer(), nullable=False),
sa.Column("channel", sa.String(length=16), nullable=False),
sa.Column("status", sa.String(length=16), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True),
server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
with op.batch_alter_table("inactivity_notification_log", schema=None) as batch_op:
batch_op.create_index(batch_op.f("ix_inactivity_notification_log_user_id"), ["user_id"], unique=False)
batch_op.create_index(batch_op.f("ix_inactivity_notification_log_created_at"), ["created_at"], unique=False)
def downgrade() -> None:
with op.batch_alter_table("inactivity_notification_log", schema=None) as batch_op:
batch_op.drop_index(batch_op.f("ix_inactivity_notification_log_created_at"))
batch_op.drop_index(batch_op.f("ix_inactivity_notification_log_user_id"))
op.drop_table("inactivity_notification_log")
with op.batch_alter_table("inactivity_reset_log", schema=None) as batch_op:
batch_op.drop_index(batch_op.f("ix_inactivity_reset_log_reset_at"))
batch_op.drop_index(batch_op.f("ix_inactivity_reset_log_user_id"))
op.drop_table("inactivity_reset_log")
-32
View File
@@ -1,32 +0,0 @@
"""phone_rebind_log 表(M2 换绑 30 天限制台账)
Revision ID: phone_rebind_log
Revises: comparison_llm_cost
"""
from alembic import op
import sqlalchemy as sa
revision = "phone_rebind_log"
down_revision = "comparison_llm_cost"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"phone_rebind_log",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("phone", sa.String(length=20), nullable=False),
sa.Column("old_user_id", sa.Integer(), nullable=True),
sa.Column("new_user_id", sa.Integer(), nullable=False),
sa.Column("source", sa.String(length=32), nullable=False, server_default="wechat_conflict"),
sa.Column("rebound_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
)
op.create_index("ix_phone_rebind_log_phone", "phone_rebind_log", ["phone"])
op.create_index("ix_phone_rebind_log_rebound_at", "phone_rebind_log", ["rebound_at"])
def downgrade() -> None:
op.drop_index("ix_phone_rebind_log_rebound_at", table_name="phone_rebind_log")
op.drop_index("ix_phone_rebind_log_phone", table_name="phone_rebind_log")
op.drop_table("phone_rebind_log")
+1 -275
View File
@@ -13,7 +13,6 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, Request
from sqlalchemy.exc import IntegrityError
from app.api.deps import CurrentUser, DbSession
from app.core import test_account
@@ -23,25 +22,14 @@ from app.core.ratelimit import (
enforce_rate_limit,
record_rate_limits,
)
from app.core.security import (
TokenError,
create_bind_ticket,
create_conflict_ticket,
decode_bind_ticket,
decode_conflict_ticket,
decode_token,
issue_token_pair,
)
from app.integrations import wxpay
from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
from app.repositories import onboarding as onboarding_repo
from app.repositories import phone_rebind as rebind_repo
from app.repositories import user as user_repo
from app.schemas.auth import (
JverifyLoginRequest,
LogoutResponse,
OccupiedAccountInfo,
RefreshRequest,
SmsLoginRequest,
SmsSendRequest,
@@ -49,13 +37,6 @@ from app.schemas.auth import (
TokenPair,
TokenWithUser,
UserOut,
WechatBindPhoneJverifyRequest,
WechatBindPhoneSmsRequest,
WechatBindResultResponse,
WechatConflictContinueRequest,
WechatConflictRebindRequest,
WechatLoginRequest,
WechatLoginResponse,
)
logger = logging.getLogger("shagua.auth")
@@ -194,261 +175,6 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
return _login_response(user, onboarding_completed=completed)
# ===================== 微信登录 =====================
@router.post(
"/wechat-login",
response_model=WechatLoginResponse,
summary="微信登录(openid 命中即登入,否则发绑号令牌)",
)
def wechat_login(req: WechatLoginRequest, db: DbSession) -> WechatLoginResponse:
from app.core.config import settings # 局部 import,避免循环
# 微信登录只需 code→openid(sns/oauth2),不需要商户转账证书;故只校验 APP_ID/SECRET。
if not (settings.WECHAT_APP_ID and settings.WECHAT_APP_SECRET):
raise HTTPException(status_code=503, detail="wechat login not configured")
try:
info = wxpay.code_to_userinfo(req.code) # {openid, nickname, avatar_url, raw};失败抛 ValueError
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
openid = info["openid"]
user = user_repo.get_user_by_wechat_openid(db, openid)
if user is not None:
# openid 命中 → 直接登入(绝不套用提现 bind-wechat 的"撞号即 409"逻辑)
if user.status != "active":
raise HTTPException(status_code=403, detail="account disabled")
user_repo.touch_last_login(db, user)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat_login hit user_id=%d openid=%s*** onboarded=%s", user.id, openid[:6], completed)
return WechatLoginResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
# 未命中 → 签发短时 bind_ticket,进手机号绑定流程(账号此刻还不建)
ticket = create_bind_ticket(
openid=openid,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
logger.info("wechat_login new openid=%s*** issue bind_ticket", openid[:6])
return WechatLoginResponse(
status="need_bind_phone",
bind_ticket=ticket,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
def _finish_wechat_bind(
db,
*,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
phone: str,
device_id: str,
) -> WechatBindResultResponse:
"""绑手机建号的公共尾段:手机号被占用 → 返回 phone_occupied(M2 处理 3 选 1);
未占用 → 新建微信账号(channel=wechat,昵称头像取微信)→ 签 token 登入。"""
existing = user_repo.get_user_by_phone(db, phone)
if existing is not None:
from app.core.config import settings # 局部 import,避免循环
ticket = create_conflict_ticket(
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
phone=phone,
)
blocked = rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
logger.info(
"wechat bind phone occupied phone=%s by user_id=%d has_wechat=%s",
mask_phone(phone), existing.id, bool(existing.wechat_openid),
)
return WechatBindResultResponse(
status="phone_occupied",
occupied_account=OccupiedAccountInfo(
nickname=existing.nickname,
avatar_url=existing.avatar_url,
created_at=existing.created_at,
has_wechat=bool(existing.wechat_openid),
),
conflict_ticket=ticket,
rebind_available=not blocked,
rebind_blocked_days=(
rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
if blocked else 0
),
)
user = user_repo.create_wechat_user(
db,
phone=phone,
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=device_id)
logger.info("wechat bind ok user_id=%d phone=%s openid=%s*** onboarded=%s",
user.id, mask_phone(phone), openid[:6], completed)
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
@router.post(
"/wechat/bind-phone/sms",
response_model=WechatBindResultResponse,
summary="微信登录·其他手机号(短信)绑定",
)
def wechat_bind_phone_sms(
req: WechatBindPhoneSmsRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
# 防刷:同 sms/login,按 设备+IP 每小时限流(放在验证码校验之前,失败也计数)
enforce_rate_limit(
request,
scope="wechat-bind-sms-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
)
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
return _finish_wechat_bind(
db,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=req.phone,
device_id=req.device_id,
)
@router.post(
"/wechat/bind-phone/jverify",
response_model=WechatBindResultResponse,
summary="微信登录·本机号(极光)绑定",
)
def wechat_bind_phone_jverify(
req: WechatBindPhoneJverifyRequest, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
try:
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
return _finish_wechat_bind(
db,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=phone,
device_id=req.device_id,
)
# ===================== 微信占用冲突(M2) =====================
@router.post(
"/wechat/conflict/continue",
response_model=WechatBindResultResponse,
summary="微信占用冲突·继续绑定(登录老账号,能绑就绑)",
)
def wechat_conflict_continue(
req: WechatConflictContinueRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_conflict_ticket(req.conflict_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
)
user = user_repo.get_user_by_phone(db, claims["phone"])
if user is None:
# P 期间被腾空(老账号改号/注销)→ 前提已变,让前端重走
raise HTTPException(status_code=409, detail="账号状态已变化,请重新登录")
if user.status != "active":
raise HTTPException(status_code=403, detail="account disabled")
if user.wechat_openid is None:
try:
user_repo.attach_wechat_to_user(
db, user, openid=claims["openid"],
wechat_nickname=claims["wnk"], wechat_avatar_url=claims["wav"],
)
except IntegrityError:
db.rollback() # openid 被别处绑走 → 只登入不绑
user_repo.touch_last_login(db, user)
else:
user_repo.touch_last_login(db, user) # X 已绑别的微信 → 只登入,丢弃本次 openid
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat conflict continue user_id=%d openid=%s***", user.id, claims["openid"][:6])
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
@router.post(
"/wechat/conflict/rebind",
response_model=WechatBindResultResponse,
summary="微信占用冲突·换绑(注销老账号+用该号重建全新账号)",
)
def wechat_conflict_rebind(
req: WechatConflictRebindRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
from app.core.config import settings # 局部 import,避免循环
try:
claims = decode_conflict_ticket(req.conflict_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
)
phone = claims["phone"]
if rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS):
days = rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
raise HTTPException(status_code=409, detail=f"该手机号 {days} 天内已换绑过,暂不能再次换绑")
user = user_repo.rebind_account(
db, phone=phone, openid=claims["openid"],
wechat_nickname=claims["wnk"], wechat_avatar_url=claims["wav"],
)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat conflict rebind new_user_id=%d phone=%s openid=%s***",
user.id, mask_phone(phone), claims["openid"][:6])
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
# ===================== Refresh =====================
@router.post("/refresh", response_model=TokenPair, summary="用 refresh_token 换新 token 对")
+20 -9
View File
@@ -44,11 +44,6 @@ class Settings(BaseSettings):
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(JWT_SECRET_KEY 签名,typ=wechat_bind;
# 见 security.create_bind_ticket)。需覆盖"授权→输手机号→收短信→输验证码"整个绑定流程。
WECHAT_BIND_TICKET_EXPIRE_MINUTES: int = 10
# 一个手机号 30 天内最多换绑一次(微信占用冲突页的"换绑"动作)。见 phone_rebind_log。
PHONE_REBIND_LIMIT_DAYS: int = 30
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
@@ -86,7 +81,6 @@ class Settings(BaseSettings):
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
# ===== 测试账号(release 包全流程联调用)=====
@@ -112,9 +106,6 @@ class Settings(BaseSettings):
# 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
# 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
MT_CPS_PROXY: str = ""
# 本地开发:开启后 /feed 接口直接返回 mock 数据,不调美团 API、不查离线库,
# 方便前端联调 feed 卡片样式、分页、距离排序等 UI。生产必须 false。
MT_CPS_MOCK_FEED: bool = True
@property
def mt_cps_configured(self) -> bool:
@@ -178,6 +169,13 @@ class Settings(BaseSettings):
# 进程内自动兑换 worker 的检查间隔(秒):每隔这么久醒一次,跨过北京 0 点就跑一轮。
# 默认 600s=10min,即 0 点后最多 10 分钟内兑完(客户端文案已注明「可能存在延迟」)。
AUTO_EXCHANGE_CHECK_INTERVAL_SEC: int = 600
# === 15 天不活跃清零(app.core.inactivity_reset_worker)===
INACTIVITY_RESET_ENABLED: bool = False # 总闸,默认关;灰度验证后再开
INACTIVITY_RESET_DAYS: int = 15 # 不活跃阈值(天),第 (N+1) 日 0 点清
INACTIVITY_WARN_DAYS_BEFORE: str = "7,2" # 清零前几天各推一次;""=不推。逗号分隔
INACTIVITY_RESET_RUN_HOUR: int = 3 # 北京时间每日执行点(0-23)
INACTIVITY_NOTIFY_CHANNEL: str = "log" # log(占位) / jpush / sms
INACTIVITY_RESET_CHECK_INTERVAL_SEC: int = 1800 # worker 唤醒间隔(秒)
# 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
# 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
# (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
@@ -198,6 +196,19 @@ class Settings(BaseSettings):
"""免确认收款授权可用 = 微信支付凭证齐全 + 授权回调地址已配。"""
return bool(self.wxpay_configured and self.WXPAY_AUTH_NOTIFY_URL)
@property
def inactivity_warn_stages(self) -> list[int]:
"""解析 INACTIVITY_WARN_DAYS_BEFORE → 降序去重的提前天数列表。
丢弃非数字 / <=0 / >=RESET_DAYS 的项(空串 → 空列表 = 不推)。"""
out: list[int] = []
for part in (self.INACTIVITY_WARN_DAYS_BEFORE or "").split(","):
part = part.strip()
if part.isdigit():
v = int(part)
if 0 < v < self.INACTIVITY_RESET_DAYS and v not in out:
out.append(v)
return sorted(out, reverse=True)
# ===== 穿山甲激励视频(服务端发奖回调)=====
# 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
# 穿山甲后台配置的"奖励校验密钥"(m-key),验签用。每个 GroMore 广告位 m-key 不同(后台各自
-85
View File
@@ -87,91 +87,6 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
}
def create_bind_ticket(
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> str:
"""微信登录未命中 openid 时,签发短时"待绑手机"令牌,承载 openid + 微信昵称头像。
typ='wechat_bind'、sub=openid;有效期 settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES 分钟。
与 access/refresh 用同一 JWT_SECRET_KEY 签名,靠 typ 区分,decode_bind_ticket 校验 typ。
"""
now = _now()
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": openid,
"typ": "wechat_bind",
"wnk": wechat_nickname,
"wav": wechat_avatar_url,
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_bind_ticket(token: str) -> dict[str, Any]:
"""解析"待绑手机"令牌,校验签名/过期/类型。失败抛 TokenError。
返回 {'openid': str, 'wnk': str|None, 'wav': str|None}。
"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
except jwt.ExpiredSignatureError as e:
raise TokenError("bind ticket expired") from e
except jwt.InvalidTokenError as e:
raise TokenError(f"invalid bind ticket: {e}") from e
if payload.get("typ") != "wechat_bind":
raise TokenError(f"wrong token type: want=wechat_bind got={payload.get('typ')}")
if "sub" not in payload:
raise TokenError("bind ticket missing sub")
return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")}
def create_conflict_ticket(
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None, phone: str
) -> str:
"""手机号占用时签发的短时"冲突处理"令牌。
比 bind_ticket 多编码 **已验证的手机号 phone** —— 换绑/继续绑定只认它,证明"这对
openid/手机号刚在绑号时验证通过",免用户重输验证码,又堵住"拿自己 openid + 任意手机号
去夺号"的接管漏洞。typ='wechat_conflict';有效期复用 WECHAT_BIND_TICKET_EXPIRE_MINUTES。
"""
now = _now()
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": openid,
"typ": "wechat_conflict",
"wnk": wechat_nickname,
"wav": wechat_avatar_url,
"phn": phone,
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_conflict_ticket(token: str) -> dict[str, Any]:
"""解析"冲突处理"令牌,校验签名/过期/类型。失败抛 TokenError。
返回 {'openid': str, 'wnk': str|None, 'wav': str|None, 'phone': str}。
"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
except jwt.ExpiredSignatureError as e:
raise TokenError("conflict ticket expired") from e
except jwt.InvalidTokenError as e:
raise TokenError(f"invalid conflict ticket: {e}") from e
if payload.get("typ") != "wechat_conflict":
raise TokenError(f"wrong token type: want=wechat_conflict got={payload.get('typ')}")
if "sub" not in payload or "phn" not in payload:
raise TokenError("conflict ticket missing sub/phn")
return {
"openid": payload["sub"],
"wnk": payload.get("wnk"),
"wav": payload.get("wav"),
"phone": payload["phn"],
}
# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
+43
View File
@@ -0,0 +1,43 @@
"""不活跃预警通知器(可插拔)。
v1 仅日志占位(LogNotifier):现状无真实推送能力(极光只用于一键登录解密 + 设备心跳告警,
心跳 worker 也只打印),先把清零主流程 + 审计做扎实。后续实现同协议的 JPushNotifier /
SmsNotifier 即可替换,worker/repo 不改。
"""
from __future__ import annotations
import logging
from typing import Protocol
logger = logging.getLogger("shagua.inactivity")
class InactivityNotifier(Protocol):
channel: str
def warn(self, *, user_id: int, coin: int, cash_cents: int, invite_cash_cents: int,
stage: int, days_until_reset: int) -> str:
"""发预警,返回状态:'sent' / 'failed' / 'placeholder'"""
...
class LogNotifier:
"""占位实现:只打印,不真推。参照 heartbeat_monitor_worker「本期先不接推送」先例。"""
channel = "log"
def warn(self, *, user_id: int, coin: int, cash_cents: int, invite_cash_cents: int,
stage: int, days_until_reset: int) -> str:
logger.warning(
"[inactivity-warn] user=%s coin=%s cash_cents=%s invite_cash_cents=%s "
"stage=T-%s days_until_reset=%s",
user_id, coin, cash_cents, invite_cash_cents, stage, days_until_reset,
)
return "placeholder"
def get_notifier(channel: str) -> InactivityNotifier:
"""按配置返回通知器。未实现的通道(jpush/sms)暂回退 LogNotifier 占位。"""
# 后续:if channel == "jpush": return JPushNotifier()
# if channel == "sms": return SmsNotifier()
return LogNotifier()
+4 -1
View File
@@ -27,12 +27,15 @@ from app.models.coupon_state import ( # noqa: F401
CouponSession,
)
from app.models.feedback import Feedback # noqa: F401
from app.models.inactivity import ( # noqa: F401
InactivityNotificationLog,
InactivityResetLog,
)
from app.models.invite import InviteRelation # noqa: F401
from app.models.invite_fingerprint import InviteFingerprint # noqa: F401
from app.models.launch_confirm_sample import LaunchConfirmSample # noqa: F401
from app.models.meituan_coupon import MeituanCoupon # noqa: F401
from app.models.onboarding import OnboardingCompletion # noqa: F401
from app.models.phone_rebind_log import PhoneRebindLog # noqa: F401
from app.models.ops_marquee_seed import OpsMarqueeSeed # noqa: F401
from app.models.ops_stat_config import OpsStatConfig # noqa: F401
from app.models.price_observation import PriceObservation # noqa: F401
+56
View File
@@ -0,0 +1,56 @@
"""15 天不活跃清零相关表。
- inactivity_reset_log:每次清零一行,记清零前三桶余额 + 原因 + 判定时活跃时间/不活跃天数,
供纠纷排查(需求①)。清零同时另写 3 条钱包流水(biz_type=inactivity_reset),资金流可逐笔回溯。
- inactivity_notification_log:每次预警一行,记推送时余额快照 + 档位 + 通道 + 状态,
兼作"预警去重"依据(created_at > last_active)与"待推送"占位 outbox(v1 通道=log)。
append-only,不更新。user_id 只索引、不设外键(同 analytics_event,避免删用户级联/历史留痕)。
"""
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 InactivityResetLog(Base):
__tablename__ = "inactivity_reset_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
coin_balance_before: Mapped[int] = mapped_column(Integer, nullable=False)
cash_balance_cents_before: Mapped[int] = mapped_column(Integer, nullable=False)
invite_cash_balance_cents_before: Mapped[int] = mapped_column(Integer, nullable=False)
last_active_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
inactive_days: Mapped[int] = mapped_column(Integer, nullable=False)
reason: Mapped[str] = mapped_column(String(32), nullable=False)
reset_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"<InactivityResetLog id={self.id} user_id={self.user_id} coin={self.coin_balance_before}>"
class InactivityNotificationLog(Base):
__tablename__ = "inactivity_notification_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
stage: Mapped[int] = mapped_column(Integer, nullable=False) # 提前天数档(如 7 / 2)
inactive_days: Mapped[int] = mapped_column(Integer, nullable=False)
coin_balance: Mapped[int] = mapped_column(Integer, nullable=False)
cash_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False)
invite_cash_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False)
channel: Mapped[str] = mapped_column(String(16), nullable=False) # log / jpush / sms
status: Mapped[str] = mapped_column(String(16), nullable=False) # placeholder / sent / failed
created_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"<InactivityNotificationLog id={self.id} user_id={self.user_id} stage={self.stage}>"
-31
View File
@@ -1,31 +0,0 @@
"""手机号换绑台账。
记录"手机号从老账号被夺走、重建为新账号(X 注销 → Y)"这一破坏性事件,支撑"一个手机号
30 天内最多换绑一次"的限制。手机号级、渠道无关(source 标来源);普通微信绑定不写此表。
见 M2 spec §4.1。
"""
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 PhoneRebindLog(Base):
__tablename__ = "phone_rebind_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# 被换绑的真实手机号(注意:存真实号,不是老账号被腾号后的 deleted_<id>)
phone: Mapped[str] = mapped_column(String(20), index=True, nullable=False)
# 被注销的老账号 X;P 换绑时已被腾空(极边界)则为空
old_user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 换绑后新建的账号 Y
new_user_id: Mapped[int] = mapped_column(Integer, nullable=False)
# 换绑来源。手机号级配额、渠道无关,留字段给未来其他换绑路径共用同一份 30 天限制。
source: Mapped[str] = mapped_column(String(32), nullable=False, default="wechat_conflict")
rebound_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
+89
View File
@@ -0,0 +1,89 @@
"""活跃口径唯一真源:worker(不活跃清零)与 admin(最近活跃/DAU)共用,防两处漂移。
口径 = max(User.created_at, AnalyticsEvent[ACTIVE_EVENTS], CouponPromptEngagement[claim_started])。
**不含 last_login_at**(登录/re-login 不代表在用 App);created_at 为恒非空基线。
清零/预警按北京自然日 0 点对齐(见 reset_cutoff)。
"""
from __future__ import annotations
from datetime import date, datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.core.rewards import CN_TZ, cn_today
from app.models.analytics_event import AnalyticsEvent
from app.models.coupon_state import CouponPromptEngagement
# —— 活跃事件名(与"用户管理"口径一致)——
# home_view:进首页(前端埋点,名称前端明天敲定,此处占位;定名后仅改这一常量)。
HOME_VIEW_EVENT = "home_view"
COMPARE_START_EVENT = "real_compare_start" # 发起比价(含浮窗触发)
COUPON_START_EVENT = "real_coupon_start" # 发起领券
ACTIVE_EVENTS = (HOME_VIEW_EVENT, COMPARE_START_EVENT, COUPON_START_EVENT)
ACTIVE_ENGAGE_TYPE = "claim_started" # coupon_prompt_engagement 一键领取
def as_utc(value: datetime) -> datetime:
"""任意 datetime → tz-aware UTC(无时区按 UTC 解释)。用于与 DateTime(timezone=True) 列比较,
比较绝对时刻、与会话时区无关(口径同 admin queries._as_utc)。"""
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
def norm_utc(dt: datetime | None) -> datetime | None:
"""naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
if dt is None:
return None
return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
def cn_midnight_utc(d: date) -> datetime:
"""北京 d 日 00:00 → tz-aware UTC datetime。"""
return as_utc(datetime(d.year, d.month, d.day, tzinfo=CN_TZ))
def reset_cutoff(reset_days: int, today: date | None = None) -> datetime:
"""应清零边界(tz-aware UTC):last_active < 此值 ⟺ 距末次活跃已满 reset_days 天(北京 0 点对齐)。
= 北京 00:00 of (today (reset_days 1))。例:reset_days=15、today=1/20 → 北京 1/6 00:00。"""
today = today or cn_today()
return cn_midnight_utc(today - timedelta(days=reset_days - 1))
def last_active_subqueries(db: Session):
"""两个按 user_id 预聚合的派生表:最近活跃事件(ACTIVE_EVENTS)、最近领券发起(claim_started)。
返回 (ev_sub, eng_sub)。口径同 admin,LEFT JOIN 用,借事件索引只扫活跃事件。"""
ev_sub = (
select(
AnalyticsEvent.user_id.label("user_id"),
func.max(AnalyticsEvent.created_at).label("last_at"),
)
.where(AnalyticsEvent.user_id.is_not(None), AnalyticsEvent.event.in_(ACTIVE_EVENTS))
.group_by(AnalyticsEvent.user_id)
.subquery()
)
eng_sub = (
select(
CouponPromptEngagement.user_id.label("user_id"),
func.max(CouponPromptEngagement.created_at).label("last_at"),
)
.where(
CouponPromptEngagement.user_id.is_not(None),
CouponPromptEngagement.engage_type == ACTIVE_ENGAGE_TYPE,
)
.group_by(CouponPromptEngagement.user_id)
.subquery()
)
return ev_sub, eng_sub
def last_active_expr(base_col, ev_sub, eng_sub, dialect: str):
"""max(base_col, 最近活跃事件, 最近领券) 的 SQL 表达式。PG 用 greatest、SQLite 用 max。
子聚合缺失(未命中)时 coalesce 到 base_col(= User.created_at,恒非空基线)。"""
greatest = func.greatest if dialect == "postgresql" else func.max
return greatest(
base_col,
func.coalesce(ev_sub.c.last_at, base_col),
func.coalesce(eng_sub.c.last_at, base_col),
)
+107
View File
@@ -0,0 +1,107 @@
"""15 天不活跃清零业务逻辑(纯同步,可单测)。worker 只是它的 asyncio 外壳。
活跃口径复用 app.repositories.activity;清零走 wallet.grant_*(负数出账写流水 commit)
逐用户独立事务,一个失败不影响其余
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from sqlalchemy import or_, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from app.core.rewards import CN_TZ
from app.models.inactivity import InactivityResetLog
from app.models.user import User
from app.models.wallet import CoinAccount
from app.repositories import activity
from app.repositories import wallet as wallet_repo
RESET_BIZ_TYPE = "inactivity_reset"
RESET_REMARK = "15天不活跃清零"
_ANY_BALANCE = or_(
CoinAccount.coin_balance > 0,
CoinAccount.cash_balance_cents > 0,
CoinAccount.invite_cash_balance_cents > 0,
)
def _base_query(db: Session):
"""select(user_id, last_active, 三桶余额),join CoinAccount + 两活跃子查询。"""
ev_sub, eng_sub = activity.last_active_subqueries(db)
dialect = db.get_bind().dialect.name
last_active = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
stmt = (
select(
User.id.label("user_id"),
last_active.label("last_active"),
CoinAccount.coin_balance,
CoinAccount.cash_balance_cents,
CoinAccount.invite_cash_balance_cents,
)
.join(CoinAccount, CoinAccount.user_id == User.id)
.outerjoin(ev_sub, ev_sub.c.user_id == User.id)
.outerjoin(eng_sub, eng_sub.c.user_id == User.id)
)
return stmt, last_active
def _cn_date(dt: datetime) -> date:
"""datetime → 北京自然日(naive 视为 UTC)。"""
return activity.norm_utc(dt).astimezone(CN_TZ).date()
def _inactive_days(last_active: datetime, today: date) -> int:
return (today - _cn_date(last_active)).days
def select_inactive_users(db: Session, *, cutoff: datetime):
"""应清零用户:last_active < cutoff 且三桶有余额。返回 Row 列表(值已快照,可跨 commit)。"""
stmt, last_active = _base_query(db)
stmt = stmt.where(_ANY_BALANCE, last_active < activity.as_utc(cutoff))
return db.execute(stmt).all()
def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int, reason: str) -> bool:
"""单用户清零(独立事务、行锁)。三桶归零 + 写审计 + 3 条流水。返回是否真清了(有余额)。"""
acc = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
coin, cash, invite = acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents
if coin == 0 and cash == 0 and invite == 0:
return False
log = InactivityResetLog(
user_id=user_id, coin_balance_before=coin, cash_balance_cents_before=cash,
invite_cash_balance_cents_before=invite, last_active_at=activity.norm_utc(last_active),
inactive_days=inactive_days, reason=reason,
)
db.add(log)
db.flush() # 拿 log.id 作 ref_id 交叉链接审计↔流水
ref = str(log.id)
if coin:
wallet_repo.grant_coins(db, user_id, -coin, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
if cash:
wallet_repo.grant_cash(db, user_id, -cash, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
if invite:
wallet_repo.grant_invite_cash(db, user_id, -invite, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
db.commit()
return True
def run_reset_once(db: Session, *, reset_days: int, today: date) -> dict:
"""扫一轮清零。逐用户独立 commit,失败隔离。"""
stats = {"scanned": 0, "cleared": 0, "failed": 0}
cutoff = activity.reset_cutoff(reset_days, today)
reason = f"inactive_{reset_days}d"
rows = select_inactive_users(db, cutoff=cutoff) # 先物化,避免边遍历边 commit
for row in rows:
stats["scanned"] += 1
idays = _inactive_days(row.last_active, today)
try:
if clear_user(db, user_id=row.user_id, last_active=row.last_active,
inactive_days=idays, reason=reason):
stats["cleared"] += 1
except SQLAlchemyError:
db.rollback()
stats["failed"] += 1
return stats
-39
View File
@@ -1,39 +0,0 @@
"""手机号换绑台账(phone_rebind_log)的查询与写入。见 M2 spec §4.1。"""
from __future__ import annotations
import math
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.phone_rebind_log import PhoneRebindLog
def rebound_within_days(db: Session, phone: str, days: int) -> bool:
"""该手机号在最近 days 天内是否换绑过(命中 → 禁止再次换绑)。"""
since = datetime.now(timezone.utc) - timedelta(days=days)
stmt = (
select(PhoneRebindLog.id)
.where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
.limit(1)
)
return db.execute(stmt).first() is not None
def remaining_block_days(db: Session, phone: str, days: int) -> int:
"""距离该手机号可再次换绑还剩几天(向上取整;无记录返回 0)。"""
last = db.execute(
select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
).scalar_one_or_none()
if last is None:
return 0
if last.tzinfo is None: # SQLite 取回 naive datetime,按 UTC 归一
last = last.replace(tzinfo=timezone.utc)
remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
return max(0, math.ceil(remaining / 86400))
def add_rebind_log(db: Session, *, phone: str, old_user_id: int | None, new_user_id: int, source: str) -> None:
"""写一条换绑台账(**不 commit**,交给调用方 rebind_account 的单事务)。"""
db.add(PhoneRebindLog(phone=phone, old_user_id=old_user_id, new_user_id=new_user_id, source=source))
-132
View File
@@ -12,7 +12,6 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.user import User
from app.repositories import phone_rebind
# ===== 创建时分配的标识:用户名(对外展示账号 ID)+ 默认昵称 =====
@@ -59,17 +58,6 @@ def is_default_nickname(nickname: str | None) -> bool:
)
def apply_wechat_display_identity(
user: User, *, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> None:
"""§10:用已有账号绑微信时,仅当展示字段仍为默认才用微信昵称/头像替换(两规则独立);
自定义(改过昵称/传过头像)则保留只改内存对象,由调用方 commit"""
if is_default_nickname(user.nickname) and wechat_nickname:
user.nickname = wechat_nickname
if user.avatar_url is None and wechat_avatar_url:
user.avatar_url = wechat_avatar_url
def get_user_by_username(db: Session, username: str) -> User | None:
return db.execute(
select(User).where(User.username == username)
@@ -98,85 +86,6 @@ def get_user_by_phone(db: Session, phone: str) -> User | None:
return db.execute(stmt).scalar_one_or_none()
def get_user_by_wechat_openid(db: Session, openid: str) -> User | None:
stmt = select(User).where(User.wechat_openid == openid)
return db.execute(stmt).scalar_one_or_none()
def touch_last_login(db: Session, user: User) -> User:
"""openid 命中登录时更新 last_login_at(手机号登录在 upsert_user_for_login 里已更新)。"""
user.last_login_at = datetime.now(timezone.utc)
db.commit()
db.refresh(user)
return user
def attach_wechat_to_user(
db: Session, user: User, *, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> User:
"""继续绑定:把微信 openid + 微信源字段并入已存在账号(调用方保证 user.wechat_openid 为空)。
wechat_openid / wechat_nickname / wechat_avatar_url,并按 §10 规则回填展示字段:
仅当昵称仍为默认值(is_default_nickname)时用微信昵称替换,仅当头像为 null 时用微信头像替换;
用户已自定义的展示昵称/头像始终保留,两规则相互独立
openid 唯一约束(O 期间被别处绑走,极罕见)时由调用方捕获 IntegrityError 兜底降级为"只登入不绑"
"""
user.wechat_openid = openid
user.wechat_nickname = wechat_nickname
user.wechat_avatar_url = wechat_avatar_url
user.last_login_at = datetime.now(timezone.utc)
apply_wechat_display_identity(user, wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url)
db.commit()
db.refresh(user)
return user
def _build_wechat_user(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
) -> User:
"""构造并 db.add 一个微信账号行(register_channel='wechat',展示昵称头像取微信,缺则默认),
** commit**create_wechat_user rebind_account 共用,保证建号逻辑单一来源"""
user = User(
phone=phone,
username=_gen_unique_username(db),
nickname=wechat_nickname or _gen_nickname(),
avatar_url=wechat_avatar_url,
register_channel="wechat",
wechat_openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
last_login_at=datetime.now(timezone.utc),
)
db.add(user)
return user
def create_wechat_user(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
) -> User:
"""微信登录新建账号(未占用分支)。见 _build_wechat_user。
openid 唯一约束是并发/重复绑定的最终防线(极罕见,openid wechat-login 刚查过为空)
"""
user = _build_wechat_user(
db, phone=phone, openid=openid,
wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url,
)
db.commit()
db.refresh(user)
return user
def upsert_user_for_login(
db: Session,
*,
@@ -245,44 +154,3 @@ def soft_delete_account(db: Session, user: User) -> None:
# 释放邀请码唯一槽
user.invite_code = None
db.commit()
def rebind_account(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
source: str = "wechat_conflict",
) -> User:
"""换绑:**单事务内**注销老账号 X(腾出手机号)+ 用该号建全新微信账号 Y + 写换绑台账。
- 老账号可能已不存在(P 被腾空) old_user_id=None,直接建 Y(幂等更稳)
- 手机号唯一约束靠时序:先把 X.phone 改名并 flush 腾号,再插 Y
- 全程不中途 commit,任一步失败整体回滚,绝不出现"X 删了 Y 没建"
X 的字段变更等价 soft_delete_account(软删 + 匿名化 + 释放 openid/邀请码唯一槽),但不在此 commit
"""
old = get_user_by_phone(db, phone)
old_id = old.id if old is not None else None
if old is not None:
old.status = "deleted"
old.phone = f"deleted_{old.id}"
old.nickname = None
old.avatar_url = None
old.wechat_openid = None
old.wechat_nickname = None
old.wechat_avatar_url = None
old.invite_code = None
db.flush() # 先落 phone 改名,腾出手机号唯一约束,才能给 Y 用
new_user = _build_wechat_user(
db, phone=phone, openid=openid,
wechat_nickname=wechat_nickname, wechat_avatar_url=wechat_avatar_url,
)
db.flush() # 拿 new_user.id
phone_rebind.add_rebind_log(
db, phone=phone, old_user_id=old_id, new_user_id=new_user.id, source=source
)
db.commit()
db.refresh(new_user)
return new_user
+4 -12
View File
@@ -20,7 +20,6 @@ from app.core.config import settings
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.repositories.user import apply_wechat_display_identity
from app.models.wallet import (
CashTransaction,
CoinAccount,
@@ -35,10 +34,6 @@ _WX_STATE_SUCCESS = "SUCCESS"
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认
_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
# 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
# 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
# (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
_NEWBIE_TIER_HELD_STATUSES = {"reviewing", "pending", "success"}
# 免确认收款授权状态
_WX_AUTH_ACTIVE = "TAKING_EFFECT" # 已生效,可免确认转账
_WX_AUTH_CLOSED = "CLOSED" # 已关闭(用户/商户/风控),需重新开启
@@ -379,7 +374,6 @@ def bind_wechat_openid(db: Session, user_id: int, code: str) -> dict:
user.wechat_openid = info["openid"]
user.wechat_nickname = info["nickname"]
user.wechat_avatar_url = info["avatar_url"]
apply_wechat_display_identity(user, wechat_nickname=info["nickname"], wechat_avatar_url=info["avatar_url"])
db.commit()
return info
@@ -628,10 +622,9 @@ def _beijing_today_start_utc() -> datetime:
def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -> list[dict]:
"""福利页(coin_cash)提现档位的可提现状态。withdraw-info 下发与 create_withdraw 校验共用此口径。
规则(2026-07-09 拍板,7-9提现ui对齐;新人档判定 2026-07-16 修正):
- 新人档(0.1/0.3):账号历史一次性进行中(reviewing/pending)或成功打款(success)即视为
已用,直接**从返回列表消失**;被拒/转账失败/解绑退回(均已退款钱没到手)则恢复可提,不永久
占用资格两档各自独立互不影响,不参与"每日选一个额度"互斥
规则(2026-07-09 拍板,7-9提现ui对齐):
- 新人档(0.1/0.3):账号历史一次性只要发起过(**任意状态**,含被拒/失败,"发起就算")
即视为已用,直接**从返回列表消失**;两档各自独立互不影响,不参与"每日选一个额度"互斥
- 常规档(0.5×3 / 10×1 / 20×1):按北京日计次,"发起就算占用"(当天创建的单不论最终状态
都计入,被拒/失败不退当天名额);三档每天只能选一个,选定后其余两档当天 other_tier_selected
- invite_cash 本轮无档位概念 返回空列表(邀请页客户端仍用本地写死档位,行为不变)
@@ -642,7 +635,7 @@ def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -
tiers = rewards.WITHDRAW_TIERS_COIN_CASH
amounts = [t.amount_cents for t in tiers]
newbie_amounts = [t.amount_cents for t in tiers if t.is_newbie]
# 新人档历史是否用过:进行中或已成功打款的单占用资格;被拒/失败/解绑退回(已退款)不算(恢复可提)
# 新人档历史是否用过:任意时间、任意状态("发起就算")
used_newbie: set[int] = set(
db.execute(
select(WithdrawOrder.amount_cents)
@@ -651,7 +644,6 @@ def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -
WithdrawOrder.user_id == user_id,
WithdrawOrder.source == "coin_cash",
WithdrawOrder.amount_cents.in_(newbie_amounts),
WithdrawOrder.status.in_(_NEWBIE_TIER_HELD_STATUSES),
)
).scalars()
) if newbie_amounts else set()
-61
View File
@@ -102,64 +102,3 @@ class RefreshRequest(BaseModel):
class LogoutResponse(BaseModel):
ok: bool = True
# ===== 微信登录 =====
class WechatLoginRequest(BaseModel):
code: str = Field(..., min_length=1, description="微信 App 授权拿到的 code(单次有效)")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
)
class WechatLoginResponse(BaseModel):
# status="logged_in" → openid 命中,token 有值;"need_bind_phone" → 未命中,bind_ticket 有值
status: str
token: TokenWithUser | None = None
bind_ticket: str | None = None
wechat_nickname: str | None = None
wechat_avatar_url: str | None = None
class OccupiedAccountInfo(BaseModel):
"""手机号被占用时返回的原账号脱敏展示信息(供冲突页)。"""
nickname: str | None = None
avatar_url: str | None = None
created_at: datetime
has_wechat: bool = False
class WechatBindResultResponse(BaseModel):
# status="logged_in" → 未占用,已建号登入,token 有值;
# "phone_occupied" → 手机号被占用,occupied_account + conflict_ticket 有值,token 为 None
status: str
token: TokenWithUser | None = None
occupied_account: OccupiedAccountInfo | None = None
conflict_ticket: str | None = None # 占用时签发,换绑/继续绑定只认它
rebind_available: bool | None = None # 该手机号 30 天内是否还能换绑(给换绑按钮预置禁用态)
rebind_blocked_days: int | None = None # 被限时剩余天数(rebind_available=False 时>0)
class WechatBindPhoneSmsRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
code: str = Field(..., min_length=4, max_length=8)
device_id: str = Field("", max_length=64)
class WechatBindPhoneJverifyRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
login_token: str = Field(..., min_length=1, description="客户端 loginAuth 拿到的 loginToken")
device_id: str = Field("", max_length=64)
class WechatConflictContinueRequest(BaseModel):
conflict_ticket: str = Field(..., min_length=1)
device_id: str = Field("", max_length=64)
class WechatConflictRebindRequest(BaseModel):
conflict_ticket: str = Field(..., min_length=1)
device_id: str = Field("", max_length=64)
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,296 @@
# 15 天不活跃自动清零(金币 + 现金)设计
- **日期**2026-07-16
- **状态**Draft — 待评审
- **所属**app-server`app/`),含一处 admin 侧重构 + 一项 Android 端埋点依赖
- **一句话**:连续 15 天不活跃的用户,自动清零其金币与现金;清零前按可配置节奏预警;全过程留审计以备纠纷排查。
---
## 1. 背景与目标
运营需要对**长期不活跃**用户的钱包余额做清理。两条硬性要求:
1. **可审计**:记录清零原因与**清零前的三桶余额**,便于后续排查与处理客户纠纷。
2. **临清预警**:在临近清零前推送信息告知用户"因账号不活跃,账户里的 xx 金币和 xx 现金将被清零"。
### 非目标(本期不做)
- 不做真实推送通道(极光 JPush / 短信)的对接 —— 仅做**可插拔通知器 + 日志占位**,接口预留、后续无缝替换。
- 不改动提现(`WithdrawOrder`)流程。
- 不新增 `User.last_active_at` 列、不改鉴权热路径。
---
## 2. 需求
| # | 需求 | 落地 |
|---|---|---|
| R1 | 连续 15 天不活跃 → 清零金币 + 现金 | 每日 worker 扫描 + 逐用户事务清零(§6) |
| R2 | 记录清零原因 + 清零前余额 | `inactivity_reset_log` 审计表 + 3 条钱包流水(§5、§7) |
| R3 | 临清前预警"xx 金币 xx 现金将清零" | 阶段 A 预警 + `inactivity_notification_log`(§6、§7 |
| R4 | 活跃口径与"用户管理"一致 | 抽共享模块 `activity.py`admin 与 worker 共用(§4、§12 |
| R5 | 预警时机完全可配置 | `INACTIVITY_*` 配置项(§8 |
---
## 3. 决策记录(来自评审问答)
| 决策点 | 结论 | 理由 |
|---|---|---|
| **活跃口径** | 与"用户管理"一致:`max(home_view, 比价, 领券)`**不含 last_login_at**;无任何信号时以 `created_at` 为非空基线 | 比价可从**浮窗**触发、不进首页;`last_login_at` 只在登录/换绑动作更新(re-login 也算),代表不了"在用 App",故彻底排除 |
| **"进首页"信号落地** | **方案 A:前端上报 `home_view` 埋点**(复用 `/analytics/events`),非新接口 | 三个活跃信号统一为同类埋点事件;零新接口零新列;与 admin 口径天然一致。B(鉴权接口 + 列)"更权威"的优势是假的——比价/领券仍是端上报事件,最弱环决定整体可信度 |
| **清零范围** | **金币 + 折算现金 + 邀请现金** 三桶全清 | 对应"账户里的金币和现金" |
| **预警推送** | **可插拔通知器 + 日志占位**v1),后续接 JPush/短信 | 现状无真实推送能力;先把清零主流程 + 审计做扎实,不阻塞 |
| **预警时机** | **完全可配置**(提前天数列表 + 次数 + 执行点 + 通道) | R5 |
| **触发方式** | **进程内每日 worker**,仿 `daily_exchange_worker` | 与项目最新模式一致,无需外部 cron |
| **admin 共享口径** | 共享模块 + **重构 admin 改用它** | 单一真源,永不漂移(R4 |
### 已知取舍(可接受)
- analytics 的 `user_id` 是**端上报、未鉴权**(可伪造)。但伪造只能"保自己活跃、避免被清",无收益,且正是本功能要防的行为,风险良性。活跃时间的非空基线由服务端权威的 `User.created_at` 提供(见 §4),不再依赖 `last_login_at`。与"用户管理"口径一致。
---
## 4. 活跃口径与共享模块 `app/repositories/activity.py`(新建)
活跃口径的**唯一真源**。app 侧模块,admin 可 import`app.main` 不 import `app.admin`,反向允许)。
### 口径
```
last_active = max(
User.created_at, # 注册基线(恒非空;re-login 不推进,只有真实使用才推进)
max AnalyticsEvent.created_at WHERE event IN ACTIVE_EVENTS,
max CouponPromptEngagement.created_at WHERE engage_type == "claim_started",
)
不活跃判定:按北京自然日、0 点对齐(非从末次活跃时刻滚动 15×24h)
last_active_date = 北京(last_active).date() # 末次活跃的北京日,记为「第 1 日」
清零边界 = 北京 00:00 of (last_active_date + RESET_DAYS 天) =「第 (RESET_DAYS+1) 日 0 点」 # 15 → 第16日0点
应清零 ⟺ (cn_today() last_active_date).days ≥ RESET_DAYS
⟺ last_active < cutoff cutoff = 北京 00:00 of (cn_today() (RESET_DAYS1)) # 供 SQL 比较
inactive_days = (cn_today() last_active_date).days # 清零当日恰 = RESET_DAYS
例:末次活跃 1/1 → 1/16 00:00(第16日0点)清零,当日 inactive_days=151/15 及之前不清
```
### 模块内容
- 常量:
- `ACTIVE_EVENTS = (HOME_VIEW_EVENT, "real_compare_start", "real_coupon_start")` —— 进首页 + 比价 + 领券。**`HOME_VIEW_EVENT` 事件名待前端确认**(明天加埋点时定;暂记 `"home_view"`)。
- `ACTIVE_ENGAGE_TYPE = "claim_started"`
- `last_active_subqueries(db)` —— 复刻现 admin `queries._last_active_parts()`:两个按 `user_id``GROUP BY max(created_at)` 聚合子查询。
- `last_active_expr(base_col, ev_sub, eng_sub, dialect)` —— 生成 `greatest`/`max`PG `func.greatest`SQLite `func.max`);子聚合缺失时 `coalesce(子聚合, User.created_at)` 兜底(注册基线恒非空,**替代原 last_login_at**)。
- `_norm_utc()` —— 沿用现 admin 的 naive→UTC 归一(SQLite naive / PG aware 混算保护)。
- `reset_cutoff(reset_days)` / `warn_cutoff(reset_days, k)` —— 生成**北京 0 点对齐**的边界 datetime(见口径):`reset_cutoff = 北京 00:00 of (cn_today() (reset_days1))`,供下面查询按 `last_active < cutoff` 比较。
- `select_inactive_users(db, *, cutoff, with_balance=True)` —— **worker 专用**join `CoinAccount`,筛 `last_active < cutoff`(cutoff = 北京 0 点对齐边界,见口径)且(`coin_balance>0 OR cash_balance_cents>0 OR invite_cash_balance_cents>0`),返回 `(user, account, last_active, inactive_days)`
- `select_warn_targets(db, *, reset_days, warn_days_before)` —— **worker 专用**:返回 `(user, account, last_active, inactive_days, stage)` 元组——各"提前天数"窗口内、有余额、本 streak 未推过档 `stage` 的用户(去重结合 `notification_log`,逻辑见 §9)。
> **参考现状**:现口径散落在 `app/admin/repositories/queries.py:38,91-124,199-204``_ACTIVE_EVENTS`/`_last_active_parts`/`greatest`)与 `app/admin/repositories/stats.py:51-52,138-146``COMPARE_START_EVENT`/`COUPON_START_EVENT`/活跃用户集)。这些改为从 `activity.py` 导入(§12)。
---
## 5. 数据模型(2 张新表,不动 `User`
两表均登记进 `app/models/__init__.py`;一个 Alembic 迁移建两表(`render_as_batch`SQLite 兼容)。
### ① `inactivity_reset_log` —— 清零审计(R2
仿 `app/models/phone_rebind_log.py` 的简单审计表风格。
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int PK autoincrement | |
| `user_id` | int, index, not null | |
| `coin_balance_before` | int, not null | 清零前金币 |
| `cash_balance_cents_before` | int, not null | 清零前折算现金(分) |
| `invite_cash_balance_cents_before` | int, not null | 清零前邀请现金(分) |
| `last_active_at` | DateTime(tz), nullable | 判定时的最近活跃时间 |
| `inactive_days` | int, not null | 判定时不活跃天数 |
| `reason` | String(32), not null | 如 `"inactive_15d"` |
| `reset_at` | DateTime(tz), server_default now(), index, not null | 清零时刻 |
### ② `inactivity_notification_log` —— 预警记录 + 去重 + 占位 outboxR3
| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | int PK autoincrement | |
| `user_id` | int, index, not null | |
| `stage` | int, not null | 提前天数档(如 7 / 2 |
| `inactive_days` | int, not null | 推送时不活跃天数 |
| `coin_balance` | int, not null | 推送快照:告知用户的金币数 |
| `cash_balance_cents` | int, not null | 推送快照:折算现金 |
| `invite_cash_balance_cents` | int, not null | 推送快照:邀请现金 |
| `channel` | String(16), not null | `"log"` / `"jpush"` / `"sms"` |
| `status` | String(16), not null | `"placeholder"` / `"sent"` / `"failed"` |
| `created_at` | DateTime(tz), server_default now(), index, not null | 去重锚点(见 §9 |
> 备注:不新增 `User.last_active_at` 列,不改 `get_current_user`。活跃时间由 §4 口径**实时计算**。
---
## 6. 清零 worker `app/core/inactivity_reset_worker.py`(新建)
**完全仿 [`app/core/daily_exchange_worker.py`](../../../app/core/daily_exchange_worker.py)**App 启动自带 asyncio 任务,文件锁(`data/inactivity_reset.lock`)防同机多进程并发,总闸 `INACTIVITY_RESET_ENABLED`(默认关)。
### 调度
- 每 `INACTIVITY_RESET_CHECK_INTERVAL_SEC` 秒醒一次;`last_run: date` 守卫**北京日**,保证每日只跑一轮。
- 仅当 `cn_today() != last_run` 且当前北京小时 `>= INACTIVITY_RESET_RUN_HOUR` 时执行(启动补跑同 daily_exchange 语义)。
- **清零资格边界 = 第 16 日 0 点(北京,见 §4),与 worker 执行点解耦**worker 于当日 `RUN_HOUR`(默认 3 点)跑,把已过边界者一并清;若要严格 0 点触发可置 `RUN_HOUR=0`,但注意与 `daily_auto_exchange` 的 0 点任务错峰。
- lifespan 里 `start_inactivity_reset_worker()` / `stop_...`(仿 `start_daily_exchange_worker``app/main.py` 的接线)。
### 一轮 `run_once(db)` 两阶段(同一次运行、各自逐用户独立 commit)
**阶段 A — 预警**
```
for user, acc, last_active, inactive_days, stage in activity.select_warn_targets(...):
notifier.send_inactivity_warning(user, balances=snapshot(acc), stage=stage, days_until_reset=RESET_DAYS-inactive_days)
db.add(InactivityNotificationLog(..., channel=notifier.channel, status=notifier.last_status))
db.commit() # 逐条独立
```
**阶段 B — 清零**`biz_type="inactivity_reset"`
```
for user, acc, last_active, inactive_days in activity.select_inactive_users(db, cutoff=activity.reset_cutoff(RESET_DAYS)): # 北京 00:00 of (今天−(RESET_DAYS1))
try:
acc = wallet.get_or_create_account(db, user.id, commit=False, lock=True) # 行锁
before = (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents)
if before == (0,0,0): continue
log = InactivityResetLog(user_id=user.id, coin_balance_before=before[0],
cash_balance_cents_before=before[1], invite_cash_balance_cents_before=before[2],
last_active_at=last_active, inactive_days=inactive_days, reason=f"inactive_{RESET_DAYS}d")
db.add(log); db.flush() # 拿 log.id 作 ref_id 交叉链接
if acc.coin_balance: wallet.grant_coins(db, user.id, -acc.coin_balance, biz_type="inactivity_reset", ref_id=str(log.id), remark="15天不活跃清零")
if acc.cash_balance_cents: wallet.grant_cash(db, user.id, -acc.cash_balance_cents, biz_type="inactivity_reset", ref_id=str(log.id), remark="15天不活跃清零")
if acc.invite_cash_balance_cents: wallet.grant_invite_cash(db, user.id, -acc.invite_cash_balance_cents, biz_type="inactivity_reset", ref_id=str(log.id), remark="15天不活跃清零")
db.commit()
except SQLAlchemyError:
db.rollback(); stats["failed"] += 1
```
- `grant_*` 负数出账、`balance_after=0`、写三条流水;`grant_coins` 负数**不**动 `total_coin_earned`(历史累计保留)。
- 逐用户独立 commit:一个失败不影响其余。返回 `stats = {scanned, warned, cleared, failed}``logger.info`
---
## 7. 预警与可插拔通知器
`app/integrations/notifier.py` 定义协议(外部投递属 integrations 层):
```python
class InactivityNotifier(Protocol):
channel: str # "log" / "jpush" / "sms"
last_status: str # "placeholder" / "sent" / "failed"
def send_inactivity_warning(self, user, *, balances, stage, days_until_reset) -> None: ...
```
- **v1 `LogNotifier`**`channel="log"`):`logger.warning("[inactivity-warn] user=%s coin=%s cash=%s invite=%s T-%s", ...)``last_status="placeholder"`。参照 `heartbeat_monitor_worker` 先例("本期先不接推送,用终端打印代替")。
- 未来 `JPushNotifier` / `SmsNotifier`:实现同协议即可替换,worker 不改。
- 选择:`INACTIVITY_NOTIFY_CHANNEL` → 工厂返回对应实现(未配到真实实现时回退 `LogNotifier`)。
- 预警文案数据来自快照 `balances`,满足 R3"告知 xx 金币 xx 现金"。
---
## 8. 配置项(`app/core/config.py`
```
INACTIVITY_RESET_ENABLED = False # 总闸,默认关;灰度验证后再开
INACTIVITY_RESET_DAYS = 15 # 不活跃阈值(天)
INACTIVITY_WARN_DAYS_BEFORE = "7,2" # 清零前几天各推一次;空串=不推。逗号分隔,降序解析
INACTIVITY_RESET_RUN_HOUR = 3 # 北京时间每日执行点(0-23)
INACTIVITY_NOTIFY_CHANNEL = "log" # log(占位) / jpush / sms
INACTIVITY_RESET_CHECK_INTERVAL_SEC = 1800 # worker 唤醒间隔(可复用现有间隔常量)
```
- 清零范围(三桶)固定为常量,不做配置。
- `INACTIVITY_WARN_DAYS_BEFORE` 语义(`inactive_days` 为北京自然日,见 §4):档位 `k` ⟹ 当 `inactive_days >= RESET_DAYS-k``< RESET_DAYS` 且本 streak 未推过档 `k` 时预警,即在北京日 `last_active_date + (RESET_DAYSk)` 触发(漏跑某天时补发最紧急未推档,§9)。
- `INACTIVITY_RESET_RUN_HOUR` 只决定 worker 每日执行点,**不改变**"第 16 日 0 点"这一资格边界(§4/§6)。
---
## 9. 幂等与重新活跃
- **重新活跃自动退出**`inactive_days` 由 §4 口径**实时算**。用户一有 `home_view`/比价/领券(**登录本身不算**),`last_active` 前移,自动移出预警与清零队列。**无需**显式"重置标记"。
- **预警去重**`inactivity_notification_log` 中存在 `stage==k 且 created_at > last_active` 的行 ⟹ 本 streak 已推过档 `k`,不重推。用户回归后 `last_active` 前移,旧预警行自然"失效",开启新 streak。
- **清零幂等**:阶段 B 只处理三桶非全 0 者;清完 = 0,次日不再匹配。worker 重启 / 多次唤醒 / 补跑均安全,不产生重复清零或重复流水。
- **稳健补发**:worker 漏跑数天后,某用户可能同时满足多档;只补发**最紧急的未推档**(最小 `k`),避免一次刷屏。
---
## 10. 边界与安全
| 场景 | 处理 |
|---|---|
| 新用户 | `created_at` 作活跃基线(恒非空)→ 注册即"第 1 日活跃";注册后连续 15 天无 home_view/比价/领券 才清 |
| 在途提现 | 提现申请时现金已扣入 `WithdrawOrder`,当前余额已不含在途;只清当前余额、不动提现单。提现失败退款到已清账户 = 用户的钱,正常 |
| 与 `daily_auto_exchange` 并存 | 各自逐用户幂等;金币多已日结折现金,三桶全清正好覆盖 |
| 时区/日界 | 统一北京(`rewards.cn_today()`/`CN_TZ`);**清零/预警按北京自然日 0 点对齐**(末次活跃记为第 1 日 → 第 16 日 0 点清零,见 §4),非滚动 24h;流水 `created_at` 沿用北京 wall-clock naive |
| 误清防护 | 总闸默认关;先灰度只看预警/清零**名单**对不对,再开清零(§13) |
---
## 11. 前端依赖:`home_view` 埋点(跨仓 — Android
- **Android 端**`shaguabijia-app-android`)需在**首页可见**`onResume`/Tab 切入)时,向现有 `POST /api/v1/analytics/events` 批量上报里加一条 `event=<首页可见事件名>`(名称明天加埋点时定,暂记 `"home_view"` 的事件,**携带登录后的 `user_id`**。
- 客户端按会话/前台去重即可(服务端只取 `max(created_at)`,多报无害)。
- **上线顺序依赖**`home_view` 全量覆盖前,"进首页"信号缺失,只有比价/领券能推进活跃、其余落到 `created_at` 基线("只开首页不操作"且注册满 15 天的用户会被误清)—— 故清零总闸必须待 `home_view` 铺满后再开(§13)。
---
## 12. admin 重构范围与影响(R4
- `app/admin/repositories/queries.py`:删本地 `_ACTIVE_EVENTS`/`_last_active_parts()`,改用 `activity.py` 的常量与子查询构造;`list_users``greatest(...)` 排序/筛选、`_attach_last_active` 均改走共享构造器。
- `app/admin/repositories/stats.py``COMPARE_START_EVENT`/`COUPON_START_EVENT`/活跃用户集(`:138-146`)改用共享常量与口径。
- **行为变化(预期内、需产品知会)**admin 的"最近活跃 / DAU"口径变化——**移除 `last_login_at`(登录不再计为活跃)、以 `created_at` 为基线、纳入 `home_view`**。net`home_view` 铺满后更准(真正把"开首页"算进活跃);铺满前"只登录不操作"的用户活跃度会下降。
- **回归底线**:现有 admin 用户列表 / stats 测试按新口径**更新预期**last_login_at 移除 + created_at 基线 + home_view 纳入);非活跃口径部分行为不变。
---
## 13. 灰度与上线顺序(安全优先)
1. **后端先行**:合入共享模块 + 两表 + worker + 通知器,`INACTIVITY_RESET_ENABLED=False`;活跃口径以 `created_at` 为非空基线、**不含 last_login_at**。
2. **Android 发版**:上报 `home_view`;观察 analytics 覆盖率。
3. **只读灰度**:临时开一个"dry-run/只出名单不清零"路径或看阶段 A 预警日志,核对预警/清零**名单**准确。
4. **开清零**:确认无误后置 `INACTIVITY_RESET_ENABLED=True`
5. **收尾/监控**:持续观察 `home_view` 覆盖率与预警/清零名单;发现"活跃却被判不活跃"的漏报即回查埋点覆盖(口径已不含 last_login_at,登录不再兜底)。
---
## 14. 测试计划
- **活跃口径(共享模块)**`home_view`/比价/领券 各单独命中都算活跃;**纯登录不算**;无信号用户以 `created_at` 计;`max` 取最新;naive/aware 混算不崩。
- **admin 回归**:用户列表 / stats 按新口径更新预期(移除 last_login_at + created_at 基线 + home_view)。
- **不活跃判定**`last_active` 分别 `<15d / =15d / >15d` × 有/无余额 的命中矩阵。
- **清零**:三桶归零;`inactivity_reset_log` 清前值正确;三条流水 `biz_type=inactivity_reset``balance_after=0``ref_id=log.id``total_coin_earned` 不变。
- **预警**:命中窗口调 notifier + 写 `notification_log`;同 streak 不重推;回归后 `last_active` 前移可再次预警;漏跑补发最紧急档。
- **worker**:总闸关不启动;文件锁互斥;逐用户失败隔离(一个抛错不影响其余,`failed` 计数);重复跑幂等。
- **配置**`INACTIVITY_WARN_DAYS_BEFORE` 解析(含空串=不推);`RESET_DAYS`/`RUN_HOUR` 生效。
- 沿用 `tests/conftest.py`(临时 SQLite、`RATE_LIMIT_ENABLED=false`);外部通知 monkeypatch。
---
## 15. 未来工作
- 接真实 `JPushNotifier`(需用户级 `registration_id` 覆盖 + JPush push API/ `SmsNotifier`
- 如需 admin 后台可视化:不活跃/预警/清零名单与历史查询接口。
- 如量级增长导致每日 join 扫描变慢:再考虑物化 `last_active_at`(当前每日一次可接受)。
---
## 附:涉及文件清单
**新增**
- `app/repositories/activity.py` — 活跃口径唯一真源
- `app/models/inactivity_reset_log.py` — 审计表
- `app/models/inactivity_notification_log.py` — 预警/占位表
- `app/core/inactivity_reset_worker.py` — 每日 worker(仿 daily_exchange_worker
- `app/integrations/notifier.py` — 通知器协议 + `LogNotifier`(真实 JPush/短信后续同层扩展)
- `alembic/versions/<...>_add_inactivity_tables.py` — 建两表迁移
- `docs/database/inactivity_reset_log.md` / `inactivity_notification_log.md` — 表字典(随实现补)
- 对应 `tests/test_inactivity_reset.py`
**改动**
- `app/models/__init__.py` — 注册两模型
- `app/core/config.py``INACTIVITY_*` 配置
- `app/main.py` — lifespan 接线 start/stop worker
- `app/admin/repositories/queries.py``stats.py` — 改用 `activity.py`(§12
+192
View File
@@ -0,0 +1,192 @@
"""15 天不活跃清零:模型 / 活跃口径 / 清零 / 预警 / 配置 / worker。"""
from __future__ import annotations
from datetime import date, datetime, timedelta, timezone
import pytest
from sqlalchemy import delete, select, update
from app.db.session import SessionLocal
from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
from app.repositories import activity
def test_reset_and_notification_models_persist() -> None:
db = SessionLocal()
try:
db.add(InactivityResetLog(
user_id=1, coin_balance_before=10, cash_balance_cents_before=20,
invite_cash_balance_cents_before=30,
last_active_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
inactive_days=15, reason="inactive_15d",
))
db.add(InactivityNotificationLog(
user_id=1, stage=7, inactive_days=8, coin_balance=10,
cash_balance_cents=20, invite_cash_balance_cents=30,
channel="log", status="placeholder",
))
db.commit()
r = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == 1)).scalar_one()
assert r.reason == "inactive_15d" and r.reset_at is not None
n = db.execute(select(InactivityNotificationLog).where(InactivityNotificationLog.user_id == 1)).scalar_one()
assert n.stage == 7 and n.created_at is not None
finally:
db.rollback()
db.close()
def test_reset_cutoff_is_cn_midnight_of_today_minus_days_minus_1() -> None:
# RESET_DAYS=15, today=1/20 → cutoff = 北京 00:00 of 1/6 = 1/5 16:00 UTC
cutoff = activity.reset_cutoff(15, today=date(2026, 1, 20))
assert cutoff == datetime(2026, 1, 5, 16, 0, tzinfo=timezone.utc)
def test_active_event_constants() -> None:
assert activity.HOME_VIEW_EVENT in activity.ACTIVE_EVENTS
assert "real_compare_start" in activity.ACTIVE_EVENTS
assert "real_coupon_start" in activity.ACTIVE_EVENTS
assert activity.ACTIVE_ENGAGE_TYPE == "claim_started"
def test_as_utc_normalizes() -> None:
assert activity.as_utc(datetime(2026, 1, 1)) == datetime(2026, 1, 1, tzinfo=timezone.utc)
cn = datetime(2026, 1, 1, tzinfo=activity.CN_TZ) # 北京 0 点 = 前一天 16:00 UTC
assert activity.as_utc(cn) == datetime(2025, 12, 31, 16, 0, tzinfo=timezone.utc)
from app.core.rewards import CN_TZ
from app.models.analytics_event import AnalyticsEvent
from app.models.coupon_state import CouponPromptEngagement
from app.models.user import User
from app.models.wallet import CoinAccount
from app.repositories import wallet as wallet_repo
_PHONE_SEQ = [0]
@pytest.fixture(autouse=True)
def _isolate_inactivity_state():
"""本文件的测试都做全表扫描 + 全局计数,而 SQLite 测试库 session 级共享、无逐用例回滚
(commit 后的 rollback no-op),故先把可能泄漏的余额清零 + 清掉活跃事件/审计行,
保证每个用例干净起步不删 User(零余额用户不会被扫描选中,避免跨文件/外键影响)"""
db = SessionLocal()
try:
db.execute(update(CoinAccount).values(
coin_balance=0, cash_balance_cents=0, invite_cash_balance_cents=0))
for model in (AnalyticsEvent, CouponPromptEngagement,
InactivityResetLog, InactivityNotificationLog):
db.execute(delete(model))
db.commit()
finally:
db.close()
yield
def _new_user(db, *, created_at, coin=0, cash=0, invite=0) -> int:
"""直接建一个 User + CoinAccount,created_at 可控。返回 user_id。"""
_PHONE_SEQ[0] += 1
u = User(phone=f"139{_PHONE_SEQ[0]:08d}", created_at=created_at,
last_login_at=created_at, status="active",
username=f"2{_PHONE_SEQ[0]:010d}")
db.add(u)
db.flush()
acc = wallet_repo.get_or_create_account(db, u.id, commit=False)
acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents = coin, cash, invite
acc.total_coin_earned = coin
db.flush()
return u.id
def _add_event(db, user_id, event, when: datetime) -> None:
db.add(AnalyticsEvent(event=event, device_id="d", user_id=user_id, client_ts=0, created_at=when))
def _add_engage(db, user_id, when: datetime, engage_type="claim_started") -> None:
db.add(CouponPromptEngagement(device_id=f"dev{user_id}", package="p", user_id=user_id,
engage_date=when.date(), engage_type=engage_type, created_at=when))
def test_last_active_expr_takes_max_of_baseline_and_events() -> None:
from sqlalchemy import select
db = SessionLocal()
try:
base = datetime(2026, 1, 1, tzinfo=timezone.utc)
uid = _new_user(db, created_at=base, coin=5)
_add_event(db, uid, "real_compare_start", datetime(2026, 1, 10, tzinfo=timezone.utc))
db.commit()
ev_sub, eng_sub = activity.last_active_subqueries(db)
dialect = db.get_bind().dialect.name
expr = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
stmt = (select(expr).select_from(User)
.outerjoin(ev_sub, ev_sub.c.user_id == User.id)
.outerjoin(eng_sub, eng_sub.c.user_id == User.id)
.where(User.id == uid))
got = activity.norm_utc(db.execute(stmt).scalar_one())
assert got == datetime(2026, 1, 10, tzinfo=timezone.utc) # 事件 > 基线
finally:
db.rollback()
db.close()
def test_inactivity_warn_stages_parsing() -> None:
from app.core.config import Settings
s = Settings(INACTIVITY_WARN_DAYS_BEFORE="7,2", INACTIVITY_RESET_DAYS=15)
assert s.inactivity_warn_stages == [7, 2] # 降序去重
s2 = Settings(INACTIVITY_WARN_DAYS_BEFORE="", INACTIVITY_RESET_DAYS=15)
assert s2.inactivity_warn_stages == [] # 空=不推
s3 = Settings(INACTIVITY_WARN_DAYS_BEFORE="2,20,7,2", INACTIVITY_RESET_DAYS=15)
assert s3.inactivity_warn_stages == [7, 2] # 去重 + 丢弃 >=RESET_DAYS(20)
def test_log_notifier_returns_placeholder(caplog) -> None:
from app.integrations.notifier import LogNotifier, get_notifier
n = get_notifier("log")
assert isinstance(n, LogNotifier) and n.channel == "log"
status = n.warn(user_id=1, coin=10, cash_cents=20, invite_cash_cents=30, stage=7, days_until_reset=8)
assert status == "placeholder"
# 未实现通道回退 LogNotifier(占位)
assert get_notifier("jpush").channel == "log"
def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None:
from sqlalchemy import select
from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction
from app.repositories import inactivity
db = SessionLocal()
try:
today = date(2026, 2, 1)
# 末次活跃 = created_at 基线 = 1/10(距 today 22 天 → 应清)
old = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc),
coin=100, cash=200, invite=300)
# 活跃用户:昨天有 home_view → 不清
fresh = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=50)
_add_event(db, fresh, "home_view", datetime(2026, 1, 31, tzinfo=timezone.utc))
db.commit()
stats = inactivity.run_reset_once(db, reset_days=15, today=today)
assert stats["cleared"] == 1 and stats["failed"] == 0
acc = db.get(CoinAccount, old)
assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (0, 0, 0)
assert acc.total_coin_earned == 100 # 历史累计不动
log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one()
assert (log.coin_balance_before, log.cash_balance_cents_before,
log.invite_cash_balance_cents_before) == (100, 200, 300)
assert log.inactive_days == 22 and log.reason == "inactive_15d"
ct = db.execute(select(CoinTransaction).where(
CoinTransaction.user_id == old, CoinTransaction.biz_type == "inactivity_reset")).scalar_one()
assert ct.amount == -100 and ct.balance_after == 0 and ct.ref_id == str(log.id)
assert db.execute(select(CashTransaction).where(
CashTransaction.user_id == old, CashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -200
assert db.execute(select(InviteCashTransaction).where(
InviteCashTransaction.user_id == old, InviteCashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -300
# 活跃用户不动;再跑一次幂等(已清零 → 不再匹配)
assert db.get(CoinAccount, fresh).coin_balance == 50
assert inactivity.run_reset_once(db, reset_days=15, today=today)["cleared"] == 0
finally:
db.rollback()
db.close()
-232
View File
@@ -1,232 +0,0 @@
"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。
沿用 tests/test_wechat_login.py 风格:HTTP client;微信 codeopenid monkeypatch;
短信走 SMS_MOCK(任意 6 位过)数据变更用"再走一遍 wechat-login 看 openid 落在哪个账号"做行为断言
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth # noqa: F401 (后续测试打桩 verify_and_get_phone 用)
from app.core import security
from app.integrations import wxpay
from app.models.phone_rebind_log import PhoneRebindLog
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
def _sms_occupy(client, phone: str) -> int:
"""用普通短信登录占用一个手机号(register_channel=sms),返回该账号 id。"""
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
return r.json()["user"]["id"]
def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id: str) -> dict:
"""微信登录(新 openid)→ 绑同一手机号 → 返回 phone_occupied 的响应体(含 conflict_ticket)。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo(openid))
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": device_id}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": device_id},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "phone_occupied"
return body
# ===== Task 1: 模型可导入(建表由 conftest 的 create_all 完成) =====
def test_phone_rebind_log_model_importable() -> None:
assert PhoneRebindLog.__tablename__ == "phone_rebind_log"
# ===== Task 2: conflict_ticket 令牌 =====
def test_conflict_ticket_roundtrip() -> None:
token = security.create_conflict_ticket(
openid="oid1", wechat_nickname="", wechat_avatar_url="http://a", phone="13900139000"
)
claims = security.decode_conflict_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
assert claims["phone"] == "13900139000"
def test_conflict_ticket_wrong_type_rejected() -> None:
# bind_ticket 冒充 conflict_ticket → TokenError(typ 不匹配)
bind = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_conflict_ticket(bind)
def test_conflict_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139000"
)
with pytest.raises(security.TokenError):
security.decode_conflict_ticket(token)
# ===== Task 3: 占用响应扩展 =====
def test_phone_occupied_returns_conflict_ticket_and_flags(client, monkeypatch) -> None:
phone = "13900139101"
_sms_occupy(client, phone) # 老账号 X(sms,无微信)
body = _occupy_via_conflict(client, monkeypatch, "openid_occ_101", phone, "devO1")
assert body["conflict_ticket"]
assert body["rebind_available"] is True # 首次,未换绑过
assert body["rebind_blocked_days"] == 0
assert body["occupied_account"]["has_wechat"] is False # X 是 sms 账号
# ===== Task 4: 继续绑定 =====
def test_continue_attaches_wechat_and_logs_into_existing(client, monkeypatch) -> None:
"""X 无微信 → 继续绑定并入 openid + 登入 X;之后同 openid 登录直接命中 X。"""
phone = "13900139201"
x_id = _sms_occupy(client, phone) # X:sms 账号,无微信
body = _occupy_via_conflict(client, monkeypatch, "openid_cont_201", phone, "devC1")
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC1"},
)
assert r.status_code == 200, r.text
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == x_id # 登入的是老账号 X
# openid 现已并入 X:再走 wechat-login 直接命中 X
r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC1"})
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == x_id
def test_continue_when_existing_has_wechat_logs_in_and_discards_openid(client, monkeypatch) -> None:
"""X 已绑别的微信 → 继续绑定只登入 X、丢弃本次 openid(不覆盖)。"""
phone = "13900139202"
# 先建一个已绑微信 O1 的账号 X(微信登录 O1 + 短信绑号)
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o1_202"))
t = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2"}).json()["bind_ticket"]
x = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": t, "phone": phone, "code": "123456", "device_id": "devC2"},
).json()
x_id = x["token"]["user"]["id"]
# 新 openid O2 撞同号 → 占用(has_wechat=True)→ 继续绑定
body = _occupy_via_conflict(client, monkeypatch, "openid_o2_202", phone, "devC2b")
assert body["occupied_account"]["has_wechat"] is True
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC2b"},
)
assert r.status_code == 200, r.text
assert r.json()["token"]["user"]["id"] == x_id # 登入 X
# O2 被丢弃:再走 wechat-login(O2)→ 仍未命中(need_bind_phone)
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o2_202"))
assert client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2b"}
).json()["status"] == "need_bind_phone"
def test_continue_expired_ticket_returns_401(client, monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139209"
)
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": expired, "device_id": "devC3"},
)
assert r.status_code == 401, r.text
# ===== Task 5: 换绑 =====
def test_rebind_creates_new_account_and_binds_openid(client, monkeypatch) -> None:
"""换绑 → 建全新微信账号 Y(≠X)+ openid 落到 Y;老账号 X 被注销(手机号归 Y)。"""
phone = "13900139301"
x_id = _sms_occupy(client, phone)
body = _occupy_via_conflict(client, monkeypatch, "openid_rb_301", phone, "devR1")
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR1"},
)
assert r.status_code == 200, r.text
y = r.json()["token"]["user"]
assert r.json()["status"] == "logged_in"
assert y["phone"] == phone
assert y["register_channel"] == "wechat"
assert y["id"] != x_id # 是全新账号,不是老账号
# openid 落到 Y:再走 wechat-login 命中 Y
r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devR1"})
assert r.json()["status"] == "logged_in"
assert r.json()["token"]["user"]["id"] == y["id"]
# ===== §10: continue 路径也应用展示身份回填规则 =====
def test_continue_applies_section10(client, monkeypatch) -> None:
"""§10 via M2 attach 路径: X 是默认昵称+null头像的 sms 账号;
continue 绑定微信后,展示昵称/头像应被微信值替换,并体现在响应的 token.user """
phone = "13900139211"
_sms_occupy(client, phone) # 建 X:默认昵称, null avatar
body = _occupy_via_conflict(
client, monkeypatch, "openid_s10", phone, "devS10"
) # fake_userinfo 默认 nickname="微信昵称", avatar="http://x/a.png"
r = client.post(
"/api/v1/auth/wechat/conflict/continue",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devS10"},
)
assert r.status_code == 200, r.text
user_out = r.json()["token"]["user"]
assert user_out["nickname"] == "微信昵称"
assert user_out["avatar_url"] == "http://x/a.png"
def test_rebind_blocked_within_30_days(client, monkeypatch) -> None:
"""同一手机号 30 天内二次换绑 → 409;占用响应 rebind_available=False。"""
phone = "13900139302"
_sms_occupy(client, phone)
body = _occupy_via_conflict(client, monkeypatch, "openid_rb_302a", phone, "devR2")
assert client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR2"},
).status_code == 200
# 第二次:新 openid 撞同号 → 占用响应此时 rebind_available=False
body2 = _occupy_via_conflict(client, monkeypatch, "openid_rb_302b", phone, "devR2b")
assert body2["rebind_available"] is False
assert body2["rebind_blocked_days"] >= 1
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": body2["conflict_ticket"], "device_id": "devR2b"},
)
assert r.status_code == 409, r.text
def test_rebind_expired_ticket_returns_401(client, monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_conflict_ticket(
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139309"
)
r = client.post(
"/api/v1/auth/wechat/conflict/rebind",
json={"conflict_ticket": expired, "device_id": "devR3"},
)
assert r.status_code == 401, r.text
-198
View File
@@ -1,198 +0,0 @@
"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
bind-phone(建号/占用/令牌过期)
沿用 tests/test_auth.py 风格:HTTP client fixture;微信 codeopenid monkeypatch
拦掉(conftest WECHAT_APP_ID/SECRET dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth
from app.core import security
from app.integrations import wxpay
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
"""返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
# ===== Task 1: bind_ticket 令牌 =====
def test_bind_ticket_roundtrip() -> None:
token = security.create_bind_ticket(openid="oid1", wechat_nickname="", wechat_avatar_url="http://a")
claims = security.decode_bind_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
def test_bind_ticket_wrong_type_rejected() -> None:
# 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
access, _ = security.create_token(user_id=1, token_type="access")
with pytest.raises(security.TokenError):
security.decode_bind_ticket(access)
def test_bind_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_bind_ticket(token)
# ===== Task 2: wechat-login =====
def test_wechat_login_new_openid_returns_bind_ticket(client, monkeypatch) -> None:
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_new_1", "小明", "http://x/m.png"))
r = client.post("/api/v1/auth/wechat-login", json={"code": "wxcode1", "device_id": "devA"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "need_bind_phone"
assert body["bind_ticket"]
assert body["wechat_nickname"] == "小明"
assert body["wechat_avatar_url"] == "http://x/m.png"
assert body["token"] is None
def test_wechat_login_invalid_code_returns_400(client, monkeypatch) -> None:
def _raise(code: str) -> dict:
raise ValueError("微信授权失败: invalid code")
monkeypatch.setattr(wxpay, "code_to_userinfo", _raise)
r = client.post("/api/v1/auth/wechat-login", json={"code": "bad", "device_id": "devA"})
assert r.status_code == 400, r.text
# ===== Task 3: bind-phone/sms =====
def test_wechat_bind_sms_creates_account_then_openid_logs_in(client, monkeypatch) -> None:
"""未占用 → 建微信账号(channel=wechat,昵称头像取微信);再次同 openid 登录 → 直接登入同一账号。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_flow_2", "阿花", "http://x/h.png"))
phone = "13900139002"
# 1) 微信登录 → 未命中 → 拿 ticket
r = client.post("/api/v1/auth/wechat-login", json={"code": "c1", "device_id": "devB"})
ticket = r.json()["bind_ticket"]
assert ticket
# 2) 短信绑号(SMS_MOCK:任意 6 位通过)→ 建号 + 登入
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devB"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "阿花"
assert user["avatar_url"] == "http://x/h.png"
uid = user["id"]
# 3) 再次微信登录(同 openid)→ 命中 → 直接登入同一账号
r = client.post("/api/v1/auth/wechat-login", json={"code": "c2", "device_id": "devB"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
assert body["token"]["user"]["id"] == uid
def test_wechat_bind_sms_phone_occupied(client, monkeypatch) -> None:
"""手机号已被其他账号占用 → 返回 phone_occupied + 原账号信息(不建号)。"""
phone = "13900139003"
# 先用普通短信登录占用该手机号(register_channel=sms)
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
occupied_nickname = r.json()["user"]["nickname"]
# 微信登录(新 openid)→ 未命中 → ticket
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_occ_3"))
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC"}
).json()["bind_ticket"]
# 绑同一手机号 → 占用
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devC"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "phone_occupied"
assert body["token"] is None
assert body["occupied_account"]["nickname"] == occupied_nickname
assert body["occupied_account"]["avatar_url"] is None # 短信注册账号无头像 → 序列化为 null
assert body["occupied_account"]["created_at"]
def test_wechat_bind_sms_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": expired, "phone": "13900139009", "code": "123456", "device_id": "devD"},
)
assert r.status_code == 401, r.text
# ===== Task 4: bind-phone/jverify =====
def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
"""本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
phone = "13900139005"
# 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "极光用户"
# 微信 userinfo 隐私脱敏 avatar=None → 头像为空(客户端兜底默认头像)
assert user["avatar_url"] is None
def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401(极光绑号路径,decode 先于极光核验)。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_jv_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": expired, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 401, r.text
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
"""极光核验失败(JiguangError)→ 502。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
def _raise(token: str) -> str:
raise auth.JiguangError("mock jg failure")
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "badtoken", "device_id": "devE"},
)
assert r.status_code == 502, r.text
-88
View File
@@ -312,94 +312,6 @@ def test_bind_rejects_openid_already_bound(client, monkeypatch) -> None:
assert r.status_code == 409, r.text
# ===== §10 绑微信时展示身份回填规则 =====
def test_bind_replaces_default_nickname_and_null_avatar(client, monkeypatch) -> None:
"""§10-A: 全默认(昵称=系统默认,头像=null) → 绑微信后两个展示字段都替换为微信值。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_a", "nickname": "微信昵称A", "avatar_url": "https://x/a.png", "raw": {}},
)
token = _login(client, "13800003001")
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
assert r.status_code == 200, r.text
body = r.json()
assert body["nickname"] == "微信昵称A"
assert body["avatar_url"] == "https://x/a.png"
def test_bind_keeps_customized_nickname(client, monkeypatch) -> None:
"""§10-B: 用户已改过昵称 → 绑微信后昵称保留,但 null 头像仍替换为微信头像。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_b", "nickname": "微信昵称B", "avatar_url": "https://x/b.png", "raw": {}},
)
token = _login(client, "13800003002")
# 先把昵称改成非默认值
r = client.patch(
"/api/v1/user/profile", json={"nickname": "我的名字"}, headers=_auth(token)
)
assert r.status_code == 200, r.text
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == "我的名字" # 保留自定义昵称
assert body["avatar_url"] == "https://x/b.png" # null 头像被微信头像替换
def test_bind_keeps_customized_avatar(client, monkeypatch) -> None:
"""§10-C: 已有自定义头像 → 绑微信后头像保留,但默认昵称替换为微信昵称。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_c", "nickname": "微信昵称C", "avatar_url": "https://x/c.png", "raw": {}},
)
token = _login(client, "13800003003")
phone = "13800003003"
# 直接用 DB 给用户写入自定义头像(保持默认昵称)
db = SessionLocal()
try:
user = db.execute(select(User).where(User.phone == phone)).scalar_one()
user.avatar_url = "https://custom/av.png"
db.commit()
finally:
db.close()
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == "微信昵称C" # 默认昵称被替换
assert body["avatar_url"] == "https://custom/av.png" # 自定义头像保留
def test_bind_wechat_empty_keeps_default(client, monkeypatch) -> None:
"""§10-D: 微信侧 nickname/avatar 均为 None → 不覆盖,展示字段维持原默认值。"""
monkeypatch.setattr(
"app.integrations.wxpay.code_to_userinfo",
lambda code: {"openid": "openid_s10_d", "nickname": None, "avatar_url": None, "raw": {}},
)
token = _login(client, "13800003004")
r = client.get("/api/v1/auth/me", headers=_auth(token))
original_nickname = r.json()["nickname"] # 系统分配的默认昵称
assert original_nickname.startswith("用户")
r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
assert r.status_code == 200, r.text
r = client.get("/api/v1/auth/me", headers=_auth(token))
body = r.json()
assert body["nickname"] == original_nickname # 默认昵称不变
assert body["avatar_url"] is None # 头像仍为 null
def test_withdraw_reject_refunds(client, monkeypatch) -> None:
"""管理员审核拒绝 → 退回现金 + 单 rejected + 理由写入 fail_reason(用户可见)。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_reject", "nickname": None, "avatar_url": None, "raw": {}})