from typing import Any from fastapi import APIRouter, Depends, HTTPException, Body from sqlalchemy.orm import Session from pydantic import BaseModel from app.api.v1 import deps from app.core import security from app.core.utils import generate_english_name from app.schemas.user import UserRegister, User as UserSchema from app.models.user import User, UserStatus, UserRole from app.models.application import Application from app.services.sms_service import SmsService from app.services.captcha_service import CaptchaService router = APIRouter() @router.post("/register", response_model=UserSchema, summary="开发者注册") def register_developer( req: UserRegister, db: Session = Depends(deps.get_db), ): """ 开发者公开注册接口。 如果是系统第一个用户,将自动成为超级管理员并激活。 否则,创建状态为 PENDING,角色为 DEVELOPER 的用户。 """ # 0. Verify SMS Code if not SmsService.verify_code(req.mobile, req.sms_code): raise HTTPException(status_code=400, detail="短信验证码无效") # 1. Check if mobile exists existing = db.query(User).filter(User.mobile == req.mobile).first() if existing: raise HTTPException(status_code=400, detail="该手机号已注册") # 2. Check if this is the first user user_count = db.query(User).count() is_first_user = (user_count == 0) role = UserRole.SUPER_ADMIN if is_first_user else UserRole.DEVELOPER status = UserStatus.ACTIVE if is_first_user else UserStatus.PENDING # 3. Create User english_name = generate_english_name(req.name) user = User( mobile=req.mobile, name=req.name, english_name=english_name, password_hash=security.get_password_hash(req.password), status=status, role=role, is_deleted=0 ) db.add(user) db.commit() db.refresh(user) return user class AppPublicInfo(BaseModel): app_id: str app_name: str protocol_type: str redirect_uris: str # JSON string @router.get("/apps/{app_id}", response_model=AppPublicInfo, summary="获取应用公开信息") def get_app_public_info( app_id: str, db: Session = Depends(deps.get_db), ): """ 获取应用的公开信息(协议类型、重定向 URIs)。 用于登录页面的重定向逻辑。 """ app = db.query(Application).filter(Application.app_id == app_id).first() if not app: raise HTTPException(status_code=404, detail="应用未找到") return AppPublicInfo( app_id=app.app_id, app_name=app.app_name, protocol_type=app.protocol_type, redirect_uris=app.redirect_uris or "[]" ) class SmsSendRequest(BaseModel): mobile: str captcha_id: str captcha_code: str class PasswordResetRequest(BaseModel): mobile: str sms_code: str new_password: str @router.post("/sms/send", summary="发送短信验证码") def send_sms( req: SmsSendRequest ): """ 发送短信验证码。需要图形验证码验证。 """ # 1. Verify Captcha if not CaptchaService.verify_captcha(req.captcha_id, req.captcha_code): raise HTTPException(status_code=400, detail="图形验证码无效") # 2. Send SMS SmsService.send_code(req.mobile) return {"message": "短信发送成功"} @router.post("/pwd/reset", summary="重置密码") def reset_password( req: PasswordResetRequest, db: Session = Depends(deps.get_db), ): """ 使用短信验证码重置密码。 """ # 1. Verify SMS Code if not SmsService.verify_code(req.mobile, req.sms_code): raise HTTPException(status_code=400, detail="短信验证码无效") # 2. Find User user = db.query(User).filter(User.mobile == req.mobile).first() if not user: raise HTTPException(status_code=404, detail="用户未找到") # 3. Update Password if not security.validate_password_strength(req.new_password): raise HTTPException(status_code=400, detail="密码强度不足,必须包含字母和数字") user.password_hash = security.get_password_hash(req.new_password) db.add(user) db.commit() return {"message": "密码重置成功"}