Unified_Authentication_Platform/backend/app/api/v1/endpoints/sms_auth.py

from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Body, Request
from sqlalchemy.orm import Session
from datetime import timedelta

from app.api.v1 import deps
from app.core import security
from app.core.config import settings
from app.core.utils import get_client_ip
from app.core.cache import redis_client
from app.models.user import User, UserStatus
from app.services.sms_service import SmsService
from app.services.system_config_service import SystemConfigService
from app.services.login_log_service import LoginLogService
from app.models.login_log import LoginMethod, AuthType
from app.schemas.login_log import LoginLogCreate
from app.schemas.token import Token

router = APIRouter()

@router.post("/send-code", summary="发送短信验证码")
def send_sms_code(
    mobile: str = Body(..., embed=True),
    platform: str = Body(..., embed=True, description="pc or mobile"),
    db: Session = Depends(deps.get_db),
) -> Any:
    """
    发送短信验证码。
    需检查系统配置是否允许对应平台的短信登录。
    """
    # Check config
    config_key = "sms_login_pc_enabled" if platform == "pc" else "sms_login_mobile_enabled"
    is_enabled = SystemConfigService.get_config(db, config_key)
    
    if is_enabled != "true":
        raise HTTPException(status_code=403, detail="当前平台未开启短信登录功能")

    # Send code
    try:
        SmsService.send_code(mobile)
    except Exception as e:
        # Pass through HTTPExceptions from service
        if isinstance(e, HTTPException):
            raise e
        # Log unexpected errors
        print(f"Error sending SMS: {e}")
        raise HTTPException(status_code=500, detail="发送验证码失败")

    return {"message": "验证码已发送"}

@router.post("/login", response_model=Token, summary="短信验证码登录")
def login_with_sms(
    request: Request,
    mobile: str = Body(..., embed=True),
    code: str = Body(..., embed=True),
    platform: str = Body("pc", embed=True),
    db: Session = Depends(deps.get_db),
) -> Any:
    """
    使用手机号和验证码登录。
    """
    # 1. Check Config (Double check)
    config_key = "sms_login_pc_enabled" if platform == "pc" else "sms_login_mobile_enabled"
    is_enabled = SystemConfigService.get_config(db, config_key)
    
    if is_enabled != "true":
        raise HTTPException(status_code=403, detail="当前平台未开启短信登录功能")

    # 2. Verify Code
    # Redis key from SmsService: SMS:{mobile}
    key = f"SMS:{mobile}"
    stored_code = redis_client.get(key)
    
    if not stored_code or stored_code != code:
         raise HTTPException(status_code=400, detail="验证码错误或已过期")

    # 3. Find User
    user = db.query(User).filter(User.mobile == mobile, User.is_deleted == 0).first()
    
    # Log preparation
    log_create = LoginLogCreate(
        mobile=mobile,
        ip_address=get_client_ip(request),
        login_method=LoginMethod.UNIFIED_PAGE,
        auth_type=AuthType.SMS,
        user_agent=request.headers.get("user-agent")
    )
    
    if not user:
        log_create.is_success = 0
        log_create.failure_reason = "用户不存在"
        LoginLogService.create_log(db, log_create)
        raise HTTPException(status_code=404, detail="用户不存在")

    log_create.user_id = user.id

    if user.status != UserStatus.ACTIVE:
        log_create.is_success = 0
        log_create.failure_reason = "用户已禁用"
        LoginLogService.create_log(db, log_create)
        raise HTTPException(status_code=400, detail="用户已禁用")

    # 4. Generate Token
    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = security.create_access_token(
        subject=user.id,
        expires_delta=access_token_expires
    )
    
    # Clear Code
    redis_client.delete(key)
    
    # Log Success
    LoginLogService.create_log(db, log_create)

    return {
        "access_token": access_token,
        "token_type": "bearer",
    }