DonaldFang 方士碩
f810ddc2ea
Features: - FastAPI backend with JWT authentication - MySQL database with SQLAlchemy ORM - KPI workflow: draft → pending → approved → evaluation → completed - Ollama LLM API integration for AI features - Gitea API integration for version control - Complete API endpoints for KPI, dashboard, notifications Tables: KPI_D_* prefix naming convention 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
134 lines
3.7 KiB
Python
134 lines
3.7 KiB
Python
"""
|
|
認證 API
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_db, get_current_user
|
|
from app.core.security import (
|
|
verify_password,
|
|
create_access_token,
|
|
create_refresh_token,
|
|
decode_token,
|
|
)
|
|
from app.models.employee import Employee
|
|
from app.schemas.auth import (
|
|
LoginRequest,
|
|
TokenResponse,
|
|
RefreshTokenRequest,
|
|
UserInfo,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["認證"])
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(data: LoginRequest, db: Session = Depends(get_db)):
|
|
"""
|
|
登入
|
|
|
|
使用員工編號和密碼登入，返回 JWT Token。
|
|
"""
|
|
# 查詢員工
|
|
employee = (
|
|
db.query(Employee).filter(Employee.employee_no == data.employee_no).first()
|
|
)
|
|
|
|
if not employee:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH001", "message": "員工編號或密碼錯誤"},
|
|
)
|
|
|
|
# 驗證密碼
|
|
if not verify_password(data.password, employee.password_hash):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH001", "message": "員工編號或密碼錯誤"},
|
|
)
|
|
|
|
# 檢查帳號狀態
|
|
if employee.status != "active":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH004", "message": "帳號已停用"},
|
|
)
|
|
|
|
# 產生 Token
|
|
access_token = create_access_token({"sub": str(employee.id)})
|
|
refresh_token = create_refresh_token({"sub": str(employee.id)})
|
|
|
|
return TokenResponse(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
)
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
def refresh_token(data: RefreshTokenRequest, db: Session = Depends(get_db)):
|
|
"""
|
|
更新 Token
|
|
|
|
使用 Refresh Token 取得新的 Access Token。
|
|
"""
|
|
payload = decode_token(data.refresh_token)
|
|
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH002", "message": "Token 過期或無效"},
|
|
)
|
|
|
|
if payload.get("type") != "refresh":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH001", "message": "Token 類型錯誤"},
|
|
)
|
|
|
|
user_id = payload.get("sub")
|
|
employee = db.query(Employee).filter(Employee.id == int(user_id)).first()
|
|
|
|
if not employee or employee.status != "active":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"code": "AUTH001", "message": "使用者不存在或已停用"},
|
|
)
|
|
|
|
# 產生新 Token
|
|
access_token = create_access_token({"sub": str(employee.id)})
|
|
new_refresh_token = create_refresh_token({"sub": str(employee.id)})
|
|
|
|
return TokenResponse(
|
|
access_token=access_token,
|
|
refresh_token=new_refresh_token,
|
|
)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout():
|
|
"""
|
|
登出
|
|
|
|
前端清除 Token 即可，後端不做處理。
|
|
"""
|
|
return {"message": "登出成功"}
|
|
|
|
|
|
@router.get("/me", response_model=UserInfo)
|
|
def get_me(current_user: Employee = Depends(get_current_user)):
|
|
"""
|
|
取得當前使用者資訊
|
|
"""
|
|
return UserInfo(
|
|
id=current_user.id,
|
|
employee_no=current_user.employee_no,
|
|
name=current_user.name,
|
|
email=current_user.email,
|
|
department_id=current_user.department_id,
|
|
department_name=current_user.department.name,
|
|
job_title=current_user.job_title,
|
|
role=current_user.role,
|
|
is_manager=current_user.is_manager,
|
|
is_admin=current_user.is_admin,
|
|
)
|