zzhyyyyy
b00d430332
- User 新增 force_onboarding 列(仿 debug_trace_enabled)+ 迁移 user_force_onboarding。
- POST /admin/api/users/{id}/force-onboarding(operator 角色 + 审计 user.force_onboarding.set)。
- UserOut(/me 与登录响应)带出该字段;/api/v1/user/onboarding/complete 走完即自动清回 false,只触发一次。
- 测试:admin 设置+审计+角色守卫、走完引导自动清标记。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
"""user 表的 CRUD 工具函数。
|
|
|
|
业务策略 = "注册即登录":phone 不存在则 insert,存在则更新 last_login_at。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.user import User
|
|
|
|
|
|
def get_user_by_id(db: Session, user_id: int) -> User | None:
|
|
return db.get(User, user_id)
|
|
|
|
|
|
def get_user_by_phone(db: Session, phone: str) -> User | None:
|
|
stmt = select(User).where(User.phone == phone)
|
|
return db.execute(stmt).scalar_one_or_none()
|
|
|
|
|
|
def upsert_user_for_login(
|
|
db: Session,
|
|
*,
|
|
phone: str,
|
|
register_channel: str,
|
|
) -> User:
|
|
"""登录入口统一调:有则更新 last_login_at,无则新建。
|
|
|
|
返回 commit 后的 user。调用方拿到 user.id 直接签 JWT。
|
|
"""
|
|
user = get_user_by_phone(db, phone)
|
|
now = datetime.now(timezone.utc)
|
|
if user is None:
|
|
user = User(
|
|
phone=phone,
|
|
register_channel=register_channel,
|
|
last_login_at=now,
|
|
)
|
|
db.add(user)
|
|
else:
|
|
user.last_login_at = now
|
|
# 被禁用的账号被尝试登录时,不自动启用,直接抛(在调用方处理)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def update_nickname(db: Session, user: User, *, nickname: str) -> User:
|
|
user.nickname = nickname
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def set_avatar_url(db: Session, user: User, *, avatar_url: str) -> User:
|
|
user.avatar_url = avatar_url
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def clear_force_onboarding(db: Session, user: User) -> None:
|
|
"""用户走完(被运营强制开启的)新手引导后,清除强制标记,避免下次启动再被拉回引导。
|
|
|
|
幂等:未置位时直接返回,不空 commit。由 /onboarding/complete 在标记完成后调用。
|
|
"""
|
|
if user.force_onboarding:
|
|
user.force_onboarding = False
|
|
db.commit()
|
|
|
|
|
|
def soft_delete_account(db: Session, user: User) -> None:
|
|
"""注销账号:软删除 + 匿名化。
|
|
|
|
把 phone 改成 `deleted_<id>` 释放唯一约束,允许同号码重新注册成全新账号;
|
|
清空 PII(昵称/头像),保留行用于审计。status=deleted 后将无法再登录该行
|
|
(登录接口校验 status==active)。
|
|
"""
|
|
user.status = "deleted"
|
|
user.phone = f"deleted_{user.id}"
|
|
user.nickname = None
|
|
user.avatar_url = None
|
|
db.commit()
|