DadaoTask/backend/app/api/auth.py

"""认证：手机号 + 验证码。M0 不做密码、不做第三方登录。"""

from __future__ import annotations

import hmac
import secrets

from fastapi import APIRouter
from pydantic import BaseModel, Field
from sqlalchemy import select

from ..config import get_settings
from ..deps import CurrentUser, SessionDep
from ..errors import AppError, ErrorCode
from ..models import AppUser, Member, Workspace
from ..schemas import ok
from ..security import create_token_pair, decode_token
from ..stores import get_store, sms_code_key, sms_throttle_key

router = APIRouter(prefix="/auth", tags=["认证"])
settings = get_settings()

PHONE_PATTERN = r"^1[3-9]\d{9}$"


def mask_phone(phone: str) -> str:
    return f"{phone[:3]}****{phone[-4:]}" if len(phone) == 11 else "***"


class SendCodeIn(BaseModel):
    phone: str = Field(pattern=PHONE_PATTERN, examples=["13800138000"])


class LoginIn(BaseModel):
    phone: str = Field(pattern=PHONE_PATTERN)
    code: str = Field(min_length=4, max_length=8)


class RefreshIn(BaseModel):
    refresh_token: str = Field(min_length=10)


@router.post("/sms/send", summary="发送登录验证码")
async def send_code(body: SendCodeIn) -> dict:
    store = get_store()
    if settings.sms_throttle_seconds > 0:
        hits = await store.incr(sms_throttle_key(body.phone), settings.sms_throttle_seconds)
        if hits > 1:
            raise AppError(
                ErrorCode.SMS_TOO_FREQUENT,
                f"发送过于频繁，请 {settings.sms_throttle_seconds} 秒后重试",
            )

    if settings.sms_provider == "dev":
        code = settings.sms_dev_code
    else:
        code = f"{secrets.randbelow(1_000_000):06d}"
        # TODO(M0 遗留)：接入阿里云/腾讯云短信；当前直接在响应里不回显验证码
        raise AppError(ErrorCode.INTERNAL, "短信通道尚未接入", http_status=501)

    await store.set(sms_code_key(body.phone), code, settings.sms_code_ttl_seconds)
    return ok({"sent": True, "expires_in": settings.sms_code_ttl_seconds, "dev_code": code})


@router.post("/login", summary="验证码登录（不存在则自动注册）")
async def login(body: LoginIn, session: SessionDep) -> dict:
    store = get_store()
    stored = await store.get(sms_code_key(body.phone))
    if stored is None:
        raise AppError(ErrorCode.SMS_CODE_INVALID, "验证码已过期，请重新获取")
    if not hmac.compare_digest(stored, body.code):
        raise AppError(ErrorCode.SMS_CODE_INVALID, "验证码错误")
    await store.delete(sms_code_key(body.phone))

    user = (
        await session.execute(select(AppUser).where(AppUser.phone == body.phone))
    ).scalar_one_or_none()
    if user is None:
        user = AppUser(phone=body.phone, nickname=f"用户{body.phone[-4:]}")
        session.add(user)
        await session.flush()

    tokens = create_token_pair(user.id)
    return ok(
        {
            "user": {
                "id": str(user.id),
                "phone": mask_phone(user.phone),
                "nickname": user.nickname,
                "created_at": user.created_at,
            },
            "tokens": {
                "access_token": tokens.access_token,
                "refresh_token": tokens.refresh_token,
                "token_type": "bearer",
                "expires_in": tokens.expires_in,
            },
        }
    )


@router.post("/refresh", summary="刷新访问令牌")
async def refresh(body: RefreshIn) -> dict:
    user_id = decode_token(body.refresh_token, "refresh")
    tokens = create_token_pair(user_id)
    return ok(
        {
            "access_token": tokens.access_token,
            "refresh_token": tokens.refresh_token,
            "token_type": "bearer",
            "expires_in": tokens.expires_in,
        }
    )


@router.get("/me", summary="当前用户与所属工作区")
async def me(user: CurrentUser, session: SessionDep) -> dict:
    rows = (
        await session.execute(
            select(Workspace, Member.role)
            .join(Member, Member.workspace_id == Workspace.id)
            .where(Member.user_id == user.id, Member.status == "active")
            .order_by(Workspace.created_at.desc())
        )
    ).all()

    return ok(
        {
            "user": {
                "id": str(user.id),
                "phone": mask_phone(user.phone),
                "nickname": user.nickname,
            },
            "workspaces": [
                {
                    "id": str(ws.id),
                    "name": ws.name,
                    "slug": ws.slug,
                    "plan": ws.plan,
                    "role": role,
                    "created_at": ws.created_at,
                }
                for ws, role in rows
            ],
        }
    )