Compare commits

...

2 Commits

Author SHA1 Message Date
9146696062 Merge branch 'main' of https://gitea.theaken.com/donald/KPI-management 2025-12-11 16:29:06 +08:00
f810ddc2ea Initial commit: KPI Management System Backend
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>
2025-12-11 16:20:57 +08:00
48 changed files with 4950 additions and 0 deletions

26
.env.example Normal file
View File

@@ -0,0 +1,26 @@
# Database (MySQL)
DB_HOST=mysql.theaken.com
DB_PORT=33306
DB_NAME=db_A102
DB_USER=A102
DB_PASSWORD=Bb123456
DATABASE_URL=mysql+pymysql://A102:Bb123456@mysql.theaken.com:33306/db_A102
# JWT
SECRET_KEY=your-super-secret-key-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
# App
APP_NAME=KPI Management System
DEBUG=True
# Ollama LLM API
OLLAMA_API_URL=https://ollama_pjapi.theaken.com
OLLAMA_DEFAULT_MODEL=qwen2.5:3b
# Gitea
GITEA_URL=https://gitea.theaken.com
GITEA_USER=donald
GITEA_TOKEN=9e0a888d1a25bde9cf2ad5dff2bb7ee6d68d6ff0

53
.gitignore vendored Normal file
View File

@@ -0,0 +1,53 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
venv/
ENV/
env/
.venv/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Environment
.env
.env.local
# Testing
.coverage
htmlcov/
.pytest_cache/
# Database
*.db
*.sqlite3
# Logs
*.log
# OS
.DS_Store
Thumbs.db

1
app/__init__.py Normal file
View File

@@ -0,0 +1 @@
# KPI Management System

18
app/api/__init__.py Normal file
View File

@@ -0,0 +1,18 @@
"""
API Routers
"""
from app.api.auth import router as auth_router
from app.api.kpi import router as kpi_router
from app.api.dashboard import router as dashboard_router
from app.api.notifications import router as notifications_router
from app.api.llm import router as llm_router
from app.api.gitea import router as gitea_router
__all__ = [
"auth_router",
"kpi_router",
"dashboard_router",
"notifications_router",
"llm_router",
"gitea_router",
]

133
app/api/auth.py Normal file
View File

@@ -0,0 +1,133 @@
"""
認證 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,
)

105
app/api/dashboard.py Normal file
View File

@@ -0,0 +1,105 @@
"""
儀表板 API
"""
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user, get_current_manager
from app.models.employee import Employee
from app.services.dashboard_service import DashboardService
from app.schemas.dashboard import (
DashboardProgressResponse,
DashboardDistributionResponse,
DashboardTrendsResponse,
DashboardAlertResponse,
)
from app.schemas.common import MessageResponse
router = APIRouter(prefix="/api/dashboard", tags=["儀表板"])
@router.get("/progress", response_model=DashboardProgressResponse)
def get_progress(
period_id: Optional[int] = Query(None),
department_id: Optional[int] = Query(None),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得進度統計
返回指定期間的 KPI 完成進度,包含各狀態的數量統計。
"""
service = DashboardService(db)
return service.get_progress(period_id, department_id)
@router.get("/distribution", response_model=DashboardDistributionResponse)
def get_distribution(
period_id: Optional[int] = Query(None),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得分佈統計
返回 KPI 的部門分佈、狀態分佈、分數區間分佈。
"""
service = DashboardService(db)
return service.get_distribution(period_id)
@router.get("/trends", response_model=DashboardTrendsResponse)
def get_trends(
limit: int = Query(4, ge=1, le=10),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得趨勢統計
返回歷史期間的平均分數趨勢。
"""
service = DashboardService(db)
return service.get_trends(limit)
@router.get("/alerts", response_model=List[DashboardAlertResponse])
def get_alerts(
is_resolved: Optional[bool] = Query(False),
limit: int = Query(50, ge=1, le=200),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""
取得儀表板警示
返回系統警示列表(主管以上可查看)。
"""
service = DashboardService(db)
return service.get_alerts(is_resolved, limit)
@router.put("/alerts/{alert_id}/resolve", response_model=DashboardAlertResponse)
def resolve_alert(
alert_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""
解決警示
將指定警示標記為已解決。
"""
service = DashboardService(db)
alert = service.resolve_alert(alert_id, current_user.id)
if not alert:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "ALERT001", "message": "警示不存在"},
)
return alert

109
app/api/deps.py Normal file
View File

@@ -0,0 +1,109 @@
"""
API 依賴注入
"""
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from app.core.database import SessionLocal
from app.core.security import decode_token
from app.models.employee import Employee
# Bearer Token 安全機制
security = HTTPBearer()
def get_db() -> Generator:
"""取得資料庫 Session"""
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db),
) -> Employee:
"""
取得當前使用者
從 Authorization header 解析 JWT Token
驗證並返回對應的員工物件。
"""
token = credentials.credentials
payload = decode_token(token)
if not payload:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "AUTH001", "message": "Token 無效"},
)
# 檢查 Token 類型
if payload.get("type") != "access":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "AUTH001", "message": "Token 類型錯誤"},
)
user_id = payload.get("sub")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "AUTH001", "message": "Token 無效"},
)
user = db.query(Employee).filter(Employee.id == int(user_id)).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "AUTH001", "message": "使用者不存在"},
)
if user.status != "active":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "AUTH004", "message": "帳號已停用"},
)
return user
def get_current_manager(
current_user: Employee = Depends(get_current_user),
) -> Employee:
"""取得當前主管使用者"""
if not current_user.is_manager:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "權限不足,需要主管權限"},
)
return current_user
def get_current_admin(
current_user: Employee = Depends(get_current_user),
) -> Employee:
"""取得當前管理員使用者"""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "權限不足,需要管理員權限"},
)
return current_user
def get_current_hr(
current_user: Employee = Depends(get_current_user),
) -> Employee:
"""取得當前人資使用者"""
if not current_user.is_hr and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "權限不足,需要人資權限"},
)
return current_user

166
app/api/gitea.py Normal file
View File

@@ -0,0 +1,166 @@
"""
Gitea API
"""
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from app.api.deps import get_current_user, get_current_admin
from app.models.employee import Employee
from app.services.gitea_service import gitea_service, create_kpi_management_repo
router = APIRouter(prefix="/api/gitea", tags=["Gitea"])
class CreateRepoRequest(BaseModel):
"""建立 Repo 請求"""
name: str
description: Optional[str] = ""
private: bool = False
class CreateFileRequest(BaseModel):
"""建立檔案請求"""
repo_name: str
file_path: str
content: str
message: str = "Add file"
branch: str = "main"
@router.get("/user")
def get_user(current_user: Employee = Depends(get_current_admin)):
"""
取得 Gitea 使用者資訊
"""
return gitea_service.get_user()
@router.get("/repos")
def list_repos(current_user: Employee = Depends(get_current_admin)):
"""
列出所有 Repo
"""
repos = gitea_service.list_repos()
return {"repos": repos}
@router.get("/repos/{repo_name}")
def get_repo(
repo_name: str,
current_user: Employee = Depends(get_current_admin),
):
"""
取得 Repo 資訊
"""
return gitea_service.get_repo(repo_name)
@router.post("/repos")
def create_repo(
data: CreateRepoRequest,
current_user: Employee = Depends(get_current_admin),
):
"""
建立新的 Repo
"""
result = gitea_service.create_repo(
name=data.name,
description=data.description,
private=data.private,
)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
@router.delete("/repos/{repo_name}")
def delete_repo(
repo_name: str,
current_user: Employee = Depends(get_current_admin),
):
"""
刪除 Repo
"""
result = gitea_service.delete_repo(repo_name)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {"message": f"Repo '{repo_name}' 已刪除"}
@router.post("/repos/kpi-management/init")
def init_kpi_repo(current_user: Employee = Depends(get_current_admin)):
"""
初始化 KPI Management Repo
"""
result = create_kpi_management_repo()
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
@router.get("/repos/{repo_name}/branches")
def list_branches(
repo_name: str,
current_user: Employee = Depends(get_current_admin),
):
"""
列出分支
"""
return gitea_service.list_branches(repo_name)
@router.get("/repos/{repo_name}/commits")
def list_commits(
repo_name: str,
branch: str = "main",
limit: int = 10,
current_user: Employee = Depends(get_current_admin),
):
"""
列出 Commits
"""
return gitea_service.list_commits(repo_name, branch, limit)
@router.post("/files")
def create_file(
data: CreateFileRequest,
current_user: Employee = Depends(get_current_admin),
):
"""
建立檔案
"""
result = gitea_service.create_file(
repo_name=data.repo_name,
file_path=data.file_path,
content=data.content,
message=data.message,
branch=data.branch,
)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
@router.get("/repos/{repo_name}/files/{file_path:path}")
def get_file(
repo_name: str,
file_path: str,
current_user: Employee = Depends(get_current_admin),
):
"""
取得檔案內容
"""
return gitea_service.get_file(repo_name, file_path)

358
app/api/kpi.py Normal file
View File

@@ -0,0 +1,358 @@
"""
KPI API
"""
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user, get_current_manager
from app.models.employee import Employee
from app.models.kpi_sheet import KPISheet
from app.models.kpi_period import KPIPeriod
from app.models.kpi_template import KPITemplate, KPIPreset
from app.services.kpi_service import KPISheetService
from app.services.notify_service import NotifyService
from app.schemas.kpi_sheet import (
KPISheetCreate,
KPISheetResponse,
KPISheetListItem,
ApproveRequest,
RejectRequest,
SelfEvalRequest,
ManagerEvalRequest,
KPIPeriodResponse,
KPITemplateResponse,
KPIPresetResponse,
)
from app.schemas.common import PaginatedResponse, MessageResponse
router = APIRouter(prefix="/api/kpi", tags=["KPI"])
# ==================== KPI 表單 ====================
@router.get("/sheets", response_model=List[KPISheetListItem])
def list_sheets(
period_id: Optional[int] = Query(None),
department_id: Optional[int] = Query(None),
status: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=500),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""查詢 KPI 表單清單"""
service = KPISheetService(db)
sheets = service.get_multi(
period_id=period_id,
department_id=department_id,
status=status,
skip=skip,
limit=limit,
)
return sheets
@router.get("/sheets/my", response_model=List[KPISheetListItem])
def get_my_sheets(
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""取得我的 KPI 表單"""
service = KPISheetService(db)
return service.get_my_sheets(current_user.id)
@router.get("/sheets/pending", response_model=List[KPISheetListItem])
def get_pending_sheets(
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""取得待審核的 KPI 表單(主管用)"""
service = KPISheetService(db)
return service.get_pending_for_manager(current_user.id)
@router.post("/sheets", response_model=KPISheetResponse, status_code=status.HTTP_201_CREATED)
def create_sheet(
data: KPISheetCreate,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""建立 KPI 表單"""
service = KPISheetService(db)
return service.create(current_user, data)
@router.post("/sheets/copy-from-previous", response_model=KPISheetResponse, status_code=status.HTTP_201_CREATED)
def copy_from_previous(
period_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""從上期複製 KPI"""
service = KPISheetService(db)
return service.copy_from_previous(current_user, period_id)
@router.post("/sheets/apply-preset/{preset_id}", response_model=KPISheetResponse, status_code=status.HTTP_201_CREATED)
def apply_preset(
preset_id: int,
period_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""套用預設組合"""
service = KPISheetService(db)
return service.apply_preset(current_user, period_id, preset_id)
@router.get("/sheets/{sheet_id}", response_model=KPISheetResponse)
def get_sheet(
sheet_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""取得單一 KPI 表單"""
service = KPISheetService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
return sheet
@router.delete("/sheets/{sheet_id}", response_model=MessageResponse)
def delete_sheet(
sheet_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""刪除 KPI 表單"""
service = KPISheetService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
if sheet.employee_id != current_user.id and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "權限不足"},
)
service.delete(sheet)
return MessageResponse(message="刪除成功")
# ==================== 審核流程 ====================
@router.post("/sheets/{sheet_id}/submit", response_model=KPISheetResponse)
def submit_sheet(
sheet_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""提交 KPI 審核"""
service = KPISheetService(db)
notify_service = NotifyService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
if sheet.employee_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "只能提交自己的 KPI"},
)
result = service.submit(sheet, current_user)
notify_service.notify_kpi_submitted(result)
return result
@router.post("/sheets/{sheet_id}/approve", response_model=KPISheetResponse)
def approve_sheet(
sheet_id: int,
data: ApproveRequest,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""審核通過"""
service = KPISheetService(db)
notify_service = NotifyService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
# 檢查是否為該員工的主管
if sheet.employee.manager_id != current_user.id and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "只能審核直屬部屬的 KPI"},
)
result = service.approve(sheet, current_user, data.comment)
notify_service.notify_kpi_approved(result)
return result
@router.post("/sheets/{sheet_id}/reject", response_model=KPISheetResponse)
def reject_sheet(
sheet_id: int,
data: RejectRequest,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""退回"""
service = KPISheetService(db)
notify_service = NotifyService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
# 檢查是否為該員工的主管
if sheet.employee.manager_id != current_user.id and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "只能審核直屬部屬的 KPI"},
)
result = service.reject(sheet, current_user, data.reason)
notify_service.notify_kpi_rejected(result, data.reason)
return result
# ==================== 評核 ====================
@router.post("/sheets/{sheet_id}/self-eval", response_model=KPISheetResponse)
def self_eval(
sheet_id: int,
data: SelfEvalRequest,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""員工自評"""
service = KPISheetService(db)
notify_service = NotifyService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
if sheet.employee_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "只能自評自己的 KPI"},
)
result = service.self_eval(sheet, current_user, data)
notify_service.notify_self_eval_completed(result)
return result
@router.post("/sheets/{sheet_id}/manager-eval", response_model=KPISheetResponse)
def manager_eval(
sheet_id: int,
data: ManagerEvalRequest,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_manager),
):
"""主管評核"""
service = KPISheetService(db)
notify_service = NotifyService(db)
sheet = service.get_by_id(sheet_id)
if not sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI004", "message": "KPI 表單不存在"},
)
# 檢查是否為該員工的主管
if sheet.employee.manager_id != current_user.id and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"code": "AUTH003", "message": "只能評核直屬部屬的 KPI"},
)
result = service.manager_eval(sheet, current_user, data)
notify_service.notify_manager_eval_completed(result)
return result
# ==================== 期間 ====================
@router.get("/periods", response_model=List[KPIPeriodResponse])
def list_periods(db: Session = Depends(get_db)):
"""取得 KPI 期間列表"""
periods = db.query(KPIPeriod).order_by(KPIPeriod.start_date.desc()).all()
return periods
@router.get("/periods/current", response_model=KPIPeriodResponse)
def get_current_period(db: Session = Depends(get_db)):
"""取得當前 KPI 期間"""
period = (
db.query(KPIPeriod)
.filter(KPIPeriod.status.in_(["setting", "self_eval", "manager_eval"]))
.order_by(KPIPeriod.start_date.desc())
.first()
)
if not period:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "PERIOD002", "message": "目前沒有進行中的 KPI 期間"},
)
return period
# ==================== 範本 ====================
@router.get("/templates", response_model=List[KPITemplateResponse])
def list_templates(
category: Optional[str] = Query(None),
db: Session = Depends(get_db),
):
"""取得 KPI 範本列表"""
query = db.query(KPITemplate).filter(KPITemplate.is_active == True)
if category:
query = query.filter(KPITemplate.category == category)
return query.all()
@router.get("/presets", response_model=List[KPIPresetResponse])
def list_presets(db: Session = Depends(get_db)):
"""取得 KPI 預設組合列表"""
presets = db.query(KPIPreset).filter(KPIPreset.is_active == True).all()
return presets

140
app/api/llm.py Normal file
View File

@@ -0,0 +1,140 @@
"""
LLM API
"""
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from app.api.deps import get_current_user
from app.models.employee import Employee
from app.services.llm_service import llm_service
router = APIRouter(prefix="/api/llm", tags=["LLM"])
class ChatMessage(BaseModel):
"""聊天訊息"""
role: str # system, user, assistant
content: str
class ChatRequest(BaseModel):
"""聊天請求"""
messages: List[ChatMessage]
model: Optional[str] = None
temperature: float = 0.7
stream: bool = False
class ChatResponse(BaseModel):
"""聊天回應"""
content: str
model: str
class SimpleAskRequest(BaseModel):
"""簡單問答請求"""
question: str
system_prompt: Optional[str] = "You are a helpful assistant."
model: Optional[str] = None
class KPIAnalyzeRequest(BaseModel):
"""KPI 分析請求"""
kpi_data: dict
@router.get("/models")
def list_models(current_user: Employee = Depends(get_current_user)):
"""
列出可用的 LLM 模型
"""
models = llm_service.list_models()
return {"models": models}
@router.post("/chat", response_model=ChatResponse)
def chat(
data: ChatRequest,
current_user: Employee = Depends(get_current_user),
):
"""
聊天完成請求
"""
if data.stream:
raise HTTPException(
status_code=400,
detail="請使用 /chat/stream 端點進行串流請求",
)
messages = [{"role": m.role, "content": m.content} for m in data.messages]
content = llm_service.chat(
messages=messages,
model=data.model,
temperature=data.temperature,
)
return ChatResponse(
content=content,
model=data.model or llm_service.default_model,
)
@router.post("/chat/stream")
async def chat_stream(
data: ChatRequest,
current_user: Employee = Depends(get_current_user),
):
"""
串流聊天請求
"""
messages = [{"role": m.role, "content": m.content} for m in data.messages]
def generate():
for chunk in llm_service.chat_stream(
messages=messages,
model=data.model,
temperature=data.temperature,
):
yield f"data: {chunk}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
)
@router.post("/ask")
def simple_ask(
data: SimpleAskRequest,
current_user: Employee = Depends(get_current_user),
):
"""
簡單問答
"""
response = llm_service.simple_ask(
question=data.question,
system_prompt=data.system_prompt,
model=data.model,
)
return {"answer": response}
@router.post("/analyze-kpi")
def analyze_kpi(
data: KPIAnalyzeRequest,
current_user: Employee = Depends(get_current_user),
):
"""
AI 分析 KPI 數據
"""
analysis = llm_service.analyze_kpi(data.kpi_data)
return {"analysis": analysis}

133
app/api/notifications.py Normal file
View File

@@ -0,0 +1,133 @@
"""
通知 API
"""
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.employee import Employee
from app.services.notify_service import NotifyService
from app.schemas.notification import (
NotificationResponse,
NotificationPreferenceResponse,
NotificationPreferenceUpdate,
UnreadCountResponse,
)
from app.schemas.common import MessageResponse
router = APIRouter(prefix="/api/notifications", tags=["通知"])
@router.get("", response_model=List[NotificationResponse])
def list_notifications(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得通知列表
返回當前使用者的通知,按時間倒序排列。
"""
service = NotifyService(db)
return service.get_by_recipient(current_user.id, skip, limit)
@router.get("/unread-count", response_model=UnreadCountResponse)
def get_unread_count(
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得未讀數量
返回當前使用者的未讀通知數量。
"""
service = NotifyService(db)
count = service.get_unread_count(current_user.id)
return UnreadCountResponse(count=count)
@router.put("/{notification_id}/read", response_model=MessageResponse)
def mark_as_read(
notification_id: int,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
標記已讀
將指定通知標記為已讀。
"""
service = NotifyService(db)
success = service.mark_as_read(notification_id, current_user.id)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "NOTIFY001", "message": "通知不存在"},
)
return MessageResponse(message="已標記為已讀")
@router.put("/read-all", response_model=MessageResponse)
def mark_all_as_read(
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
全部標記已讀
將當前使用者的所有通知標記為已讀。
"""
service = NotifyService(db)
count = service.mark_all_as_read(current_user.id)
return MessageResponse(message=f"已將 {count} 則通知標記為已讀")
@router.get("/preferences", response_model=NotificationPreferenceResponse)
def get_preferences(
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
取得通知偏好
返回當前使用者的通知偏好設定。
"""
service = NotifyService(db)
pref = service.get_preferences(current_user.id)
if not pref:
# 返回預設值
return NotificationPreferenceResponse(
email_enabled=True,
in_app_enabled=True,
reminder_days_before=3,
)
return pref
@router.put("/preferences", response_model=NotificationPreferenceResponse)
def update_preferences(
data: NotificationPreferenceUpdate,
db: Session = Depends(get_db),
current_user: Employee = Depends(get_current_user),
):
"""
更新通知偏好
更新當前使用者的通知偏好設定。
"""
service = NotifyService(db)
return service.update_preferences(
current_user.id,
email_enabled=data.email_enabled,
in_app_enabled=data.in_app_enabled,
reminder_days_before=data.reminder_days_before,
)

1
app/core/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Core module

52
app/core/config.py Normal file
View File

@@ -0,0 +1,52 @@
"""
應用程式設定
"""
from functools import lru_cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""應用程式設定"""
# App
APP_NAME: str = "KPI Management System"
DEBUG: bool = True
# Database (MySQL)
DB_HOST: str = "mysql.theaken.com"
DB_PORT: int = 33306
DB_NAME: str = "db_A102"
DB_USER: str = "A102"
DB_PASSWORD: str = "Bb123456"
@property
def DATABASE_URL(self) -> str:
return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
# JWT
SECRET_KEY: str = "your-super-secret-key-change-in-production"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# Ollama LLM API
OLLAMA_API_URL: str = "https://ollama_pjapi.theaken.com"
OLLAMA_DEFAULT_MODEL: str = "qwen2.5:3b"
# Gitea
GITEA_URL: str = "https://gitea.theaken.com"
GITEA_USER: str = "donald"
GITEA_TOKEN: str = "9e0a888d1a25bde9cf2ad5dff2bb7ee6d68d6ff0"
class Config:
env_file = ".env"
case_sensitive = True
@lru_cache
def get_settings() -> Settings:
"""取得設定單例"""
return Settings()
settings = get_settings()

39
app/core/database.py Normal file
View File

@@ -0,0 +1,39 @@
"""
資料庫連線設定 (MySQL)
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from app.core.config import settings
# 建立資料庫引擎 (MySQL)
engine = create_engine(
settings.DATABASE_URL,
pool_pre_ping=True,
pool_size=10,
max_overflow=20,
pool_recycle=3600, # MySQL 連線回收
)
# 建立 Session 工廠
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 建立 Base 類別
Base = declarative_base()
def get_db():
"""
取得資料庫 Session
用於 FastAPI 依賴注入
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
def init_db():
"""初始化資料庫(建立所有表)"""
Base.metadata.create_all(bind=engine)

83
app/core/security.py Normal file
View File

@@ -0,0 +1,83 @@
"""
安全模組JWT 認證與密碼雜湊
"""
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
# 密碼雜湊上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""驗證密碼"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""產生密碼雜湊"""
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""
建立 Access Token
Args:
data: Token 內容 (通常包含 sub: user_id)
expires_delta: 過期時間
Returns:
JWT Token 字串
"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire, "type": "access"})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def create_refresh_token(data: dict) -> str:
"""
建立 Refresh Token
Args:
data: Token 內容
Returns:
JWT Token 字串
"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire, "type": "refresh"})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def decode_token(token: str) -> Optional[dict]:
"""
解碼 Token
Args:
token: JWT Token 字串
Returns:
Token 內容或 None (解碼失敗)
"""
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError:
return None

97
app/main.py Normal file
View File

@@ -0,0 +1,97 @@
"""
KPI 管理系統 - FastAPI 主程式
"""
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.core.config import settings
from app.core.database import init_db
from app.api import (
auth_router,
kpi_router,
dashboard_router,
notifications_router,
llm_router,
gitea_router,
)
# 建立 FastAPI 應用程式
app = FastAPI(
title=settings.APP_NAME,
description="KPI 管理系統 API - 整合 Ollama LLM 與 Gitea 版本控制",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
)
# CORS 設定
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生產環境應限制
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 全域例外處理
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""全域例外處理器"""
return JSONResponse(
status_code=500,
content={
"error": {
"code": "INTERNAL_ERROR",
"message": "內部伺服器錯誤",
"details": str(exc) if settings.DEBUG else None,
}
},
)
# 註冊路由
app.include_router(auth_router)
app.include_router(kpi_router)
app.include_router(dashboard_router)
app.include_router(notifications_router)
app.include_router(llm_router)
app.include_router(gitea_router)
# 啟動事件
@app.on_event("startup")
async def startup_event():
"""應用程式啟動時初始化資料庫"""
init_db()
# 健康檢查
@app.get("/health")
def health_check():
"""健康檢查端點"""
return {"status": "healthy", "app": settings.APP_NAME}
# 根路徑
@app.get("/")
def root():
"""API 根路徑"""
return {
"app": settings.APP_NAME,
"version": "1.0.0",
"docs": "/docs",
"features": [
"KPI 管理",
"員工績效考核",
"Ollama LLM 整合",
"Gitea 版本控制",
],
}
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

27
app/models/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
"""
SQLAlchemy Models
"""
from app.models.department import Department
from app.models.employee import Employee
from app.models.kpi_period import KPIPeriod
from app.models.kpi_template import KPITemplate, KPIPreset, KPIPresetItem
from app.models.kpi_sheet import KPISheet
from app.models.kpi_item import KPIItem
from app.models.kpi_review_log import KPIReviewLog
from app.models.notification import Notification, NotificationPreference
from app.models.dashboard_alert import DashboardAlert
__all__ = [
"Department",
"Employee",
"KPIPeriod",
"KPITemplate",
"KPIPreset",
"KPIPresetItem",
"KPISheet",
"KPIItem",
"KPIReviewLog",
"Notification",
"NotificationPreference",
"DashboardAlert",
]

View File

@@ -0,0 +1,41 @@
"""
儀表板警示 Model
"""
from datetime import datetime
from typing import Optional, TYPE_CHECKING
from sqlalchemy import String, Integer, Boolean, DateTime, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.employee import Employee
from app.models.kpi_sheet import KPISheet
class DashboardAlert(Base):
"""儀表板警示"""
__tablename__ = "KPI_D_alerts"
id: Mapped[int] = mapped_column(primary_key=True)
alert_type: Mapped[str] = mapped_column(String(50), nullable=False)
# deadline_approaching, overdue, weight_invalid
severity: Mapped[str] = mapped_column(String(20), default="warning") # info, warning, error
title: Mapped[str] = mapped_column(String(200), nullable=False)
description: Mapped[Optional[str]] = mapped_column(Text)
related_sheet_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_sheets.id", ondelete="CASCADE"))
related_employee_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
is_resolved: Mapped[bool] = mapped_column(Boolean, default=False)
resolved_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
resolved_by: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Relationships
related_sheet: Mapped[Optional["KPISheet"]] = relationship("KPISheet")
related_employee: Mapped[Optional["Employee"]] = relationship("Employee", foreign_keys=[related_employee_id])
resolver: Mapped[Optional["Employee"]] = relationship("Employee", foreign_keys=[resolved_by])
def __repr__(self) -> str:
return f"<DashboardAlert id={self.id} type={self.alert_type} severity={self.severity}>"

42
app/models/department.py Normal file
View File

@@ -0,0 +1,42 @@
"""
部門 Model
"""
from datetime import datetime
from typing import Optional, List, TYPE_CHECKING
from sqlalchemy import String, Integer, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.employee import Employee
class Department(Base):
"""部門"""
__tablename__ = "KPI_D_departments"
id: Mapped[int] = mapped_column(primary_key=True)
code: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(100), nullable=False)
level: Mapped[str] = mapped_column(String(20), default="DEPT") # COMPANY, BU, DEPT, TEAM
parent_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_departments.id"))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
parent: Mapped[Optional["Department"]] = relationship(
"Department", remote_side=[id], back_populates="children"
)
children: Mapped[List["Department"]] = relationship(
"Department", back_populates="parent"
)
employees: Mapped[List["Employee"]] = relationship(
"Employee", back_populates="department"
)
def __repr__(self) -> str:
return f"<Department {self.code}: {self.name}>"

68
app/models/employee.py Normal file
View File

@@ -0,0 +1,68 @@
"""
員工 Model
"""
from datetime import datetime, date
from typing import Optional, List, TYPE_CHECKING
from sqlalchemy import String, Integer, Boolean, DateTime, Date, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.department import Department
from app.models.kpi_sheet import KPISheet
from app.models.notification import Notification
class Employee(Base):
"""員工"""
__tablename__ = "KPI_D_employees"
id: Mapped[int] = mapped_column(primary_key=True)
employee_no: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(100), nullable=False)
email: Mapped[str] = mapped_column(String(200), unique=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(200), nullable=False)
department_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_departments.id"), nullable=False)
manager_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
job_title: Mapped[Optional[str]] = mapped_column(String(100))
role: Mapped[str] = mapped_column(String(20), default="employee") # employee, manager, admin, hr
status: Mapped[str] = mapped_column(String(20), default="active") # active, inactive, resigned
hire_date: Mapped[Optional[date]] = mapped_column(Date)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
department: Mapped["Department"] = relationship("Department", back_populates="employees")
manager: Mapped[Optional["Employee"]] = relationship(
"Employee", remote_side=[id], back_populates="subordinates"
)
subordinates: Mapped[List["Employee"]] = relationship(
"Employee", back_populates="manager"
)
kpi_sheets: Mapped[List["KPISheet"]] = relationship(
"KPISheet", back_populates="employee", foreign_keys="KPISheet.employee_id"
)
notifications: Mapped[List["Notification"]] = relationship(
"Notification", back_populates="recipient"
)
@property
def is_manager(self) -> bool:
"""是否為主管"""
return self.role in ("manager", "admin")
@property
def is_admin(self) -> bool:
"""是否為管理員"""
return self.role == "admin"
@property
def is_hr(self) -> bool:
"""是否為人資"""
return self.role == "hr"
def __repr__(self) -> str:
return f"<Employee {self.employee_no}: {self.name}>"

55
app/models/kpi_item.py Normal file
View File

@@ -0,0 +1,55 @@
"""
KPI 項目 Model
"""
from datetime import datetime
from typing import Optional, TYPE_CHECKING
from sqlalchemy import String, Integer, DateTime, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.kpi_sheet import KPISheet
from app.models.kpi_template import KPITemplate
class KPIItem(Base):
"""KPI 項目"""
__tablename__ = "KPI_D_items"
id: Mapped[int] = mapped_column(primary_key=True)
sheet_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_sheets.id", ondelete="CASCADE"), nullable=False)
template_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_templates.id"))
sort_order: Mapped[int] = mapped_column(Integer, default=0)
# 項目資訊
name: Mapped[str] = mapped_column(String(200), nullable=False)
category: Mapped[str] = mapped_column(String(50), nullable=False) # financial, customer, internal, learning
weight: Mapped[int] = mapped_column(Integer, nullable=False) # 權重百分比 (1-100)
# 等級標準
level0_criteria: Mapped[Optional[str]] = mapped_column(Text)
level1_criteria: Mapped[Optional[str]] = mapped_column(Text)
level2_criteria: Mapped[Optional[str]] = mapped_column(Text)
level3_criteria: Mapped[Optional[str]] = mapped_column(Text)
level4_criteria: Mapped[Optional[str]] = mapped_column(Text)
# 自評
self_eval_level: Mapped[Optional[int]] = mapped_column(Integer) # 0-4
self_eval_note: Mapped[Optional[str]] = mapped_column(Text)
# 主管評核
final_level: Mapped[Optional[int]] = mapped_column(Integer) # 0-4
final_note: Mapped[Optional[str]] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
sheet: Mapped["KPISheet"] = relationship("KPISheet", back_populates="items")
template: Mapped[Optional["KPITemplate"]] = relationship("KPITemplate", back_populates="kpi_items")
def __repr__(self) -> str:
return f"<KPIItem id={self.id} name={self.name} weight={self.weight}>"

63
app/models/kpi_period.py Normal file
View File

@@ -0,0 +1,63 @@
"""
KPI 期間 Model
"""
from datetime import datetime, date
from typing import Optional, List, TYPE_CHECKING
from sqlalchemy import String, DateTime, Date
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.kpi_sheet import KPISheet
class KPIPeriod(Base):
"""KPI 期間"""
__tablename__ = "KPI_D_periods"
id: Mapped[int] = mapped_column(primary_key=True)
code: Mapped[str] = mapped_column(String(20), unique=True, nullable=False) # 例: 2024H1
name: Mapped[str] = mapped_column(String(100), nullable=False)
start_date: Mapped[date] = mapped_column(Date, nullable=False)
end_date: Mapped[date] = mapped_column(Date, nullable=False)
setting_start: Mapped[date] = mapped_column(Date, nullable=False)
setting_end: Mapped[date] = mapped_column(Date, nullable=False)
self_eval_start: Mapped[Optional[date]] = mapped_column(Date)
self_eval_end: Mapped[Optional[date]] = mapped_column(Date)
manager_eval_start: Mapped[Optional[date]] = mapped_column(Date)
manager_eval_end: Mapped[Optional[date]] = mapped_column(Date)
status: Mapped[str] = mapped_column(String(20), default="draft")
# draft, setting, approved, self_eval, manager_eval, completed
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
kpi_sheets: Mapped[List["KPISheet"]] = relationship("KPISheet", back_populates="period")
@property
def is_setting_period(self) -> bool:
"""是否在設定期間"""
today = date.today()
return self.setting_start <= today <= self.setting_end
@property
def is_self_eval_period(self) -> bool:
"""是否在自評期間"""
if not self.self_eval_start or not self.self_eval_end:
return False
today = date.today()
return self.self_eval_start <= today <= self.self_eval_end
@property
def is_manager_eval_period(self) -> bool:
"""是否在主管評核期間"""
if not self.manager_eval_start or not self.manager_eval_end:
return False
today = date.today()
return self.manager_eval_start <= today <= self.manager_eval_end
def __repr__(self) -> str:
return f"<KPIPeriod {self.code}: {self.name}>"

View File

@@ -0,0 +1,36 @@
"""
KPI 審核紀錄 Model
"""
from datetime import datetime
from typing import Optional, TYPE_CHECKING
from sqlalchemy import String, Integer, DateTime, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.kpi_sheet import KPISheet
from app.models.employee import Employee
class KPIReviewLog(Base):
"""KPI 審核紀錄"""
__tablename__ = "KPI_D_review_logs"
id: Mapped[int] = mapped_column(primary_key=True)
sheet_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_sheets.id", ondelete="CASCADE"), nullable=False)
action: Mapped[str] = mapped_column(String(50), nullable=False) # submit, approve, reject, self_eval, manager_eval
actor_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_employees.id"), nullable=False)
from_status: Mapped[Optional[str]] = mapped_column(String(20))
to_status: Mapped[Optional[str]] = mapped_column(String(20))
comment: Mapped[Optional[str]] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Relationships
sheet: Mapped["KPISheet"] = relationship("KPISheet", back_populates="review_logs")
actor: Mapped["Employee"] = relationship("Employee")
def __repr__(self) -> str:
return f"<KPIReviewLog id={self.id} action={self.action} sheet={self.sheet_id}>"

103
app/models/kpi_sheet.py Normal file
View File

@@ -0,0 +1,103 @@
"""
KPI 表單 Model
"""
from datetime import datetime
from decimal import Decimal
from typing import Optional, List, TYPE_CHECKING
from sqlalchemy import String, Integer, DateTime, Text, Numeric, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.employee import Employee
from app.models.kpi_period import KPIPeriod
from app.models.department import Department
from app.models.kpi_item import KPIItem
from app.models.kpi_review_log import KPIReviewLog
# KPI 表單狀態
class KPISheetStatus:
DRAFT = "draft" # 草稿
PENDING = "pending" # 待審核
APPROVED = "approved" # 已核准
SELF_EVAL = "self_eval" # 自評中
MANAGER_EVAL = "manager_eval" # 主管評核中
COMPLETED = "completed" # 已完成
SETTLED = "settled" # 已結算
# 狀態轉換規則
VALID_STATUS_TRANSITIONS = {
KPISheetStatus.DRAFT: [KPISheetStatus.PENDING],
KPISheetStatus.PENDING: [KPISheetStatus.APPROVED, KPISheetStatus.DRAFT],
KPISheetStatus.APPROVED: [KPISheetStatus.SELF_EVAL],
KPISheetStatus.SELF_EVAL: [KPISheetStatus.MANAGER_EVAL],
KPISheetStatus.MANAGER_EVAL: [KPISheetStatus.COMPLETED],
KPISheetStatus.COMPLETED: [KPISheetStatus.SETTLED],
KPISheetStatus.SETTLED: [],
}
class KPISheet(Base):
"""KPI 表單"""
__tablename__ = "KPI_D_sheets"
id: Mapped[int] = mapped_column(primary_key=True)
employee_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_employees.id"), nullable=False)
period_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_periods.id"), nullable=False)
department_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_departments.id"), nullable=False)
status: Mapped[str] = mapped_column(String(20), default=KPISheetStatus.DRAFT)
# 提交資訊
submitted_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
# 審核資訊
approved_by: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
approved_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
approve_comment: Mapped[Optional[str]] = mapped_column(Text)
# 退回資訊
rejected_by: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
rejected_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
reject_reason: Mapped[Optional[str]] = mapped_column(Text)
# 自評資訊
self_eval_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
# 主管評核資訊
manager_eval_by: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_employees.id"))
manager_eval_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
manager_eval_comment: Mapped[Optional[str]] = mapped_column(Text)
# 分數
total_score: Mapped[Optional[Decimal]] = mapped_column(Numeric(5, 4))
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
employee: Mapped["Employee"] = relationship(
"Employee", back_populates="kpi_sheets", foreign_keys=[employee_id]
)
period: Mapped["KPIPeriod"] = relationship("KPIPeriod", back_populates="kpi_sheets")
department: Mapped["Department"] = relationship("Department")
items: Mapped[List["KPIItem"]] = relationship(
"KPIItem", back_populates="sheet", cascade="all, delete-orphan"
)
review_logs: Mapped[List["KPIReviewLog"]] = relationship(
"KPIReviewLog", back_populates="sheet", cascade="all, delete-orphan"
)
approver: Mapped[Optional["Employee"]] = relationship("Employee", foreign_keys=[approved_by])
rejecter: Mapped[Optional["Employee"]] = relationship("Employee", foreign_keys=[rejected_by])
manager_evaluator: Mapped[Optional["Employee"]] = relationship("Employee", foreign_keys=[manager_eval_by])
def can_transition_to(self, new_status: str) -> bool:
"""檢查是否可以轉換到指定狀態"""
return new_status in VALID_STATUS_TRANSITIONS.get(self.status, [])
def __repr__(self) -> str:
return f"<KPISheet id={self.id} employee={self.employee_id} period={self.period_id} status={self.status}>"

View File

@@ -0,0 +1,88 @@
"""
KPI 範本 Model
"""
from datetime import datetime
from typing import Optional, List, TYPE_CHECKING
from sqlalchemy import String, Integer, Boolean, DateTime, Text, ForeignKey, JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.kpi_item import KPIItem
class KPITemplate(Base):
"""KPI 範本"""
__tablename__ = "KPI_D_templates"
id: Mapped[int] = mapped_column(primary_key=True)
code: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(200), nullable=False)
category: Mapped[str] = mapped_column(String(50), nullable=False) # financial, customer, internal, learning
description: Mapped[Optional[str]] = mapped_column(Text)
default_weight: Mapped[int] = mapped_column(Integer, default=20)
level0_desc: Mapped[str] = mapped_column(Text, nullable=False)
level1_desc: Mapped[str] = mapped_column(Text, nullable=False)
level2_desc: Mapped[str] = mapped_column(Text, nullable=False)
level3_desc: Mapped[str] = mapped_column(Text, nullable=False)
level4_desc: Mapped[str] = mapped_column(Text, nullable=False)
applicable_roles: Mapped[Optional[str]] = mapped_column(JSON) # JSON array for MySQL
applicable_depts: Mapped[Optional[str]] = mapped_column(JSON) # JSON array for MySQL
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
kpi_items: Mapped[List["KPIItem"]] = relationship("KPIItem", back_populates="template")
preset_items: Mapped[List["KPIPresetItem"]] = relationship("KPIPresetItem", back_populates="template")
def __repr__(self) -> str:
return f"<KPITemplate {self.code}: {self.name}>"
class KPIPreset(Base):
"""KPI 預設組合"""
__tablename__ = "KPI_D_presets"
id: Mapped[int] = mapped_column(primary_key=True)
code: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(200), nullable=False)
description: Mapped[Optional[str]] = mapped_column(Text)
applicable_roles: Mapped[Optional[str]] = mapped_column(JSON) # JSON array for MySQL
applicable_depts: Mapped[Optional[str]] = mapped_column(JSON) # JSON array for MySQL
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
items: Mapped[List["KPIPresetItem"]] = relationship(
"KPIPresetItem", back_populates="preset", cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<KPIPreset {self.code}: {self.name}>"
class KPIPresetItem(Base):
"""KPI 預設項目"""
__tablename__ = "KPI_D_preset_items"
id: Mapped[int] = mapped_column(primary_key=True)
preset_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_presets.id", ondelete="CASCADE"), nullable=False)
template_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_templates.id"), nullable=False)
default_weight: Mapped[int] = mapped_column(Integer, nullable=False)
is_mandatory: Mapped[bool] = mapped_column(Boolean, default=False)
sort_order: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Relationships
preset: Mapped["KPIPreset"] = relationship("KPIPreset", back_populates="items")
template: Mapped["KPITemplate"] = relationship("KPITemplate", back_populates="preset_items")
def __repr__(self) -> str:
return f"<KPIPresetItem preset={self.preset_id} template={self.template_id}>"

View File

@@ -0,0 +1,58 @@
"""
通知 Model
"""
from datetime import datetime
from typing import Optional, TYPE_CHECKING
from sqlalchemy import String, Integer, Boolean, DateTime, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.employee import Employee
from app.models.kpi_sheet import KPISheet
class Notification(Base):
"""通知"""
__tablename__ = "KPI_D_notifications"
id: Mapped[int] = mapped_column(primary_key=True)
recipient_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_employees.id"), nullable=False)
type: Mapped[str] = mapped_column(String(50), nullable=False)
# kpi_submitted, kpi_approved, kpi_rejected, eval_reminder, etc.
title: Mapped[str] = mapped_column(String(200), nullable=False)
content: Mapped[Optional[str]] = mapped_column(Text)
related_sheet_id: Mapped[Optional[int]] = mapped_column(ForeignKey("KPI_D_sheets.id", ondelete="SET NULL"))
is_read: Mapped[bool] = mapped_column(Boolean, default=False)
read_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Relationships
recipient: Mapped["Employee"] = relationship("Employee", back_populates="notifications")
related_sheet: Mapped[Optional["KPISheet"]] = relationship("KPISheet")
def __repr__(self) -> str:
return f"<Notification id={self.id} type={self.type} recipient={self.recipient_id}>"
class NotificationPreference(Base):
"""通知偏好"""
__tablename__ = "KPI_D_notification_preferences"
id: Mapped[int] = mapped_column(primary_key=True)
employee_id: Mapped[int] = mapped_column(ForeignKey("KPI_D_employees.id"), unique=True, nullable=False)
email_enabled: Mapped[bool] = mapped_column(Boolean, default=True)
in_app_enabled: Mapped[bool] = mapped_column(Boolean, default=True)
reminder_days_before: Mapped[int] = mapped_column(Integer, default=3)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
employee: Mapped["Employee"] = relationship("Employee")
def __repr__(self) -> str:
return f"<NotificationPreference employee={self.employee_id}>"

9
app/schemas/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""
Pydantic Schemas
"""
from app.schemas.auth import *
from app.schemas.employee import *
from app.schemas.kpi_sheet import *
from app.schemas.kpi_item import *
from app.schemas.notification import *
from app.schemas.common import *

44
app/schemas/auth.py Normal file
View File

@@ -0,0 +1,44 @@
"""
認證 Schemas
"""
from typing import Optional
from pydantic import BaseModel, EmailStr
class LoginRequest(BaseModel):
"""登入請求"""
employee_no: str
password: str
class TokenResponse(BaseModel):
"""Token 回應"""
access_token: str
refresh_token: str
token_type: str = "bearer"
class RefreshTokenRequest(BaseModel):
"""更新 Token 請求"""
refresh_token: str
class UserInfo(BaseModel):
"""使用者資訊"""
id: int
employee_no: str
name: str
email: str
department_id: int
department_name: str
job_title: Optional[str]
role: str
is_manager: bool
is_admin: bool
class Config:
from_attributes = True

37
app/schemas/common.py Normal file
View File

@@ -0,0 +1,37 @@
"""
共用 Schemas
"""
from typing import Generic, TypeVar, Optional, List
from pydantic import BaseModel
T = TypeVar("T")
class ErrorDetail(BaseModel):
"""錯誤詳情"""
code: str
message: str
details: Optional[dict] = None
class ErrorResponse(BaseModel):
"""錯誤回應"""
error: ErrorDetail
class PaginatedResponse(BaseModel, Generic[T]):
"""分頁回應"""
items: List[T]
total: int
page: int
page_size: int
pages: int
class MessageResponse(BaseModel):
"""訊息回應"""
message: str

74
app/schemas/dashboard.py Normal file
View File

@@ -0,0 +1,74 @@
"""
儀表板 Schemas
"""
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel
class ProgressStats(BaseModel):
"""進度統計"""
total: int
draft: int
pending: int
approved: int
self_eval: int
manager_eval: int
completed: int
class DistributionItem(BaseModel):
"""分佈項目"""
label: str
count: int
percentage: float
class TrendItem(BaseModel):
"""趨勢項目"""
period: str
average_score: float
completed_count: int
class DashboardAlertResponse(BaseModel):
"""儀表板警示回應"""
id: int
alert_type: str
severity: str
title: str
description: Optional[str]
related_sheet_id: Optional[int]
related_employee_id: Optional[int]
is_resolved: bool
created_at: datetime
class Config:
from_attributes = True
class DashboardProgressResponse(BaseModel):
"""儀表板進度回應"""
period_code: str
period_name: str
stats: ProgressStats
completion_rate: float
class DashboardDistributionResponse(BaseModel):
"""儀表板分佈回應"""
by_department: List[DistributionItem]
by_status: List[DistributionItem]
by_score_range: List[DistributionItem]
class DashboardTrendsResponse(BaseModel):
"""儀表板趨勢回應"""
trends: List[TrendItem]

87
app/schemas/employee.py Normal file
View File

@@ -0,0 +1,87 @@
"""
員工 Schemas
"""
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, EmailStr
class DepartmentBase(BaseModel):
"""部門基本資訊"""
id: int
code: str
name: str
level: str
class DepartmentResponse(DepartmentBase):
"""部門回應"""
parent_id: Optional[int]
is_active: bool
class Config:
from_attributes = True
class EmployeeBase(BaseModel):
"""員工基本資訊"""
id: int
employee_no: str
name: str
email: str
class EmployeeSimple(EmployeeBase):
"""員工簡要資訊"""
job_title: Optional[str]
role: str
class Config:
from_attributes = True
class EmployeeResponse(EmployeeBase):
"""員工回應"""
department_id: int
department: DepartmentBase
manager_id: Optional[int]
manager: Optional[EmployeeBase]
job_title: Optional[str]
role: str
status: str
hire_date: Optional[date]
created_at: datetime
class Config:
from_attributes = True
class EmployeeCreate(BaseModel):
"""建立員工"""
employee_no: str
name: str
email: EmailStr
password: str
department_id: int
manager_id: Optional[int] = None
job_title: Optional[str] = None
role: str = "employee"
hire_date: Optional[date] = None
class EmployeeUpdate(BaseModel):
"""更新員工"""
name: Optional[str] = None
email: Optional[EmailStr] = None
department_id: Optional[int] = None
manager_id: Optional[int] = None
job_title: Optional[str] = None
role: Optional[str] = None
status: Optional[str] = None

74
app/schemas/kpi_item.py Normal file
View File

@@ -0,0 +1,74 @@
"""
KPI 項目 Schemas
"""
from typing import Optional
from pydantic import BaseModel, Field
class KPIItemBase(BaseModel):
"""KPI 項目基本資訊"""
name: str
category: str # financial, customer, internal, learning
weight: int = Field(..., ge=1, le=100, description="權重百分比 (1-100)")
class KPIItemCreate(KPIItemBase):
"""建立 KPI 項目"""
template_id: Optional[int] = None
level0_criteria: Optional[str] = None
level1_criteria: Optional[str] = None
level2_criteria: Optional[str] = None
level3_criteria: Optional[str] = None
level4_criteria: Optional[str] = None
class KPIItemUpdate(BaseModel):
"""更新 KPI 項目"""
name: Optional[str] = None
category: Optional[str] = None
weight: Optional[int] = Field(None, ge=1, le=100)
level0_criteria: Optional[str] = None
level1_criteria: Optional[str] = None
level2_criteria: Optional[str] = None
level3_criteria: Optional[str] = None
level4_criteria: Optional[str] = None
class KPIItemResponse(KPIItemBase):
"""KPI 項目回應"""
id: int
sheet_id: int
template_id: Optional[int]
sort_order: int
level0_criteria: Optional[str]
level1_criteria: Optional[str]
level2_criteria: Optional[str]
level3_criteria: Optional[str]
level4_criteria: Optional[str]
self_eval_level: Optional[int]
self_eval_note: Optional[str]
final_level: Optional[int]
final_note: Optional[str]
class Config:
from_attributes = True
class SelfEvalItem(BaseModel):
"""自評項目"""
id: int
level: int = Field(..., ge=0, le=4, description="自評等級 (0-4)")
note: Optional[str] = None
class ManagerEvalItem(BaseModel):
"""主管評核項目"""
id: int
level: int = Field(..., ge=0, le=4, description="評核等級 (0-4)")
note: Optional[str] = None

163
app/schemas/kpi_sheet.py Normal file
View File

@@ -0,0 +1,163 @@
"""
KPI 表單 Schemas
"""
from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel
from app.schemas.employee import EmployeeSimple
from app.schemas.kpi_item import KPIItemCreate, KPIItemResponse, SelfEvalItem, ManagerEvalItem
class KPIPeriodBase(BaseModel):
"""KPI 期間基本資訊"""
id: int
code: str
name: str
class KPIPeriodResponse(KPIPeriodBase):
"""KPI 期間回應"""
start_date: str
end_date: str
setting_start: str
setting_end: str
self_eval_start: Optional[str]
self_eval_end: Optional[str]
manager_eval_start: Optional[str]
manager_eval_end: Optional[str]
status: str
class Config:
from_attributes = True
class KPISheetCreate(BaseModel):
"""建立 KPI 表單"""
period_id: int
items: List[KPIItemCreate]
class KPISheetUpdate(BaseModel):
"""更新 KPI 表單"""
items: Optional[List[KPIItemCreate]] = None
class KPISheetResponse(BaseModel):
"""KPI 表單回應"""
id: int
employee: EmployeeSimple
period: KPIPeriodBase
department_id: int
status: str
items: List[KPIItemResponse]
# 提交資訊
submitted_at: Optional[datetime]
# 審核資訊
approved_by: Optional[int]
approved_at: Optional[datetime]
approve_comment: Optional[str]
# 退回資訊
rejected_by: Optional[int]
rejected_at: Optional[datetime]
reject_reason: Optional[str]
# 自評資訊
self_eval_at: Optional[datetime]
# 主管評核資訊
manager_eval_by: Optional[int]
manager_eval_at: Optional[datetime]
manager_eval_comment: Optional[str]
# 分數
total_score: Optional[Decimal]
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class KPISheetListItem(BaseModel):
"""KPI 表單列表項目"""
id: int
employee: EmployeeSimple
period: KPIPeriodBase
status: str
total_score: Optional[Decimal]
submitted_at: Optional[datetime]
created_at: datetime
class Config:
from_attributes = True
class ApproveRequest(BaseModel):
"""審核通過請求"""
comment: Optional[str] = None
class RejectRequest(BaseModel):
"""退回請求"""
reason: str
class SelfEvalRequest(BaseModel):
"""自評請求"""
items: List[SelfEvalItem]
class ManagerEvalRequest(BaseModel):
"""主管評核請求"""
items: List[ManagerEvalItem]
comment: Optional[str] = None
# KPI 範本相關
class KPITemplateResponse(BaseModel):
"""KPI 範本回應"""
id: int
code: str
name: str
category: str
description: Optional[str]
default_weight: int
level0_desc: str
level1_desc: str
level2_desc: str
level3_desc: str
level4_desc: str
is_active: bool
class Config:
from_attributes = True
class KPIPresetResponse(BaseModel):
"""KPI 預設組合回應"""
id: int
code: str
name: str
description: Optional[str]
items: List[KPITemplateResponse]
class Config:
from_attributes = True

View File

@@ -0,0 +1,47 @@
"""
通知 Schemas
"""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class NotificationResponse(BaseModel):
"""通知回應"""
id: int
type: str
title: str
content: Optional[str]
related_sheet_id: Optional[int]
is_read: bool
read_at: Optional[datetime]
created_at: datetime
class Config:
from_attributes = True
class NotificationPreferenceResponse(BaseModel):
"""通知偏好回應"""
email_enabled: bool
in_app_enabled: bool
reminder_days_before: int
class Config:
from_attributes = True
class NotificationPreferenceUpdate(BaseModel):
"""更新通知偏好"""
email_enabled: Optional[bool] = None
in_app_enabled: Optional[bool] = None
reminder_days_before: Optional[int] = None
class UnreadCountResponse(BaseModel):
"""未讀數量回應"""
count: int

12
app/services/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""
Business Logic Services
"""
from app.services.kpi_service import KPISheetService
from app.services.notify_service import NotifyService
from app.services.dashboard_service import DashboardService
__all__ = [
"KPISheetService",
"NotifyService",
"DashboardService",
]

View File

@@ -0,0 +1,277 @@
"""
儀表板服務
"""
from datetime import datetime
from typing import Optional, List
from decimal import Decimal
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.models.kpi_sheet import KPISheet, KPISheetStatus
from app.models.kpi_period import KPIPeriod
from app.models.department import Department
from app.models.employee import Employee
from app.models.dashboard_alert import DashboardAlert
from app.schemas.dashboard import (
ProgressStats,
DistributionItem,
TrendItem,
DashboardProgressResponse,
DashboardDistributionResponse,
DashboardTrendsResponse,
)
class DashboardService:
"""儀表板服務"""
def __init__(self, db: Session):
self.db = db
def get_progress(
self, period_id: Optional[int] = None, department_id: Optional[int] = None
) -> DashboardProgressResponse:
"""取得進度統計"""
# 取得期間
if period_id:
period = self.db.query(KPIPeriod).filter(KPIPeriod.id == period_id).first()
else:
period = (
self.db.query(KPIPeriod)
.filter(KPIPeriod.status != "completed")
.order_by(KPIPeriod.start_date.desc())
.first()
)
if not period:
return DashboardProgressResponse(
period_code="N/A",
period_name="無期間資料",
stats=ProgressStats(
total=0,
draft=0,
pending=0,
approved=0,
self_eval=0,
manager_eval=0,
completed=0,
),
completion_rate=0.0,
)
# 查詢統計
query = self.db.query(KPISheet).filter(KPISheet.period_id == period.id)
if department_id:
query = query.filter(KPISheet.department_id == department_id)
sheets = query.all()
total = len(sheets)
stats = ProgressStats(
total=total,
draft=sum(1 for s in sheets if s.status == KPISheetStatus.DRAFT),
pending=sum(1 for s in sheets if s.status == KPISheetStatus.PENDING),
approved=sum(1 for s in sheets if s.status == KPISheetStatus.APPROVED),
self_eval=sum(1 for s in sheets if s.status == KPISheetStatus.SELF_EVAL),
manager_eval=sum(
1 for s in sheets if s.status == KPISheetStatus.MANAGER_EVAL
),
completed=sum(1 for s in sheets if s.status == KPISheetStatus.COMPLETED),
)
completion_rate = (stats.completed / total * 100) if total > 0 else 0.0
return DashboardProgressResponse(
period_code=period.code,
period_name=period.name,
stats=stats,
completion_rate=round(completion_rate, 1),
)
def get_distribution(
self, period_id: Optional[int] = None
) -> DashboardDistributionResponse:
"""取得分佈統計"""
# 取得期間
if period_id:
period = self.db.query(KPIPeriod).filter(KPIPeriod.id == period_id).first()
else:
period = (
self.db.query(KPIPeriod)
.filter(KPIPeriod.status != "completed")
.order_by(KPIPeriod.start_date.desc())
.first()
)
if not period:
return DashboardDistributionResponse(
by_department=[],
by_status=[],
by_score_range=[],
)
sheets = (
self.db.query(KPISheet).filter(KPISheet.period_id == period.id).all()
)
total = len(sheets) or 1 # 避免除以零
# 按部門分佈
dept_counts = {}
for sheet in sheets:
dept_id = sheet.department_id
dept = self.db.query(Department).filter(Department.id == dept_id).first()
dept_name = dept.name if dept else "未知"
dept_counts[dept_name] = dept_counts.get(dept_name, 0) + 1
by_department = [
DistributionItem(
label=name, count=count, percentage=round(count / total * 100, 1)
)
for name, count in dept_counts.items()
]
# 按狀態分佈
status_labels = {
KPISheetStatus.DRAFT: "草稿",
KPISheetStatus.PENDING: "待審核",
KPISheetStatus.APPROVED: "已核准",
KPISheetStatus.SELF_EVAL: "自評中",
KPISheetStatus.MANAGER_EVAL: "主管評核中",
KPISheetStatus.COMPLETED: "已完成",
}
status_counts = {}
for sheet in sheets:
label = status_labels.get(sheet.status, sheet.status)
status_counts[label] = status_counts.get(label, 0) + 1
by_status = [
DistributionItem(
label=name, count=count, percentage=round(count / total * 100, 1)
)
for name, count in status_counts.items()
]
# 按分數區間分佈(僅已完成)
completed_sheets = [
s for s in sheets if s.status == KPISheetStatus.COMPLETED and s.total_score
]
score_ranges = {
"0-0.25": 0,
"0.25-0.5": 0,
"0.5-0.75": 0,
"0.75-1.0": 0,
}
for sheet in completed_sheets:
score = float(sheet.total_score)
if score < 0.25:
score_ranges["0-0.25"] += 1
elif score < 0.5:
score_ranges["0.25-0.5"] += 1
elif score < 0.75:
score_ranges["0.5-0.75"] += 1
else:
score_ranges["0.75-1.0"] += 1
completed_total = len(completed_sheets) or 1
by_score_range = [
DistributionItem(
label=name,
count=count,
percentage=round(count / completed_total * 100, 1),
)
for name, count in score_ranges.items()
]
return DashboardDistributionResponse(
by_department=by_department,
by_status=by_status,
by_score_range=by_score_range,
)
def get_trends(self, limit: int = 4) -> DashboardTrendsResponse:
"""取得趨勢統計"""
periods = (
self.db.query(KPIPeriod)
.filter(KPIPeriod.status == "completed")
.order_by(KPIPeriod.end_date.desc())
.limit(limit)
.all()
)
trends = []
for period in reversed(periods):
completed_sheets = (
self.db.query(KPISheet)
.filter(
KPISheet.period_id == period.id,
KPISheet.status == KPISheetStatus.COMPLETED,
)
.all()
)
if completed_sheets:
scores = [
float(s.total_score) for s in completed_sheets if s.total_score
]
avg_score = sum(scores) / len(scores) if scores else 0
else:
avg_score = 0
trends.append(
TrendItem(
period=period.code,
average_score=round(avg_score, 3),
completed_count=len(completed_sheets),
)
)
return DashboardTrendsResponse(trends=trends)
# ==================== 警示 ====================
def get_alerts(
self, is_resolved: Optional[bool] = False, limit: int = 50
) -> List[DashboardAlert]:
"""取得警示"""
query = self.db.query(DashboardAlert)
if is_resolved is not None:
query = query.filter(DashboardAlert.is_resolved == is_resolved)
return query.order_by(DashboardAlert.created_at.desc()).limit(limit).all()
def create_alert(
self,
alert_type: str,
severity: str,
title: str,
description: Optional[str] = None,
related_sheet_id: Optional[int] = None,
related_employee_id: Optional[int] = None,
) -> DashboardAlert:
"""建立警示"""
alert = DashboardAlert(
alert_type=alert_type,
severity=severity,
title=title,
description=description,
related_sheet_id=related_sheet_id,
related_employee_id=related_employee_id,
)
self.db.add(alert)
self.db.commit()
self.db.refresh(alert)
return alert
def resolve_alert(self, alert_id: int, resolver_id: int) -> Optional[DashboardAlert]:
"""解決警示"""
alert = self.db.query(DashboardAlert).filter(DashboardAlert.id == alert_id).first()
if not alert:
return None
alert.is_resolved = True
alert.resolved_at = datetime.utcnow()
alert.resolved_by = resolver_id
self.db.commit()
self.db.refresh(alert)
return alert

View File

@@ -0,0 +1,198 @@
"""
Gitea 版本控制服務
"""
from typing import Optional, List, Dict, Any
import requests
from app.core.config import settings
class GiteaService:
"""Gitea API 服務"""
def __init__(self):
self.base_url = settings.GITEA_URL
self.user = settings.GITEA_USER
self.token = settings.GITEA_TOKEN
self.headers = {
"Authorization": f"token {self.token}",
"Content-Type": "application/json",
}
def _request(
self,
method: str,
endpoint: str,
data: Optional[dict] = None,
) -> Dict[str, Any]:
"""
發送 API 請求
Args:
method: HTTP 方法
endpoint: API 端點
data: 請求資料
Returns:
API 回應
"""
url = f"{self.base_url}/api/v1{endpoint}"
try:
response = requests.request(
method=method,
url=url,
headers=self.headers,
json=data,
timeout=30,
)
if response.status_code == 204:
return {"success": True}
return response.json()
except Exception as e:
return {"error": str(e)}
def get_user(self) -> Dict[str, Any]:
"""取得當前使用者資訊"""
return self._request("GET", "/user")
def list_repos(self) -> List[Dict[str, Any]]:
"""列出使用者的所有 Repo"""
return self._request("GET", f"/users/{self.user}/repos")
def get_repo(self, repo_name: str) -> Dict[str, Any]:
"""
取得 Repo 資訊
Args:
repo_name: Repo 名稱
"""
return self._request("GET", f"/repos/{self.user}/{repo_name}")
def create_repo(
self,
name: str,
description: str = "",
private: bool = False,
auto_init: bool = True,
default_branch: str = "main",
) -> Dict[str, Any]:
"""
建立新的 Repo
Args:
name: Repo 名稱
description: 描述
private: 是否為私有
auto_init: 是否自動初始化 (建立 README)
default_branch: 預設分支名稱
Returns:
建立結果
"""
data = {
"name": name,
"description": description,
"private": private,
"auto_init": auto_init,
"default_branch": default_branch,
}
return self._request("POST", "/user/repos", data)
def delete_repo(self, repo_name: str) -> Dict[str, Any]:
"""
刪除 Repo
Args:
repo_name: Repo 名稱
"""
return self._request("DELETE", f"/repos/{self.user}/{repo_name}")
def create_file(
self,
repo_name: str,
file_path: str,
content: str,
message: str = "Add file",
branch: str = "main",
) -> Dict[str, Any]:
"""
在 Repo 中建立檔案
Args:
repo_name: Repo 名稱
file_path: 檔案路徑
content: 檔案內容
message: Commit 訊息
branch: 分支名稱
"""
import base64
data = {
"content": base64.b64encode(content.encode()).decode(),
"message": message,
"branch": branch,
}
return self._request(
"POST", f"/repos/{self.user}/{repo_name}/contents/{file_path}", data
)
def get_file(self, repo_name: str, file_path: str) -> Dict[str, Any]:
"""
取得檔案內容
Args:
repo_name: Repo 名稱
file_path: 檔案路徑
"""
return self._request(
"GET", f"/repos/{self.user}/{repo_name}/contents/{file_path}"
)
def list_branches(self, repo_name: str) -> List[Dict[str, Any]]:
"""
列出分支
Args:
repo_name: Repo 名稱
"""
return self._request("GET", f"/repos/{self.user}/{repo_name}/branches")
def list_commits(
self, repo_name: str, branch: str = "main", limit: int = 10
) -> List[Dict[str, Any]]:
"""
列出 Commits
Args:
repo_name: Repo 名稱
branch: 分支名稱
limit: 數量限制
"""
return self._request(
"GET", f"/repos/{self.user}/{repo_name}/commits?sha={branch}&limit={limit}"
)
# 單例
gitea_service = GiteaService()
def create_kpi_management_repo() -> Dict[str, Any]:
"""
建立 KPI Management Repo
Returns:
建立結果
"""
result = gitea_service.create_repo(
name="KPI-management",
description="KPI 管理系統 - 員工績效考核管理平台",
private=False,
auto_init=True,
default_branch="main",
)
return result

408
app/services/kpi_service.py Normal file
View File

@@ -0,0 +1,408 @@
"""
KPI 表單服務
"""
from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.models.employee import Employee
from app.models.kpi_period import KPIPeriod
from app.models.kpi_sheet import KPISheet, KPISheetStatus, VALID_STATUS_TRANSITIONS
from app.models.kpi_item import KPIItem
from app.models.kpi_template import KPITemplate, KPIPreset
from app.models.kpi_review_log import KPIReviewLog
from app.schemas.kpi_sheet import KPISheetCreate, SelfEvalRequest, ManagerEvalRequest
from app.schemas.kpi_item import KPIItemCreate
# 等級對應獎金月數
LEVEL_MONTHS = {
0: Decimal("0"),
1: Decimal("0.25"),
2: Decimal("0.5"),
3: Decimal("0.75"),
4: Decimal("1.0"),
}
class KPISheetService:
"""KPI 表單服務"""
def __init__(self, db: Session):
self.db = db
# ==================== 查詢 ====================
def get_by_id(self, sheet_id: int) -> Optional[KPISheet]:
"""根據 ID 取得表單"""
return self.db.query(KPISheet).filter(KPISheet.id == sheet_id).first()
def get_by_employee_period(
self, employee_id: int, period_id: int
) -> Optional[KPISheet]:
"""根據員工和期間取得表單"""
return (
self.db.query(KPISheet)
.filter(
KPISheet.employee_id == employee_id, KPISheet.period_id == period_id
)
.first()
)
def get_multi(
self,
period_id: Optional[int] = None,
department_id: Optional[int] = None,
status: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[KPISheet]:
"""查詢多筆表單"""
query = self.db.query(KPISheet)
if period_id:
query = query.filter(KPISheet.period_id == period_id)
if department_id:
query = query.filter(KPISheet.department_id == department_id)
if status:
query = query.filter(KPISheet.status == status)
return query.offset(skip).limit(limit).all()
def get_my_sheets(self, employee_id: int) -> List[KPISheet]:
"""取得我的 KPI 表單"""
return (
self.db.query(KPISheet)
.filter(KPISheet.employee_id == employee_id)
.order_by(KPISheet.created_at.desc())
.all()
)
def get_pending_for_manager(self, manager_id: int) -> List[KPISheet]:
"""取得待主管審核的表單"""
return (
self.db.query(KPISheet)
.join(Employee, KPISheet.employee_id == Employee.id)
.filter(
Employee.manager_id == manager_id,
KPISheet.status == KPISheetStatus.PENDING,
)
.all()
)
# ==================== 建立 ====================
def create(self, employee: Employee, data: KPISheetCreate) -> KPISheet:
"""建立 KPI 表單"""
# 檢查期間是否在設定期間
period = self.db.query(KPIPeriod).filter(KPIPeriod.id == data.period_id).first()
if not period:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "PERIOD001", "message": "期間不存在"},
)
if period.status != "setting":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"code": "PERIOD001", "message": "目前不在 KPI 設定期間"},
)
# 檢查是否已存在
existing = self.get_by_employee_period(employee.id, data.period_id)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={"code": "KPI005", "message": "該期間已有 KPI 表單"},
)
# 建立表單
sheet = KPISheet(
employee_id=employee.id,
period_id=data.period_id,
department_id=employee.department_id,
status=KPISheetStatus.DRAFT,
)
self.db.add(sheet)
self.db.flush()
# 建立項目
for i, item_data in enumerate(data.items):
item = KPIItem(
sheet_id=sheet.id,
template_id=item_data.template_id,
sort_order=i,
name=item_data.name,
category=item_data.category,
weight=item_data.weight,
level0_criteria=item_data.level0_criteria,
level1_criteria=item_data.level1_criteria,
level2_criteria=item_data.level2_criteria,
level3_criteria=item_data.level3_criteria,
level4_criteria=item_data.level4_criteria,
)
self.db.add(item)
self.db.commit()
self.db.refresh(sheet)
return sheet
def copy_from_previous(self, employee: Employee, period_id: int) -> KPISheet:
"""從上期複製 KPI"""
# 找到上一期表單
prev_sheet = (
self.db.query(KPISheet)
.filter(
KPISheet.employee_id == employee.id,
KPISheet.period_id != period_id,
KPISheet.status == KPISheetStatus.COMPLETED,
)
.order_by(KPISheet.created_at.desc())
.first()
)
if not prev_sheet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "KPI006", "message": "找不到上期 KPI 可供複製"},
)
# 轉換為建立資料
items = [
KPIItemCreate(
template_id=item.template_id,
name=item.name,
category=item.category,
weight=item.weight,
level0_criteria=item.level0_criteria,
level1_criteria=item.level1_criteria,
level2_criteria=item.level2_criteria,
level3_criteria=item.level3_criteria,
level4_criteria=item.level4_criteria,
)
for item in prev_sheet.items
]
data = KPISheetCreate(period_id=period_id, items=items)
return self.create(employee, data)
def apply_preset(
self, employee: Employee, period_id: int, preset_id: int
) -> KPISheet:
"""套用預設組合"""
preset = self.db.query(KPIPreset).filter(KPIPreset.id == preset_id).first()
if not preset:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "PRESET001", "message": "預設組合不存在"},
)
# 轉換為建立資料
items = []
for preset_item in preset.items:
template = preset_item.template
items.append(
KPIItemCreate(
template_id=template.id,
name=template.name,
category=template.category,
weight=preset_item.default_weight,
level0_criteria=template.level0_desc,
level1_criteria=template.level1_desc,
level2_criteria=template.level2_desc,
level3_criteria=template.level3_desc,
level4_criteria=template.level4_desc,
)
)
data = KPISheetCreate(period_id=period_id, items=items)
return self.create(employee, data)
# ==================== 驗證 ====================
def validate_weight(self, sheet_id: int) -> bool:
"""驗證權重總和是否等於 100"""
items = self.db.query(KPIItem).filter(KPIItem.sheet_id == sheet_id).all()
total_weight = sum(item.weight for item in items)
return total_weight == 100
def _check_status_transition(self, sheet: KPISheet, new_status: str) -> None:
"""檢查狀態轉換是否合法"""
if not sheet.can_transition_to(new_status):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"code": "KPI003",
"message": f"無法從 {sheet.status} 轉換到 {new_status}",
},
)
def _log_action(
self,
sheet: KPISheet,
action: str,
actor: Employee,
from_status: str,
to_status: str,
comment: Optional[str] = None,
) -> None:
"""記錄審核動作"""
log = KPIReviewLog(
sheet_id=sheet.id,
action=action,
actor_id=actor.id,
from_status=from_status,
to_status=to_status,
comment=comment,
)
self.db.add(log)
# ==================== 審核流程 ====================
def submit(self, sheet: KPISheet, actor: Employee) -> KPISheet:
"""提交 KPI"""
from_status = sheet.status
self._check_status_transition(sheet, KPISheetStatus.PENDING)
# 檢查權重
if not self.validate_weight(sheet.id):
total = sum(item.weight for item in sheet.items)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"code": "KPI001",
"message": "權重總和必須等於 100%",
"details": {"current_total": total, "expected": 100},
},
)
# 檢查項目數量
if len(sheet.items) < 3:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"code": "KPI002",
"message": "KPI 項目數量不足,至少需要 3 項",
},
)
sheet.status = KPISheetStatus.PENDING
sheet.submitted_at = datetime.utcnow()
self._log_action(sheet, "submit", actor, from_status, sheet.status)
self.db.commit()
self.db.refresh(sheet)
return sheet
def approve(
self, sheet: KPISheet, actor: Employee, comment: Optional[str] = None
) -> KPISheet:
"""審核通過"""
from_status = sheet.status
self._check_status_transition(sheet, KPISheetStatus.APPROVED)
sheet.status = KPISheetStatus.APPROVED
sheet.approved_by = actor.id
sheet.approved_at = datetime.utcnow()
sheet.approve_comment = comment
self._log_action(sheet, "approve", actor, from_status, sheet.status, comment)
self.db.commit()
self.db.refresh(sheet)
return sheet
def reject(self, sheet: KPISheet, actor: Employee, reason: str) -> KPISheet:
"""退回"""
from_status = sheet.status
self._check_status_transition(sheet, KPISheetStatus.DRAFT)
sheet.status = KPISheetStatus.DRAFT
sheet.rejected_by = actor.id
sheet.rejected_at = datetime.utcnow()
sheet.reject_reason = reason
self._log_action(sheet, "reject", actor, from_status, sheet.status, reason)
self.db.commit()
self.db.refresh(sheet)
return sheet
# ==================== 評核 ====================
def self_eval(
self, sheet: KPISheet, actor: Employee, data: SelfEvalRequest
) -> KPISheet:
"""員工自評"""
from_status = sheet.status
# 如果是 approved 狀態,先轉換到 self_eval
if sheet.status == KPISheetStatus.APPROVED:
sheet.status = KPISheetStatus.SELF_EVAL
self._check_status_transition(sheet, KPISheetStatus.MANAGER_EVAL)
# 更新項目自評
item_map = {item.id: item for item in sheet.items}
for eval_item in data.items:
item = item_map.get(eval_item.id)
if item:
item.self_eval_level = eval_item.level
item.self_eval_note = eval_item.note
sheet.status = KPISheetStatus.MANAGER_EVAL
sheet.self_eval_at = datetime.utcnow()
self._log_action(sheet, "self_eval", actor, from_status, sheet.status)
self.db.commit()
self.db.refresh(sheet)
return sheet
def manager_eval(
self, sheet: KPISheet, actor: Employee, data: ManagerEvalRequest
) -> KPISheet:
"""主管評核"""
from_status = sheet.status
self._check_status_transition(sheet, KPISheetStatus.COMPLETED)
# 更新項目評核
item_map = {item.id: item for item in sheet.items}
for eval_item in data.items:
item = item_map.get(eval_item.id)
if item:
item.final_level = eval_item.level
item.final_note = eval_item.note
# 計算總分
total_score = Decimal("0")
for item in sheet.items:
if item.final_level is not None:
weight_ratio = Decimal(str(item.weight)) / Decimal("100")
level_month = LEVEL_MONTHS.get(item.final_level, Decimal("0"))
total_score += weight_ratio * level_month
sheet.status = KPISheetStatus.COMPLETED
sheet.manager_eval_by = actor.id
sheet.manager_eval_at = datetime.utcnow()
sheet.manager_eval_comment = data.comment
sheet.total_score = total_score
self._log_action(
sheet, "manager_eval", actor, from_status, sheet.status, data.comment
)
self.db.commit()
self.db.refresh(sheet)
return sheet
# ==================== 刪除 ====================
def delete(self, sheet: KPISheet) -> None:
"""刪除表單(只能刪除草稿)"""
if sheet.status != KPISheetStatus.DRAFT:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"code": "KPI007", "message": "只能刪除草稿狀態的表單"},
)
self.db.delete(sheet)
self.db.commit()

173
app/services/llm_service.py Normal file
View File

@@ -0,0 +1,173 @@
"""
Ollama LLM API 服務
"""
import json
from typing import Optional, List, Generator
import requests
from app.core.config import settings
class LLMService:
"""Ollama LLM API 服務"""
def __init__(self):
self.api_url = settings.OLLAMA_API_URL
self.default_model = settings.OLLAMA_DEFAULT_MODEL
def list_models(self) -> List[str]:
"""
列出可用模型
Returns:
模型 ID 列表
"""
try:
response = requests.get(f"{self.api_url}/v1/models", timeout=10)
response.raise_for_status()
models = response.json()
return [m["id"] for m in models.get("data", [])]
except Exception as e:
return [f"Error: {str(e)}"]
def chat(
self,
messages: List[dict],
model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: Optional[int] = None,
) -> str:
"""
聊天完成請求
Args:
messages: 對話訊息列表 [{"role": "user", "content": "..."}]
model: 模型名稱,預設使用設定檔中的模型
temperature: 溫度參數 (0-1)
max_tokens: 最大 token 數
Returns:
AI 回應內容
"""
chat_request = {
"model": model or self.default_model,
"messages": messages,
"temperature": temperature,
}
if max_tokens:
chat_request["max_tokens"] = max_tokens
try:
response = requests.post(
f"{self.api_url}/v1/chat/completions",
json=chat_request,
timeout=60,
)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
except Exception as e:
return f"Error: {str(e)}"
def chat_stream(
self,
messages: List[dict],
model: Optional[str] = None,
temperature: float = 0.7,
) -> Generator[str, None, None]:
"""
串流聊天請求
Args:
messages: 對話訊息列表
model: 模型名稱
temperature: 溫度參數
Yields:
AI 回應內容片段
"""
chat_request = {
"model": model or self.default_model,
"messages": messages,
"temperature": temperature,
"stream": True,
}
try:
response = requests.post(
f"{self.api_url}/v1/chat/completions",
json=chat_request,
stream=True,
timeout=120,
)
response.raise_for_status()
for line in response.iter_lines():
if line:
if line.startswith(b"data: "):
data_str = line[6:].decode("utf-8")
if data_str.strip() != "[DONE]":
try:
data = json.loads(data_str)
if "choices" in data:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
yield delta["content"]
except json.JSONDecodeError:
continue
except Exception as e:
yield f"Error: {str(e)}"
def simple_ask(
self,
question: str,
system_prompt: str = "You are a helpful assistant.",
model: Optional[str] = None,
) -> str:
"""
簡單問答
Args:
question: 使用者問題
system_prompt: 系統提示詞
model: 模型名稱
Returns:
AI 回應
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": question},
]
return self.chat(messages, model=model)
def analyze_kpi(self, kpi_data: dict) -> str:
"""
分析 KPI 數據
Args:
kpi_data: KPI 相關數據
Returns:
AI 分析結果
"""
system_prompt = """你是一位專業的 KPI 分析師。
請根據提供的 KPI 數據,給出專業的分析和建議。
回應請使用繁體中文,並保持專業且易懂。"""
question = f"""請分析以下 KPI 數據:
{json.dumps(kpi_data, ensure_ascii=False, indent=2)}
請提供:
1. 數據摘要
2. 表現評估
3. 改善建議"""
return self.simple_ask(question, system_prompt)
# 單例
llm_service = LLMService()

View File

@@ -0,0 +1,188 @@
"""
通知服務
"""
from datetime import datetime
from typing import Optional, List
from sqlalchemy.orm import Session
from app.models.notification import Notification, NotificationPreference
from app.models.employee import Employee
from app.models.kpi_sheet import KPISheet
class NotifyService:
"""通知服務"""
def __init__(self, db: Session):
self.db = db
def create(
self,
recipient_id: int,
type: str,
title: str,
content: Optional[str] = None,
related_sheet_id: Optional[int] = None,
) -> Notification:
"""建立通知"""
notification = Notification(
recipient_id=recipient_id,
type=type,
title=title,
content=content,
related_sheet_id=related_sheet_id,
)
self.db.add(notification)
self.db.commit()
self.db.refresh(notification)
return notification
def get_by_recipient(
self, recipient_id: int, skip: int = 0, limit: int = 50
) -> List[Notification]:
"""取得通知列表"""
return (
self.db.query(Notification)
.filter(Notification.recipient_id == recipient_id)
.order_by(Notification.created_at.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_unread_count(self, recipient_id: int) -> int:
"""取得未讀數量"""
return (
self.db.query(Notification)
.filter(
Notification.recipient_id == recipient_id,
Notification.is_read == False,
)
.count()
)
def mark_as_read(self, notification_id: int, recipient_id: int) -> bool:
"""標記已讀"""
notification = (
self.db.query(Notification)
.filter(
Notification.id == notification_id,
Notification.recipient_id == recipient_id,
)
.first()
)
if not notification:
return False
notification.is_read = True
notification.read_at = datetime.utcnow()
self.db.commit()
return True
def mark_all_as_read(self, recipient_id: int) -> int:
"""全部標記已讀"""
result = (
self.db.query(Notification)
.filter(
Notification.recipient_id == recipient_id,
Notification.is_read == False,
)
.update({"is_read": True, "read_at": datetime.utcnow()})
)
self.db.commit()
return result
def get_preferences(self, employee_id: int) -> Optional[NotificationPreference]:
"""取得通知偏好"""
return (
self.db.query(NotificationPreference)
.filter(NotificationPreference.employee_id == employee_id)
.first()
)
def update_preferences(
self,
employee_id: int,
email_enabled: Optional[bool] = None,
in_app_enabled: Optional[bool] = None,
reminder_days_before: Optional[int] = None,
) -> NotificationPreference:
"""更新通知偏好"""
pref = self.get_preferences(employee_id)
if not pref:
pref = NotificationPreference(employee_id=employee_id)
self.db.add(pref)
if email_enabled is not None:
pref.email_enabled = email_enabled
if in_app_enabled is not None:
pref.in_app_enabled = in_app_enabled
if reminder_days_before is not None:
pref.reminder_days_before = reminder_days_before
self.db.commit()
self.db.refresh(pref)
return pref
# ==================== 事件通知 ====================
def notify_kpi_submitted(self, sheet: KPISheet) -> None:
"""通知 KPI 已提交"""
employee = sheet.employee
if not employee.manager_id:
return
self.create(
recipient_id=employee.manager_id,
type="kpi_submitted",
title="KPI 待審核",
content=f"{employee.name}{sheet.period.code} KPI 已提交,請審核。",
related_sheet_id=sheet.id,
)
def notify_kpi_approved(self, sheet: KPISheet) -> None:
"""通知 KPI 已核准"""
self.create(
recipient_id=sheet.employee_id,
type="kpi_approved",
title="KPI 已核准",
content=f"您的 {sheet.period.code} KPI 已審核通過。",
related_sheet_id=sheet.id,
)
def notify_kpi_rejected(self, sheet: KPISheet, reason: str) -> None:
"""通知 KPI 已退回"""
self.create(
recipient_id=sheet.employee_id,
type="kpi_rejected",
title="KPI 已退回",
content=f"您的 {sheet.period.code} KPI 已退回,原因:{reason}",
related_sheet_id=sheet.id,
)
def notify_self_eval_completed(self, sheet: KPISheet) -> None:
"""通知自評已完成"""
employee = sheet.employee
if not employee.manager_id:
return
self.create(
recipient_id=employee.manager_id,
type="self_eval_completed",
title="員工自評已完成",
content=f"{employee.name}{sheet.period.code} 自評已完成,請進行覆核。",
related_sheet_id=sheet.id,
)
def notify_manager_eval_completed(self, sheet: KPISheet) -> None:
"""通知主管評核已完成"""
self.create(
recipient_id=sheet.employee_id,
type="manager_eval_completed",
title="KPI 評核已完成",
content=f"您的 {sheet.period.code} KPI 評核已完成,獎金月數:{sheet.total_score}",
related_sheet_id=sheet.id,
)

337
ddl/01_create_tables.sql Normal file
View File

@@ -0,0 +1,337 @@
-- KPI 管理系統 - 資料庫建立腳本
-- 版本: 1.0
-- 建立日期: 2024-12
-- ============================================
-- 1. 部門表 (departments)
-- ============================================
CREATE TABLE departments (
id SERIAL PRIMARY KEY,
code VARCHAR(20) NOT NULL UNIQUE,
name VARCHAR(100) NOT NULL,
level VARCHAR(20) NOT NULL DEFAULT 'DEPT', -- COMPANY, BU, DEPT, TEAM
parent_id INTEGER REFERENCES departments(id),
is_active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_departments_parent ON departments(parent_id);
CREATE INDEX idx_departments_code ON departments(code);
COMMENT ON TABLE departments IS '部門組織表';
COMMENT ON COLUMN departments.level IS '層級: COMPANY=公司, BU=事業單位, DEPT=部門, TEAM=團隊';
-- ============================================
-- 2. 員工表 (employees)
-- ============================================
CREATE TABLE employees (
id SERIAL PRIMARY KEY,
employee_no VARCHAR(20) NOT NULL UNIQUE,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL UNIQUE,
password_hash VARCHAR(200) NOT NULL,
department_id INTEGER NOT NULL REFERENCES departments(id),
manager_id INTEGER REFERENCES employees(id),
job_title VARCHAR(100),
role VARCHAR(20) NOT NULL DEFAULT 'employee', -- employee, manager, admin, hr
status VARCHAR(20) NOT NULL DEFAULT 'active', -- active, inactive, resigned
hire_date DATE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_employees_department ON employees(department_id);
CREATE INDEX idx_employees_manager ON employees(manager_id);
CREATE INDEX idx_employees_status ON employees(status);
COMMENT ON TABLE employees IS '員工表';
COMMENT ON COLUMN employees.role IS '角色: employee=員工, manager=主管, admin=管理員, hr=人資';
-- ============================================
-- 3. KPI 期間表 (kpi_periods)
-- ============================================
CREATE TABLE kpi_periods (
id SERIAL PRIMARY KEY,
code VARCHAR(20) NOT NULL UNIQUE, -- 例: 2024H1, 2024H2
name VARCHAR(100) NOT NULL,
start_date DATE NOT NULL,
end_date DATE NOT NULL,
setting_start DATE NOT NULL, -- KPI 設定開始日
setting_end DATE NOT NULL, -- KPI 設定截止日
self_eval_start DATE, -- 自評開始日
self_eval_end DATE, -- 自評截止日
manager_eval_start DATE, -- 主管評核開始日
manager_eval_end DATE, -- 主管評核截止日
status VARCHAR(20) NOT NULL DEFAULT 'draft', -- draft, setting, approved, self_eval, manager_eval, completed
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_kpi_periods_status ON kpi_periods(status);
COMMENT ON TABLE kpi_periods IS 'KPI 期間表';
COMMENT ON COLUMN kpi_periods.status IS '狀態: draft=草稿, setting=設定中, approved=已核准, self_eval=自評中, manager_eval=主管評核中, completed=已完成';
-- ============================================
-- 4. KPI 範本表 (kpi_templates)
-- ============================================
CREATE TABLE kpi_templates (
id SERIAL PRIMARY KEY,
code VARCHAR(50) NOT NULL UNIQUE,
name VARCHAR(200) NOT NULL,
category VARCHAR(50) NOT NULL, -- financial, customer, internal, learning
description TEXT,
default_weight INTEGER DEFAULT 20,
level0_desc TEXT NOT NULL, -- 等級 0 說明
level1_desc TEXT NOT NULL, -- 等級 1 說明
level2_desc TEXT NOT NULL, -- 等級 2 說明
level3_desc TEXT NOT NULL, -- 等級 3 說明
level4_desc TEXT NOT NULL, -- 等級 4 說明
applicable_roles TEXT[], -- 適用角色
applicable_depts INTEGER[], -- 適用部門 ID
is_active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_kpi_templates_category ON kpi_templates(category);
CREATE INDEX idx_kpi_templates_active ON kpi_templates(is_active);
COMMENT ON TABLE kpi_templates IS 'KPI 範本表';
COMMENT ON COLUMN kpi_templates.category IS '類別: financial=財務, customer=客戶, internal=內部流程, learning=學習成長';
-- ============================================
-- 5. KPI 預設組合表 (kpi_presets)
-- ============================================
CREATE TABLE kpi_presets (
id SERIAL PRIMARY KEY,
code VARCHAR(50) NOT NULL UNIQUE,
name VARCHAR(200) NOT NULL,
description TEXT,
applicable_roles TEXT[],
applicable_depts INTEGER[],
is_active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
COMMENT ON TABLE kpi_presets IS 'KPI 預設組合表';
-- ============================================
-- 6. KPI 預設項目表 (kpi_preset_items)
-- ============================================
CREATE TABLE kpi_preset_items (
id SERIAL PRIMARY KEY,
preset_id INTEGER NOT NULL REFERENCES kpi_presets(id) ON DELETE CASCADE,
template_id INTEGER NOT NULL REFERENCES kpi_templates(id),
default_weight INTEGER NOT NULL,
is_mandatory BOOLEAN NOT NULL DEFAULT FALSE,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_kpi_preset_items_preset ON kpi_preset_items(preset_id);
COMMENT ON TABLE kpi_preset_items IS 'KPI 預設項目表';
-- ============================================
-- 7. KPI 表單表 (kpi_sheets)
-- ============================================
CREATE TABLE kpi_sheets (
id SERIAL PRIMARY KEY,
employee_id INTEGER NOT NULL REFERENCES employees(id),
period_id INTEGER NOT NULL REFERENCES kpi_periods(id),
department_id INTEGER NOT NULL REFERENCES departments(id),
status VARCHAR(20) NOT NULL DEFAULT 'draft', -- draft, pending, approved, self_eval, manager_eval, completed, settled
-- 提交資訊
submitted_at TIMESTAMP,
-- 審核資訊
approved_by INTEGER REFERENCES employees(id),
approved_at TIMESTAMP,
approve_comment TEXT,
-- 退回資訊
rejected_by INTEGER REFERENCES employees(id),
rejected_at TIMESTAMP,
reject_reason TEXT,
-- 自評資訊
self_eval_at TIMESTAMP,
-- 主管評核資訊
manager_eval_by INTEGER REFERENCES employees(id),
manager_eval_at TIMESTAMP,
manager_eval_comment TEXT,
-- 分數
total_score DECIMAL(5,4), -- 總獎金月數 (0~1)
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(employee_id, period_id)
);
CREATE INDEX idx_kpi_sheets_employee ON kpi_sheets(employee_id);
CREATE INDEX idx_kpi_sheets_period ON kpi_sheets(period_id);
CREATE INDEX idx_kpi_sheets_department ON kpi_sheets(department_id);
CREATE INDEX idx_kpi_sheets_status ON kpi_sheets(status);
COMMENT ON TABLE kpi_sheets IS 'KPI 表單表';
COMMENT ON COLUMN kpi_sheets.status IS '狀態: draft=草稿, pending=待審核, approved=已核准, self_eval=自評中, manager_eval=主管評核中, completed=已完成, settled=已結算';
-- ============================================
-- 8. KPI 項目表 (kpi_items)
-- ============================================
CREATE TABLE kpi_items (
id SERIAL PRIMARY KEY,
sheet_id INTEGER NOT NULL REFERENCES kpi_sheets(id) ON DELETE CASCADE,
template_id INTEGER REFERENCES kpi_templates(id),
sort_order INTEGER NOT NULL DEFAULT 0,
-- 項目資訊
name VARCHAR(200) NOT NULL,
category VARCHAR(50) NOT NULL,
weight INTEGER NOT NULL, -- 權重百分比 (1-100)
-- 等級標準
level0_criteria TEXT,
level1_criteria TEXT,
level2_criteria TEXT,
level3_criteria TEXT,
level4_criteria TEXT,
-- 自評
self_eval_level INTEGER, -- 0-4
self_eval_note TEXT,
-- 主管評核
final_level INTEGER, -- 0-4 (最終等級)
final_note TEXT,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_kpi_items_sheet ON kpi_items(sheet_id);
COMMENT ON TABLE kpi_items IS 'KPI 項目表';
COMMENT ON COLUMN kpi_items.weight IS '權重百分比,同一 sheet 的所有項目權重總和必須等於 100';
-- ============================================
-- 9. KPI 指派表 (kpi_assignments)
-- ============================================
CREATE TABLE kpi_assignments (
id SERIAL PRIMARY KEY,
employee_id INTEGER NOT NULL REFERENCES employees(id),
period_id INTEGER NOT NULL REFERENCES kpi_periods(id),
assigned_by INTEGER NOT NULL REFERENCES employees(id),
items JSONB NOT NULL, -- 指派的 KPI 項目 JSON
note TEXT,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(employee_id, period_id)
);
CREATE INDEX idx_kpi_assignments_employee ON kpi_assignments(employee_id);
COMMENT ON TABLE kpi_assignments IS '主管指派 KPI 表';
-- ============================================
-- 10. KPI 審核紀錄表 (kpi_review_logs)
-- ============================================
CREATE TABLE kpi_review_logs (
id SERIAL PRIMARY KEY,
sheet_id INTEGER NOT NULL REFERENCES kpi_sheets(id) ON DELETE CASCADE,
action VARCHAR(50) NOT NULL, -- submit, approve, reject, self_eval, manager_eval
actor_id INTEGER NOT NULL REFERENCES employees(id),
from_status VARCHAR(20),
to_status VARCHAR(20),
comment TEXT,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_kpi_review_logs_sheet ON kpi_review_logs(sheet_id);
CREATE INDEX idx_kpi_review_logs_actor ON kpi_review_logs(actor_id);
COMMENT ON TABLE kpi_review_logs IS 'KPI 審核紀錄表';
-- ============================================
-- 11. 通知表 (notifications)
-- ============================================
CREATE TABLE notifications (
id SERIAL PRIMARY KEY,
recipient_id INTEGER NOT NULL REFERENCES employees(id),
type VARCHAR(50) NOT NULL, -- kpi_submitted, kpi_approved, kpi_rejected, eval_reminder, etc.
title VARCHAR(200) NOT NULL,
content TEXT,
related_sheet_id INTEGER REFERENCES kpi_sheets(id) ON DELETE SET NULL,
is_read BOOLEAN NOT NULL DEFAULT FALSE,
read_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_notifications_recipient ON notifications(recipient_id);
CREATE INDEX idx_notifications_unread ON notifications(recipient_id, is_read) WHERE is_read = FALSE;
COMMENT ON TABLE notifications IS '通知表';
-- ============================================
-- 12. 通知偏好表 (notification_preferences)
-- ============================================
CREATE TABLE notification_preferences (
id SERIAL PRIMARY KEY,
employee_id INTEGER NOT NULL UNIQUE REFERENCES employees(id),
email_enabled BOOLEAN NOT NULL DEFAULT TRUE,
in_app_enabled BOOLEAN NOT NULL DEFAULT TRUE,
reminder_days_before INTEGER NOT NULL DEFAULT 3,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
COMMENT ON TABLE notification_preferences IS '通知偏好表';
-- ============================================
-- 13. 儀表板警示表 (dashboard_alerts)
-- ============================================
CREATE TABLE dashboard_alerts (
id SERIAL PRIMARY KEY,
alert_type VARCHAR(50) NOT NULL, -- deadline_approaching, overdue, weight_invalid
severity VARCHAR(20) NOT NULL DEFAULT 'warning', -- info, warning, error
title VARCHAR(200) NOT NULL,
description TEXT,
related_sheet_id INTEGER REFERENCES kpi_sheets(id) ON DELETE CASCADE,
related_employee_id INTEGER REFERENCES employees(id),
is_resolved BOOLEAN NOT NULL DEFAULT FALSE,
resolved_at TIMESTAMP,
resolved_by INTEGER REFERENCES employees(id),
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_dashboard_alerts_unresolved ON dashboard_alerts(is_resolved) WHERE is_resolved = FALSE;
COMMENT ON TABLE dashboard_alerts IS '儀表板警示表';
-- ============================================
-- 更新時間觸發器
-- ============================================
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = CURRENT_TIMESTAMP;
RETURN NEW;
END;
$$ language 'plpgsql';
-- 為需要 updated_at 的表建立觸發器
CREATE TRIGGER update_departments_updated_at BEFORE UPDATE ON departments FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_employees_updated_at BEFORE UPDATE ON employees FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_kpi_periods_updated_at BEFORE UPDATE ON kpi_periods FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_kpi_templates_updated_at BEFORE UPDATE ON kpi_templates FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_kpi_presets_updated_at BEFORE UPDATE ON kpi_presets FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_kpi_sheets_updated_at BEFORE UPDATE ON kpi_sheets FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_kpi_items_updated_at BEFORE UPDATE ON kpi_items FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_notification_preferences_updated_at BEFORE UPDATE ON notification_preferences FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();

125
ddl/02_seed_data.sql Normal file
View File

@@ -0,0 +1,125 @@
-- KPI 管理系統 - 種子資料
-- 版本: 1.0
-- 建立日期: 2024-12
-- ============================================
-- 1. 部門資料
-- ============================================
INSERT INTO departments (code, name, level, parent_id) VALUES
('COMPANY', '總公司', 'COMPANY', NULL),
('BU_TECH', '技術事業部', 'BU', 1),
('BU_SALES', '業務事業部', 'BU', 1),
('DEPT_RD', '研發部', 'DEPT', 2),
('DEPT_QA', '品保部', 'DEPT', 2),
('DEPT_SALES', '業務部', 'DEPT', 3),
('DEPT_MARKETING', '行銷部', 'DEPT', 3),
('DEPT_HR', '人力資源部', 'DEPT', 1),
('DEPT_FINANCE', '財務部', 'DEPT', 1);
-- ============================================
-- 2. 員工資料 (密碼: password123)
-- bcrypt hash for 'password123'
-- ============================================
INSERT INTO employees (employee_no, name, email, password_hash, department_id, manager_id, job_title, role, status, hire_date) VALUES
-- 管理層
('EMP00001', '王大明', 'admin@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 1, NULL, '總經理', 'admin', 'active', '2020-01-01'),
-- 技術事業部
('EMP00002', '陳志強', 'chen.zhiqiang@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 2, 1, '技術長', 'manager', 'active', '2020-03-15'),
('EMP00003', '林小華', 'lin.xiaohua@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 4, 2, '研發經理', 'manager', 'active', '2021-01-10'),
('EMP00004', '張美玲', 'zhang.meiling@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 4, 3, '資深工程師', 'employee', 'active', '2021-06-01'),
('EMP00005', '李建國', 'li.jianguo@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 4, 3, '工程師', 'employee', 'active', '2022-03-01'),
('EMP00006', '黃雅琪', 'huang.yaqi@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 5, 2, '品保經理', 'manager', 'active', '2021-02-15'),
('EMP00007', '吳宗翰', 'wu.zonghan@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 5, 6, 'QA工程師', 'employee', 'active', '2022-08-01'),
-- 業務事業部
('EMP00008', '趙文傑', 'zhao.wenjie@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 3, 1, '業務長', 'manager', 'active', '2020-05-01'),
('EMP00009', '周曉明', 'zhou.xiaoming@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 6, 8, '業務經理', 'manager', 'active', '2021-04-01'),
('EMP00010', '蔡佳蓉', 'cai.jiarong@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 6, 9, '業務專員', 'employee', 'active', '2022-01-15'),
-- 人資部
('EMP00011', '許淑芬', 'xu.shufen@example.com', '$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.yHvLt7c2bF5Eni', 8, 1, '人資經理', 'hr', 'active', '2020-06-01');
-- ============================================
-- 3. KPI 期間資料
-- ============================================
INSERT INTO kpi_periods (code, name, start_date, end_date, setting_start, setting_end, self_eval_start, self_eval_end, manager_eval_start, manager_eval_end, status) VALUES
('2024H1', '2024年上半年', '2024-01-01', '2024-06-30', '2024-01-01', '2024-01-14', '2024-06-15', '2024-06-25', '2024-06-26', '2024-06-30', 'completed'),
('2024H2', '2024年下半年', '2024-07-01', '2024-12-31', '2024-07-01', '2024-07-14', '2024-12-15', '2024-12-25', '2024-12-26', '2024-12-31', 'setting'),
('2025H1', '2025年上半年', '2025-01-01', '2025-06-30', '2025-01-01', '2025-01-14', '2025-06-15', '2025-06-25', '2025-06-26', '2025-06-30', 'draft');
-- ============================================
-- 4. KPI 範本資料
-- ============================================
INSERT INTO kpi_templates (code, name, category, description, default_weight, level0_desc, level1_desc, level2_desc, level3_desc, level4_desc) VALUES
-- 財務類
('FIN001', '單位製造成本降低率', 'financial', '製造成本控制指標', 25,
'成本反升或降低 <2%', '降低 2.0%-2.9%', '降低 3.0%', '降低 3.1%-3.9%', '降低 ≥4.0%'),
('FIN002', '營收達成率', 'financial', '營收目標達成指標', 30,
'達成率 <80%', '達成率 80%-89%', '達成率 90%-99%', '達成率 100%-109%', '達成率 ≥110%'),
('FIN003', '毛利率提升', 'financial', '毛利率改善指標', 20,
'毛利率下降', '維持不變', '提升 0.1%-0.5%', '提升 0.6%-1.0%', '提升 >1.0%'),
-- 客戶類
('CUS001', '客戶滿意度', 'customer', '客戶滿意度調查結果', 25,
'滿意度 <70%', '滿意度 70%-79%', '滿意度 80%-89%', '滿意度 90%-94%', '滿意度 ≥95%'),
('CUS002', '客訴處理時效', 'customer', '客訴回應與解決時間', 15,
'平均 >72 小時', '平均 48-72 小時', '平均 24-48 小時', '平均 12-24 小時', '平均 <12 小時'),
('CUS003', '新客戶開發數', 'customer', '新客戶獲取數量', 20,
'0 家', '1-2 家', '3-4 家', '5-6 家', '≥7 家'),
-- 內部流程類
('INT001', '專案準時完成率', 'internal', '專案如期交付比率', 25,
'準時率 <70%', '準時率 70%-79%', '準時率 80%-89%', '準時率 90%-94%', '準時率 ≥95%'),
('INT002', '流程改善提案', 'internal', '提出並執行的流程改善', 15,
'0 件', '1 件', '2 件', '3 件', '≥4 件'),
('INT003', '系統穩定度', 'internal', '系統可用率指標', 20,
'可用率 <95%', '可用率 95%-96.9%', '可用率 97%-98.4%', '可用率 98.5%-99.4%', '可用率 ≥99.5%'),
-- 學習成長類
('LRN001', '專業證照取得', 'learning', '取得專業認證', 15,
'未取得', '考試中', '取得 1 張', '取得 2 張', '取得 ≥3 張'),
('LRN002', '內部訓練時數', 'learning', '參與內部培訓時數', 10,
'<10 小時', '10-19 小時', '20-29 小時', '30-39 小時', '≥40 小時'),
('LRN003', '知識分享場次', 'learning', '進行內部知識分享', 10,
'0 場', '1 場', '2 場', '3 場', '≥4 場');
-- ============================================
-- 5. KPI 預設組合資料
-- ============================================
INSERT INTO kpi_presets (code, name, description, applicable_roles) VALUES
('PRESET_ENG', '工程師標準組合', '適用於研發、品保工程師', ARRAY['employee']),
('PRESET_SALES', '業務人員標準組合', '適用於業務專員', ARRAY['employee']),
('PRESET_MGR', '主管標準組合', '適用於部門主管', ARRAY['manager']);
-- ============================================
-- 6. KPI 預設項目資料
-- ============================================
-- 工程師組合
INSERT INTO kpi_preset_items (preset_id, template_id, default_weight, is_mandatory, sort_order) VALUES
(1, 7, 30, TRUE, 1), -- 專案準時完成率
(1, 9, 25, TRUE, 2), -- 系統穩定度
(1, 8, 15, FALSE, 3), -- 流程改善提案
(1, 10, 15, FALSE, 4), -- 專業證照取得
(1, 12, 15, FALSE, 5); -- 知識分享場次
-- 業務人員組合
INSERT INTO kpi_preset_items (preset_id, template_id, default_weight, is_mandatory, sort_order) VALUES
(2, 2, 40, TRUE, 1), -- 營收達成率
(2, 6, 25, TRUE, 2), -- 新客戶開發數
(2, 4, 20, FALSE, 3), -- 客戶滿意度
(2, 11, 15, FALSE, 4); -- 內部訓練時數
-- 主管組合
INSERT INTO kpi_preset_items (preset_id, template_id, default_weight, is_mandatory, sort_order) VALUES
(3, 2, 30, TRUE, 1), -- 營收達成率
(3, 7, 25, TRUE, 2), -- 專案準時完成率
(3, 4, 20, FALSE, 3), -- 客戶滿意度
(3, 8, 15, FALSE, 4), -- 流程改善提案
(3, 12, 10, FALSE, 5); -- 知識分享場次
-- ============================================
-- 7. 通知偏好預設資料
-- ============================================
INSERT INTO notification_preferences (employee_id, email_enabled, in_app_enabled, reminder_days_before)
SELECT id, TRUE, TRUE, 3 FROM employees;

34
requirements.txt Normal file
View File

@@ -0,0 +1,34 @@
# FastAPI
fastapi==0.109.0
uvicorn[standard]==0.27.0
# Database (MySQL)
sqlalchemy==2.0.25
pymysql==1.1.0
cryptography==41.0.7
alembic==1.13.1
# Authentication
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
# Validation & Settings
pydantic==2.5.3
pydantic-settings==2.1.0
python-multipart==0.0.6
email-validator==2.1.0
# Utilities
python-dotenv==1.0.0
requests==2.31.0
# Testing
pytest==7.4.4
pytest-asyncio==0.23.3
pytest-cov==4.1.0
httpx==0.26.0
faker==22.0.0
# Development
black==23.12.1
isort==5.13.2

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Tests

126
tests/conftest.py Normal file
View File

@@ -0,0 +1,126 @@
"""
共用測試 Fixture
"""
import pytest
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from fastapi.testclient import TestClient
from app.main import app
from app.core.database import Base, get_db
from app.core.security import create_access_token, get_password_hash
from app.models.employee import Employee
from app.models.department import Department
# 測試資料庫 URL (使用 SQLite 進行測試)
TEST_DATABASE_URL = "sqlite:///./test.db"
# 建立測試引擎
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="session", autouse=True)
def setup_database():
"""建立測試資料庫"""
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def db() -> Generator[Session, None, None]:
"""
每個測試函式使用獨立的資料庫 Session
測試結束後自動 rollback
"""
connection = engine.connect()
transaction = connection.begin()
session = TestingSessionLocal(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()
@pytest.fixture(scope="function")
def client(db: Session) -> Generator[TestClient, None, None]:
"""FastAPI 測試客戶端"""
def override_get_db():
yield db
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
@pytest.fixture
def test_department(db: Session) -> Department:
"""建立測試部門"""
dept = Department(
code="TEST_DEPT",
name="測試部門",
level="DEPT",
is_active=True,
)
db.add(dept)
db.flush()
return dept
@pytest.fixture
def test_employee(db: Session, test_department: Department) -> Employee:
"""建立測試員工"""
employee = Employee(
employee_no="TEST001",
name="測試員工",
email="test@example.com",
password_hash=get_password_hash("password123"),
department_id=test_department.id,
job_title="工程師",
role="employee",
status="active",
)
db.add(employee)
db.flush()
return employee
@pytest.fixture
def test_manager(db: Session, test_department: Department) -> Employee:
"""建立測試主管"""
manager = Employee(
employee_no="MGR001",
name="測試主管",
email="manager@example.com",
password_hash=get_password_hash("password123"),
department_id=test_department.id,
job_title="經理",
role="manager",
status="active",
)
db.add(manager)
db.flush()
return manager
@pytest.fixture
def auth_headers(test_employee: Employee) -> dict:
"""取得認證 Headers"""
token = create_access_token({"sub": str(test_employee.id)})
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def manager_headers(test_manager: Employee) -> dict:
"""取得主管認證 Headers"""
token = create_access_token({"sub": str(test_manager.id)})
return {"Authorization": f"Bearer {token}"}

204
tests/factories.py Normal file
View File

@@ -0,0 +1,204 @@
"""
測試資料工廠
"""
from datetime import date
from typing import Optional
from sqlalchemy.orm import Session
from app.core.security import get_password_hash
from app.models.department import Department
from app.models.employee import Employee
from app.models.kpi_period import KPIPeriod
from app.models.kpi_template import KPITemplate
from app.models.kpi_sheet import KPISheet
from app.models.kpi_item import KPIItem
class DepartmentFactory:
"""部門工廠"""
_counter = 0
@classmethod
def create(
cls,
db: Session,
code: Optional[str] = None,
name: Optional[str] = None,
level: str = "DEPT",
parent_id: Optional[int] = None,
) -> Department:
cls._counter += 1
dept = Department(
code=code or f"DEPT{cls._counter:03d}",
name=name or f"測試部門{cls._counter}",
level=level,
parent_id=parent_id,
is_active=True,
)
db.add(dept)
db.flush()
return dept
class EmployeeFactory:
"""員工工廠"""
_counter = 0
@classmethod
def create(
cls,
db: Session,
employee_no: Optional[str] = None,
name: Optional[str] = None,
department_id: Optional[int] = None,
manager_id: Optional[int] = None,
role: str = "employee",
status: str = "active",
) -> Employee:
cls._counter += 1
# 如果沒有指定部門,建立一個
if not department_id:
dept = DepartmentFactory.create(db)
department_id = dept.id
employee = Employee(
employee_no=employee_no or f"EMP{cls._counter:05d}",
name=name or f"測試員工{cls._counter}",
email=f"test{cls._counter}@example.com",
password_hash=get_password_hash("password123"),
department_id=department_id,
manager_id=manager_id,
job_title="工程師",
status=status,
role=role,
)
db.add(employee)
db.flush()
return employee
class KPIPeriodFactory:
"""KPI 期間工廠"""
_counter = 0
@classmethod
def create(
cls,
db: Session,
code: Optional[str] = None,
status: str = "setting",
year: int = 2024,
half: int = 2,
) -> KPIPeriod:
cls._counter += 1
if half == 1:
start = date(year, 1, 1)
end = date(year, 6, 30)
setting_end = date(year, 1, 14)
else:
start = date(year, 7, 1)
end = date(year, 12, 31)
setting_end = date(year, 7, 14)
period = KPIPeriod(
code=code or f"{year}H{half}_{cls._counter}",
name=f"{year}{'' if half == 1 else ''}半年",
start_date=start,
end_date=end,
setting_start=start,
setting_end=setting_end,
status=status,
)
db.add(period)
db.flush()
return period
class KPITemplateFactory:
"""KPI 範本工廠"""
_counter = 0
@classmethod
def create(
cls,
db: Session,
code: Optional[str] = None,
name: Optional[str] = None,
category: str = "financial",
) -> KPITemplate:
cls._counter += 1
template = KPITemplate(
code=code or f"TPL{cls._counter:03d}",
name=name or f"測試 KPI 範本 {cls._counter}",
category=category,
default_weight=20,
level0_desc="未達標",
level1_desc="基本達成",
level2_desc="達成目標",
level3_desc="挑戰目標",
level4_desc="超越目標",
is_active=True,
)
db.add(template)
db.flush()
return template
class KPISheetFactory:
"""KPI 表單工廠"""
@classmethod
def create(
cls,
db: Session,
employee: Optional[Employee] = None,
period: Optional[KPIPeriod] = None,
status: str = "draft",
with_items: bool = True,
) -> KPISheet:
if not employee:
employee = EmployeeFactory.create(db)
if not period:
period = KPIPeriodFactory.create(db)
sheet = KPISheet(
employee_id=employee.id,
period_id=period.id,
department_id=employee.department_id,
status=status,
)
db.add(sheet)
db.flush()
# 建立 KPI 項目(權重總和 100%
if with_items:
weights = [30, 25, 25, 20]
categories = ["financial", "customer", "internal", "learning"]
for i, (weight, category) in enumerate(zip(weights, categories)):
template = KPITemplateFactory.create(db, category=category)
item = KPIItem(
sheet_id=sheet.id,
template_id=template.id,
sort_order=i,
name=f"KPI 項目 {i + 1}",
category=category,
weight=weight,
level0_criteria="未達標",
level1_criteria="基本達成",
level2_criteria="達成目標",
level3_criteria="挑戰目標",
level4_criteria="超越目標",
)
db.add(item)
db.flush()
return sheet

97
tests/test_auth.py Normal file
View File

@@ -0,0 +1,97 @@
"""
認證 API 測試
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.core.security import get_password_hash
from tests.factories import EmployeeFactory
class TestAuthAPI:
"""認證 API 測試"""
def test_login_success(self, client: TestClient, db: Session):
"""測試:登入成功"""
# Arrange
employee = EmployeeFactory.create(db)
# Act
response = client.post(
"/api/auth/login",
json={
"employee_no": employee.employee_no,
"password": "password123",
},
)
# Assert
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
def test_login_wrong_password(self, client: TestClient, db: Session):
"""測試:密碼錯誤"""
# Arrange
employee = EmployeeFactory.create(db)
# Act
response = client.post(
"/api/auth/login",
json={
"employee_no": employee.employee_no,
"password": "wrong_password",
},
)
# Assert
assert response.status_code == 401
def test_login_user_not_found(self, client: TestClient):
"""測試:使用者不存在"""
response = client.post(
"/api/auth/login",
json={
"employee_no": "NOTEXIST",
"password": "password123",
},
)
assert response.status_code == 401
def test_login_inactive_user(self, client: TestClient, db: Session):
"""測試:帳號停用"""
# Arrange
employee = EmployeeFactory.create(db, status="inactive")
# Act
response = client.post(
"/api/auth/login",
json={
"employee_no": employee.employee_no,
"password": "password123",
},
)
# Assert
assert response.status_code == 401
def test_get_me_success(
self, client: TestClient, auth_headers: dict, test_employee
):
"""測試:取得當前使用者"""
response = client.get("/api/auth/me", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["id"] == test_employee.id
assert data["employee_no"] == test_employee.employee_no
def test_get_me_unauthorized(self, client: TestClient):
"""測試:未認證"""
response = client.get("/api/auth/me")
assert response.status_code == 403

140
tests/test_kpi.py Normal file
View File

@@ -0,0 +1,140 @@
"""
KPI API 測試
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.factories import (
EmployeeFactory,
KPIPeriodFactory,
KPISheetFactory,
)
class TestKPISheetAPI:
"""KPI 表單 API 測試"""
def test_create_sheet_success(
self, client: TestClient, db: Session, auth_headers: dict
):
"""測試:建立 KPI 表單"""
# Arrange
period = KPIPeriodFactory.create(db, status="setting")
db.commit()
payload = {
"period_id": period.id,
"items": [
{
"name": "營收達成率",
"category": "financial",
"weight": 50,
},
{
"name": "客戶滿意度",
"category": "customer",
"weight": 50,
},
],
}
# Act
response = client.post(
"/api/kpi/sheets",
json=payload,
headers=auth_headers,
)
# Assert
assert response.status_code == 201
data = response.json()
assert data["status"] == "draft"
assert len(data["items"]) == 2
def test_get_my_sheets(
self, client: TestClient, db: Session, auth_headers: dict, test_employee
):
"""測試:取得我的 KPI 表單"""
# Arrange
period = KPIPeriodFactory.create(db)
KPISheetFactory.create(db, employee=test_employee, period=period)
db.commit()
# Act
response = client.get("/api/kpi/sheets/my", headers=auth_headers)
# Assert
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
def test_get_sheet_not_found(self, client: TestClient, auth_headers: dict):
"""測試:查詢不存在的表單"""
response = client.get("/api/kpi/sheets/99999", headers=auth_headers)
assert response.status_code == 404
def test_submit_sheet_success(
self, client: TestClient, db: Session, auth_headers: dict, test_employee
):
"""測試:提交 KPI 審核"""
# Arrange
period = KPIPeriodFactory.create(db, status="setting")
sheet = KPISheetFactory.create(
db, employee=test_employee, period=period, status="draft", with_items=True
)
db.commit()
# Act
response = client.post(
f"/api/kpi/sheets/{sheet.id}/submit",
headers=auth_headers,
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["status"] == "pending"
def test_submit_sheet_weight_invalid(
self, client: TestClient, db: Session, auth_headers: dict, test_employee
):
"""測試:權重不正確提交應失敗"""
# Arrange
period = KPIPeriodFactory.create(db, status="setting")
sheet = KPISheetFactory.create(
db,
employee=test_employee,
period=period,
status="draft",
with_items=False, # 不建立項目
)
db.commit()
# Act
response = client.post(
f"/api/kpi/sheets/{sheet.id}/submit",
headers=auth_headers,
)
# Assert
assert response.status_code == 400
class TestKPIPeriodAPI:
"""KPI 期間 API 測試"""
def test_list_periods(self, client: TestClient, db: Session, auth_headers: dict):
"""測試:取得期間列表"""
# Arrange
KPIPeriodFactory.create(db)
db.commit()
# Act
response = client.get("/api/kpi/periods", headers=auth_headers)
# Assert
assert response.status_code == 200
data = response.json()
assert len(data) >= 1