"""Password, development, and DingTalk authentication endpoints.""" from datetime import UTC, datetime from typing import Annotated from fastapi import APIRouter, Depends from app.api.dependencies import DatabaseSession, get_current_user from app.core.config import get_settings from app.core.exceptions import AppException from app.core.security import create_access_token, hash_password from app.models.entities import OperationAudit, UserAccount from app.repositories.users import UserRepository from app.schemas.auth import DingTalkConfigData, DingTalkLoginRequest, LoginRequest, TokenData, UserData from app.schemas.common import ApiResponse from app.services.auth import AuthService from app.services.dingtalk_auth import DingTalkAuthService router = APIRouter(prefix="/auth", tags=["auth"]) @router.post("/login", response_model=ApiResponse[TokenData]) async def login(request: LoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]: return ApiResponse(data=await AuthService(session).login(request.username, request.password)) @router.get("/me", response_model=ApiResponse[UserData]) async def me( current_user: Annotated[UserAccount, Depends(get_current_user)], session: DatabaseSession, ) -> ApiResponse[UserData]: return ApiResponse(data=await AuthService(session).me(current_user)) @router.get("/dingtalk/config", response_model=ApiResponse[DingTalkConfigData]) async def dingtalk_config() -> ApiResponse[DingTalkConfigData]: return ApiResponse(data=DingTalkAuthService.config()) @router.post("/dingtalk/login", response_model=ApiResponse[TokenData]) async def dingtalk_login(request: DingTalkLoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]: return ApiResponse(data=await DingTalkAuthService(session).login(request.auth_code)) @router.post("/dev-login", response_model=ApiResponse[TokenData]) async def dev_login(request: LoginRequest, session: DatabaseSession) -> ApiResponse[TokenData]: """Development-only login helper used by local frontend verification.""" settings = get_settings() if settings.environment != "development": raise AppException("开发登录入口未启用", code=404, status_code=404) user = await UserRepository(session).get_by_dingtalk_id(f"local-{request.username}") if user is None: user = UserAccount( emp_no=f"dev_{request.username}"[:20], name="开发测试用户", role="admin", status="启用", dingtalk_id=f"local-{request.username}", password_hash=hash_password(request.password), ) session.add(user) await session.flush() user.last_login_at = datetime.now(UTC) session.add(OperationAudit(actor_user_id=user.id, actor_emp_no=user.emp_no, action="auth.dev_login")) return ApiResponse( data=TokenData( access_token=create_access_token(user.id, role=user.role), expires_in=settings.jwt_expire_minutes * 60, user=await AuthService(session).me(user), ) )