""" 회원가입 요청 및 관리자 승인 API """ from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from sqlalchemy import text from pydantic import BaseModel from typing import Optional, List from datetime import datetime from ..database import get_db from .auth_service import AuthService from .models import UserRepository from .middleware import get_current_user from ..utils.logger import get_logger logger = get_logger(__name__) router = APIRouter(prefix="/auth", tags=["signup"]) class SignupRequest(BaseModel): username: str password: str name: str email: Optional[str] = None department: Optional[str] = None position: Optional[str] = None phone: Optional[str] = None reason: Optional[str] = None # 가입 사유 @router.post("/signup-request") async def signup_request( signup_data: SignupRequest, db: Session = Depends(get_db) ): """ 회원가입 요청 (관리자 승인 대기) Args: signup_data: 가입 신청 정보 Returns: dict: 요청 결과 """ try: # 중복 사용자명 확인 user_repo = UserRepository(db) existing_user = user_repo.find_by_username(signup_data.username) if existing_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="이미 존재하는 사용자명입니다" ) # 중복 이메일 확인 if signup_data.email: check_email = text("SELECT id FROM users WHERE email = :email") existing_email = db.execute(check_email, {"email": signup_data.email}).fetchone() if existing_email: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="이미 등록된 이메일입니다" ) # 비밀번호 해싱 import bcrypt hashed_password = bcrypt.hashpw( signup_data.password.encode('utf-8'), bcrypt.gensalt() ).decode('utf-8') # 승인 대기 상태로 사용자 생성 new_user = user_repo.create_user({ 'username': signup_data.username, 'password': hashed_password, # 필드명은 'password' 'name': signup_data.name, 'email': signup_data.email, 'access_level': 'worker', # 기본 레벨 (승인 시 변경 가능) 'department': signup_data.department, 'position': signup_data.position, 'phone': signup_data.phone, 'role': 'user', 'is_active': False # 비활성 상태로 승인 대기 표시 }) # 가입 사유 저장 (notes 컬럼 활용) if signup_data.reason: update_notes = text("UPDATE users SET notes = :reason WHERE id = :user_id") db.execute(update_notes, {"reason": signup_data.reason, "user_id": new_user.id}) db.commit() return { "success": True, "message": "회원가입 요청이 전송되었습니다. 관리자 승인 후 이용 가능합니다.", "user_id": new_user.id, "username": new_user.username, "status": "pending_approval" } except HTTPException: raise except Exception as e: db.rollback() logger.error(f"회원가입 요청 실패: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"회원가입 요청 처리 중 오류가 발생했습니다: {str(e)}" ) @router.get("/signup-requests") async def get_signup_requests( current_user: dict = Depends(get_current_user), db: Session = Depends(get_db) ): """ 회원가입 요청 목록 조회 (관리자 전용) Returns: dict: 승인 대기 중인 사용자 목록 """ try: # 관리자 권한 확인 if current_user.get('role') not in ['admin', 'system']: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="관리자만 접근 가능합니다" ) # 승인 대기 중인 사용자 조회 (is_active=False인 사용자) query = text(""" SELECT user_id as id, username, name, email, department, position, phone, notes, created_at FROM users WHERE is_active = FALSE ORDER BY created_at DESC """) results = db.execute(query).fetchall() pending_users = [] for row in results: pending_users.append({ "id": row.id, "username": row.username, "name": row.name, "email": row.email, "department": row.department, "position": row.position, "phone": row.phone, "reason": row.notes, "requested_at": row.created_at.isoformat() if row.created_at else None }) return { "success": True, "requests": pending_users, "count": len(pending_users) } except HTTPException: raise except Exception as e: logger.error(f"가입 요청 목록 조회 실패: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"가입 요청 목록 조회 실패: {str(e)}" ) @router.post("/approve-signup/{user_id}") async def approve_signup( user_id: int, access_level: str = 'worker', current_user: dict = Depends(get_current_user), db: Session = Depends(get_db) ): """ 회원가입 승인 (관리자 전용) Args: user_id: 승인할 사용자 ID access_level: 부여할 접근 레벨 (worker, manager, admin) Returns: dict: 승인 결과 """ try: # 관리자 권한 확인 if current_user.get('role') not in ['admin', 'system']: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="관리자만 접근 가능합니다" ) # 사용자 활성화 및 접근 레벨 설정 update_query = text(""" UPDATE users SET is_active = TRUE, access_level = :access_level, updated_at = CURRENT_TIMESTAMP WHERE user_id = :user_id AND is_active = FALSE RETURNING user_id as id, username, name """) result = db.execute(update_query, { "user_id": user_id, "access_level": access_level }).fetchone() if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="승인 대기 중인 사용자를 찾을 수 없습니다" ) db.commit() return { "success": True, "message": f"{result.name}님의 가입이 승인되었습니다", "user": { "id": result.id, "username": result.username, "name": result.name, "access_level": access_level } } except HTTPException: raise except Exception as e: db.rollback() logger.error(f"가입 승인 실패: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"가입 승인 처리 중 오류가 발생했습니다: {str(e)}" ) @router.delete("/reject-signup/{user_id}") async def reject_signup( user_id: int, current_user: dict = Depends(get_current_user), db: Session = Depends(get_db) ): """ 회원가입 거부 (관리자 전용) Args: user_id: 거부할 사용자 ID Returns: dict: 거부 결과 """ try: # 관리자 권한 확인 if current_user.get('role') not in ['admin', 'system']: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="관리자만 접근 가능합니다" ) # 승인 대기 사용자 삭제 delete_query = text(""" DELETE FROM users WHERE user_id = :user_id AND is_active = FALSE RETURNING username, name """) result = db.execute(delete_query, {"user_id": user_id}).fetchone() if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="승인 대기 중인 사용자를 찾을 수 없습니다" ) db.commit() return { "success": True, "message": f"{result.name}님의 가입 요청이 거부되었습니다" } except HTTPException: raise except Exception as e: db.rollback() logger.error(f"가입 거부 실패: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"가입 거부 처리 중 오류가 발생했습니다: {str(e)}" )