from datetime import datetime, timezone from fastapi import APIRouter, Depends from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.dependencies import get_db, get_current_user from app.models.system_config import SystemConfig from app.models.user import User from app.schemas.auth import LoginRequest, ChangePasswordRequest, RegisterRequest from app.schemas.user import UserOut from app.services.auth import ( authenticate_user, create_access_token, decode_access_token, hash_password, verify_password, ) from app.utils.id_gen import generate_id router = APIRouter(prefix="/auth", tags=["auth"]) @router.post("/login") async def login(req: LoginRequest, db: AsyncSession = Depends(get_db)): # Validate captcha in production (when SMS_MOCK is false) if not settings.SMS_MOCK: if not req.captcha_token: from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="需要验证码", ) token_sub = decode_access_token(req.captcha_token) if not token_sub or not token_sub.startswith("captcha:"): from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="验证码无效或已过期", ) user = await authenticate_user(db, req.username, req.password) if not user: from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误", ) # Only allow frontend users to login via this endpoint if user.user_type != "frontend": from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="该账号不允许在此登录", ) user.last_login_at = datetime.now() user.credits = round(user.credits, 2) await db.flush() token = create_access_token(user.id, req.remember_me) return {"access_token": token, "token_type": "bearer", "user": UserOut.model_validate(user)} @router.post("/register") async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)): """Register a new user with phone + SMS code + password.""" # Verify SMS code from app.services.sms import verify_sms_code ok = await verify_sms_code(req.phone, req.code) if not ok: from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期", ) # Check if phone already registered existing = await db.execute(select(User).where(User.phone == req.phone)) if existing.scalar_one_or_none(): from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="该手机号已注册", ) # Create user with default username: 用户 + last 4 digits of phone import random username = f"用户{req.phone[-4:]}" existing_name = await db.execute(select(User).where(User.username == username)) if existing_name.scalar_one_or_none(): username = f"用户{req.phone[-4:]}{random.randint(10, 99)}" user = User( id=generate_id(), username=username, phone=req.phone, hashed_password=hash_password(req.password), credits=100, is_admin=False, user_type="frontend", ) db.add(user) await db.flush() # Assign default menus to new user from app.models.menu_config import MenuConfig result = await db.execute( select(MenuConfig).where( MenuConfig.is_default == True, MenuConfig.is_active == True, MenuConfig.menu_target.in_(["frontend", "both"]), MenuConfig.menu_type == "page", ) ) default_menus = result.scalars().all() if default_menus: user.allowed_menus = [m.path for m in default_menus if m.path] user.credits = round(user.credits, 2) token = create_access_token(user.id) return {"access_token": token, "token_type": "bearer", "user": UserOut.model_validate(user)} @router.post("/logout") async def logout(current_user: User = Depends(get_current_user)): return {"message": "ok"} @router.get("/me", response_model=UserOut) async def get_me(current_user: User = Depends(get_current_user)): current_user.credits = round(current_user.credits, 2) return current_user @router.post("/change-password") async def change_password( req: ChangePasswordRequest, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not verify_password(req.old_password, current_user.hashed_password): from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="原密码错误", ) if len(req.new_password) < 6: from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="密码至少6位", ) current_user.hashed_password = hash_password(req.new_password) await db.flush() return {"message": "密码修改成功"} @router.get("/site-info") async def get_site_info(db: AsyncSession = Depends(get_db)): """Public endpoint returning site name and logo.""" result = await db.execute( select(SystemConfig).where(SystemConfig.key.in_([ "site_name", "site_logo", "user_agreement_url", "privacy_policy_url" ])) ) configs = result.scalars().all() info = {c.key: c.value for c in configs} return { "site_name": info.get("site_name", "VideoGen.AI"), "site_logo": info.get("site_logo", ""), "user_agreement_url": info.get("user_agreement_url", ""), "privacy_policy_url": info.get("privacy_policy_url", ""), } @router.post("/admin-login") async def admin_login(req: LoginRequest, db: AsyncSession = Depends(get_db)): """Admin-only login endpoint.""" user = await authenticate_user(db, req.username, req.password) if not user: from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误", ) if user.user_type != "admin": from fastapi import HTTPException, status raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="该账号不是管理员账号", ) user.last_login_at = datetime.now() user.credits = round(user.credits, 2) await db.flush() token = create_access_token(user.id, req.remember_me) return {"access_token": token, "token_type": "bearer", "user": UserOut.model_validate(user)}