142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
"""
|
|
API路由 - 考核指标管理
|
|
"""
|
|
from typing import Annotated, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_active_user, get_current_manager_user
|
|
from app.schemas.schemas import (
|
|
IndicatorCreate, IndicatorUpdate, IndicatorResponse,
|
|
ResponseBase
|
|
)
|
|
from app.services.indicator_service import IndicatorService
|
|
from app.models.models import User
|
|
|
|
router = APIRouter(prefix="/indicators", tags=["考核指标"])
|
|
|
|
|
|
@router.get("", summary="获取指标列表")
|
|
async def get_indicators(
|
|
indicator_type: Optional[str] = Query(None, description="指标类型"),
|
|
bs_dimension: Optional[str] = Query(None, description="BSC 维度"),
|
|
is_active: Optional[bool] = Query(None, description="是否启用"),
|
|
page: int = Query(1, ge=1, description="页码"),
|
|
page_size: int = Query(20, ge=1, le=100, description="每页数量"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""获取指标列表"""
|
|
indicators, total = await IndicatorService.get_list(
|
|
db, indicator_type, bs_dimension, is_active, page, page_size
|
|
)
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": [IndicatorResponse.model_validate(i, from_attributes=True) for i in indicators],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size
|
|
}
|
|
|
|
|
|
@router.get("/active", summary="获取所有启用的指标")
|
|
async def get_active_indicators(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""获取所有启用的指标"""
|
|
indicators = await IndicatorService.get_active_indicators(db)
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": [IndicatorResponse.model_validate(i, from_attributes=True) for i in indicators]
|
|
}
|
|
|
|
|
|
@router.get("/{indicator_id}", summary="获取指标详情")
|
|
async def get_indicator(
|
|
indicator_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""获取指标详情"""
|
|
indicator = await IndicatorService.get_by_id(db, indicator_id)
|
|
if not indicator:
|
|
raise HTTPException(status_code=404, detail="指标不存在")
|
|
return {"code": 200, "message": "success", "data": IndicatorResponse.model_validate(indicator, from_attributes=True)}
|
|
|
|
|
|
@router.post("", summary="创建指标")
|
|
async def create_indicator(
|
|
indicator_data: IndicatorCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Annotated[User, Depends(get_current_manager_user)] = None
|
|
):
|
|
"""创建指标（需要管理员或经理权限）"""
|
|
# 检查编码是否已存在
|
|
existing = await IndicatorService.get_by_code(db, indicator_data.code)
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="指标编码已存在")
|
|
|
|
indicator = await IndicatorService.create(db, indicator_data)
|
|
return {"code": 200, "message": "创建成功", "data": IndicatorResponse.model_validate(indicator, from_attributes=True)}
|
|
|
|
|
|
@router.put("/{indicator_id}", summary="更新指标")
|
|
async def update_indicator(
|
|
indicator_id: int,
|
|
indicator_data: IndicatorUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Annotated[User, Depends(get_current_manager_user)] = None
|
|
):
|
|
"""更新指标（需要管理员或经理权限）"""
|
|
indicator = await IndicatorService.update(db, indicator_id, indicator_data)
|
|
if not indicator:
|
|
raise HTTPException(status_code=404, detail="指标不存在")
|
|
return {"code": 200, "message": "更新成功", "data": IndicatorResponse.model_validate(indicator, from_attributes=True)}
|
|
|
|
|
|
@router.delete("/{indicator_id}", summary="删除指标")
|
|
async def delete_indicator(
|
|
indicator_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Annotated[User, Depends(get_current_manager_user)] = None
|
|
):
|
|
"""删除指标（需要管理员或经理权限）"""
|
|
success = await IndicatorService.delete(db, indicator_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="指标不存在")
|
|
return {"code": 200, "message": "删除成功"}
|
|
|
|
|
|
@router.get("/templates/list", summary="获取指标模板列表")
|
|
async def get_indicator_templates(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""获取指标模板列表"""
|
|
templates = await IndicatorService.get_templates()
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": templates
|
|
}
|
|
|
|
|
|
@router.post("/templates/import", summary="导入指标模板")
|
|
async def import_indicator_template(
|
|
template_data: dict,
|
|
overwrite: bool = Query(False, description="是否覆盖已存在的指标"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Annotated[User, Depends(get_current_manager_user)] = None
|
|
):
|
|
"""导入指标模板"""
|
|
count = await IndicatorService.import_template(db, template_data, overwrite)
|
|
return {
|
|
"code": 200,
|
|
"message": f"成功导入 {count} 个指标",
|
|
"data": {"created_count": count}
|
|
}
|