左辰勇
139 lines
6.9 KiB
Python
139 lines
6.9 KiB
Python
"""重置指定用户的新手引导完成标记,让这个账号重新进新手引导页,方便反复测试引导流程。
|
|
|
|
原理:是否跳过引导只由 onboarding_completion 表里 (user_id, device_id) 那一行决定
|
|
(见 app/models/onboarding.py)。删掉该用户的行 → 登录响应 onboarding_completed=false、
|
|
GET /api/v1/user/onboarding/status 也返 false → 客户端下次登录/启动重走引导。
|
|
本地 SharedPreferences 标记卸载即丢、以后端为准,所以删这一行就够,不用重装 App。
|
|
|
|
默认删该用户**所有设备**的记录(换机/多设备一起放开);只想放开某一台用 --device-id
|
|
(device_id = 客户端硬件级 ANDROID_ID,与登录 / onboarding/complete 传的是同一个值)。
|
|
|
|
与已有两个入口的分工:
|
|
- admin「设备维度引导管理」按**设备**重置(该设备上所有账号一起),本脚本按**账号**;
|
|
- POST /api/v1/user/onboarding/reset 要客户端自己带 device_id 调,本脚本从库里反查设备。
|
|
|
|
用法(在项目根、已 pip install -e . 的环境里跑):
|
|
python scripts/reset_onboarding.py # 默认测试号 11111111111
|
|
python scripts/reset_onboarding.py 13800138000 # 指定手机号
|
|
python scripts/reset_onboarding.py --user-id 5 # 直接指定 user_id
|
|
python scripts/reset_onboarding.py --dry-run # 预览(照常执行再回滚),不落库
|
|
python scripts/reset_onboarding.py --device-id abc123 # 只放开这一台设备,其余设备照旧跳过
|
|
|
|
走 SessionLocal 连 DATABASE_URL(SQLite / Postgres 都行),因此**默认只允许 APP_ENV=dev 改库**
|
|
(--dry-run 只读,任何环境都能跑)。线上确实要给某个用户开引导时加 --force —— 这张表只存
|
|
"引导走过没"的标记,删了最坏结果是用户多看一次引导,不涉及金额/账目。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
|
|
from sqlalchemy import delete, select
|
|
|
|
from app.core.config import settings
|
|
from app.db.session import SessionLocal, engine
|
|
from app.models.onboarding import OnboardingCompletion
|
|
from app.models.user import User
|
|
|
|
# Windows 控制台默认 GBK,强制 UTF-8 否则中文输出乱码。stderr 也要设:
|
|
# SystemExit(如"用户不存在")的中文提示走的是 stderr。
|
|
for _stream in (sys.stdout, sys.stderr):
|
|
if hasattr(_stream, "reconfigure"):
|
|
_stream.reconfigure(encoding="utf-8")
|
|
|
|
# dev 下 engine 是 echo=True(APP_DEBUG),几十行 SQL 会把前后对比刷没。echo 走 SQLAlchemy 自己的
|
|
# InstanceLogger,不吃 logging.setLevel,只能改 engine.echo。
|
|
engine.echo = False
|
|
|
|
DEFAULT_PHONE = "11111111111"
|
|
|
|
|
|
def resolve_user(db, phone: str, user_id: int | None) -> User:
|
|
if user_id is not None:
|
|
user = db.get(User, user_id)
|
|
if user is None:
|
|
raise SystemExit(f"user_id={user_id} 不存在")
|
|
return user
|
|
user = db.execute(select(User).where(User.phone == phone)).scalar_one_or_none()
|
|
if user is None:
|
|
raise SystemExit(f"手机号 {phone} 没有对应用户(注意 phone 才是登录账号,username 是展示 ID)")
|
|
return user
|
|
|
|
|
|
def print_state(db, user: User, device_id: str | None, label: str) -> None:
|
|
"""打印该用户当前的引导完成标记。--device-id 时只看那一台,便于确认没误伤别的设备。"""
|
|
stmt = (
|
|
select(OnboardingCompletion.device_id, OnboardingCompletion.completed_at)
|
|
.where(OnboardingCompletion.user_id == user.id)
|
|
.order_by(OnboardingCompletion.completed_at.desc())
|
|
)
|
|
if device_id:
|
|
stmt = stmt.where(OnboardingCompletion.device_id == device_id)
|
|
rows = db.execute(stmt).all()
|
|
|
|
print(f"--- {label} ---")
|
|
scope = f"device_id={device_id}" if device_id else "全部设备"
|
|
if not rows:
|
|
print(f" onboarding_completion({scope}): (无) → 该用户会走引导")
|
|
return
|
|
print(f" onboarding_completion({scope}): {len(rows)} 条 → 这些设备上会跳过引导")
|
|
for did, at in rows:
|
|
print(f" device_id={did} 完成于 {at}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="重置指定用户的新手引导,让其重新进引导页")
|
|
parser.add_argument("phone", nargs="?", default=DEFAULT_PHONE,
|
|
help=f"手机号(默认 {DEFAULT_PHONE})")
|
|
parser.add_argument("--user-id", type=int, default=None, help="直接按 user_id 定位,优先于 phone")
|
|
parser.add_argument("--device-id", default=None,
|
|
help="只重置这一台设备(硬件级 ANDROID_ID);默认重置该用户所有设备")
|
|
parser.add_argument("--dry-run", action="store_true", help="预览,最后回滚不落库")
|
|
parser.add_argument("--force", action="store_true", help="非 dev 环境也允许改库(仅删引导标记,不涉及账目)")
|
|
args = parser.parse_args()
|
|
|
|
if not args.dry_run and settings.APP_ENV != "dev" and not args.force:
|
|
raise SystemExit(
|
|
f"APP_ENV={settings.APP_ENV},默认只有 dev 能改库。确认要在该环境重置请加 --force"
|
|
"(--dry-run 只读,任意环境可跑)"
|
|
)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
user = resolve_user(db, args.phone, args.user_id)
|
|
print(f"DB: {settings.DATABASE_URL} APP_ENV: {settings.APP_ENV}")
|
|
print(f"用户: id={user.id} phone={user.phone} username={user.username}")
|
|
print_state(db, user, args.device_id, "before")
|
|
|
|
stmt = delete(OnboardingCompletion).where(OnboardingCompletion.user_id == user.id)
|
|
if args.device_id:
|
|
stmt = stmt.where(OnboardingCompletion.device_id == args.device_id)
|
|
deleted = db.execute(stmt).rowcount
|
|
|
|
if not deleted:
|
|
# 没记录本来就会走引导 —— 常见于:换了新设备、或运营/客户端已经重置过一次。
|
|
print("该用户(该范围内)本来就没有完成标记,已经会走引导了,无需处理。")
|
|
db.rollback()
|
|
else:
|
|
print_state(db, user, args.device_id, "after")
|
|
if args.dry_run:
|
|
db.rollback()
|
|
print(f"(dry-run:以上 after 为预览,已回滚,库没动;真跑会删 {deleted} 条)")
|
|
return
|
|
db.commit()
|
|
print(f"完成:删掉 {deleted} 条完成标记,{user.phone} 下次登录会重走新手引导。")
|
|
|
|
# 测试号无论库里有没有记录都恒走引导(见 app/core/test_account.py),提醒一句免得白跑
|
|
if settings.test_account_phone and user.phone == settings.test_account_phone:
|
|
print(f"提示:{user.phone} 是配置的测试账号(TEST_ACCOUNT_PHONE),"
|
|
"登录响应 onboarding_completed 恒为 false、本就每次都走引导,无需重置。")
|
|
|
|
print("提醒:客户端是在登录响应 / 启动时查 onboarding/status 的,已经在首页的 App 不会自动跳转,"
|
|
"退出登录重进(或杀掉重开)才会看到引导页。")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|