197 lines
6.8 KiB
Python
197 lines
6.8 KiB
Python
"""
|
|
考核指标服务层
|
|
"""
|
|
import json
|
|
from typing import Optional, List, Dict, Any
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.models import Indicator, IndicatorType, BSCDimension
|
|
from app.schemas.schemas import IndicatorCreate, IndicatorUpdate
|
|
|
|
|
|
class IndicatorService:
|
|
"""考核指标服务"""
|
|
|
|
@staticmethod
|
|
async def get_list(
|
|
db: AsyncSession,
|
|
indicator_type: Optional[str] = None,
|
|
bs_dimension: Optional[str] = None,
|
|
is_active: Optional[bool] = None,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> tuple[List[Indicator], int]:
|
|
"""获取指标列表"""
|
|
query = select(Indicator)
|
|
|
|
if indicator_type:
|
|
query = query.where(Indicator.indicator_type == indicator_type)
|
|
if bs_dimension:
|
|
query = query.where(Indicator.bs_dimension == bs_dimension)
|
|
if is_active is not None:
|
|
query = query.where(Indicator.is_active == is_active)
|
|
|
|
# 统计总数
|
|
count_query = select(func.count()).select_from(query.subquery())
|
|
total = await db.scalar(count_query)
|
|
|
|
# 分页
|
|
query = query.order_by(Indicator.indicator_type, Indicator.id)
|
|
query = query.offset((page - 1) * page_size).limit(page_size)
|
|
|
|
result = await db.execute(query)
|
|
indicators = result.scalars().all()
|
|
|
|
return indicators, total or 0
|
|
|
|
@staticmethod
|
|
async def get_by_id(db: AsyncSession, indicator_id: int) -> Optional[Indicator]:
|
|
"""根据 ID 获取指标"""
|
|
result = await db.execute(
|
|
select(Indicator).where(Indicator.id == indicator_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
@staticmethod
|
|
async def get_active_indicators(db: AsyncSession) -> List[Indicator]:
|
|
"""获取所有启用的指标"""
|
|
result = await db.execute(
|
|
select(Indicator)
|
|
.where(Indicator.is_active == True)
|
|
.order_by(Indicator.indicator_type, Indicator.id)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
@staticmethod
|
|
async def create(db: AsyncSession, indicator_data: IndicatorCreate) -> Indicator:
|
|
"""创建指标"""
|
|
indicator = Indicator(**indicator_data.model_dump())
|
|
db.add(indicator)
|
|
await db.commit()
|
|
await db.refresh(indicator)
|
|
return indicator
|
|
|
|
@staticmethod
|
|
async def update(
|
|
db: AsyncSession,
|
|
indicator_id: int,
|
|
indicator_data: IndicatorUpdate
|
|
) -> Optional[Indicator]:
|
|
"""更新指标"""
|
|
indicator = await IndicatorService.get_by_id(db, indicator_id)
|
|
if not indicator:
|
|
return None
|
|
|
|
update_data = indicator_data.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(indicator, key, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(indicator)
|
|
return indicator
|
|
|
|
@staticmethod
|
|
async def delete(db: AsyncSession, indicator_id: int) -> bool:
|
|
"""删除指标"""
|
|
indicator = await IndicatorService.get_by_id(db, indicator_id)
|
|
if not indicator:
|
|
return False
|
|
|
|
await db.delete(indicator)
|
|
await db.commit()
|
|
return True
|
|
|
|
@staticmethod
|
|
async def import_template(
|
|
db: AsyncSession,
|
|
template_data: Dict[str, Any],
|
|
overwrite: bool = False
|
|
) -> int:
|
|
"""导入指标模板"""
|
|
dept_type = template_data.get('dept_type')
|
|
indicators_data = template_data.get('indicators', [])
|
|
created_count = 0
|
|
|
|
for ind_data in indicators_data:
|
|
# 检查是否已存在
|
|
existing = await db.execute(
|
|
select(Indicator).where(Indicator.code == ind_data['code'])
|
|
)
|
|
|
|
if existing.scalar_one_or_none():
|
|
if overwrite:
|
|
# 更新现有指标
|
|
indicator = existing.scalar_one_or_none()
|
|
if indicator:
|
|
for key, value in ind_data.items():
|
|
if hasattr(indicator, key):
|
|
setattr(indicator, key, value)
|
|
continue
|
|
|
|
# 创建新指标
|
|
indicator = Indicator(
|
|
name=ind_data.get('name'),
|
|
code=ind_data.get('code'),
|
|
indicator_type=ind_data.get('indicator_type'),
|
|
bs_dimension=ind_data.get('bs_dimension'),
|
|
weight=ind_data.get('weight', 1.0),
|
|
max_score=ind_data.get('max_score', 100.0),
|
|
target_value=ind_data.get('target_value'),
|
|
target_unit=ind_data.get('target_unit'),
|
|
calculation_method=ind_data.get('calculation_method'),
|
|
assessment_method=ind_data.get('assessment_method'),
|
|
deduction_standard=ind_data.get('deduction_standard'),
|
|
data_source=ind_data.get('data_source'),
|
|
applicable_dept_types=json.dumps([dept_type]) if dept_type else None,
|
|
is_veto=ind_data.get('is_veto', False),
|
|
is_active=ind_data.get('is_active', True)
|
|
)
|
|
db.add(indicator)
|
|
created_count += 1
|
|
|
|
await db.commit()
|
|
return created_count
|
|
|
|
@staticmethod
|
|
async def get_templates() -> List[Dict[str, Any]]:
|
|
"""获取指标模板列表"""
|
|
return [
|
|
{
|
|
"name": "手术临床科室考核指标",
|
|
"dept_type": "clinical_surgical",
|
|
"description": "适用于外科系统各手术科室",
|
|
"indicator_count": 12
|
|
},
|
|
{
|
|
"name": "非手术有病房科室考核指标",
|
|
"dept_type": "clinical_nonsurgical_ward",
|
|
"description": "适用于内科系统等有病房科室",
|
|
"indicator_count": 10
|
|
},
|
|
{
|
|
"name": "非手术无病房科室考核指标",
|
|
"dept_type": "clinical_nonsurgical_noward",
|
|
"description": "适用于门诊科室",
|
|
"indicator_count": 8
|
|
},
|
|
{
|
|
"name": "医技科室考核指标",
|
|
"dept_type": "medical_tech",
|
|
"description": "适用于检验科、放射科等医技科室",
|
|
"indicator_count": 8
|
|
},
|
|
{
|
|
"name": "行政科室考核指标",
|
|
"dept_type": "admin",
|
|
"description": "适用于党办、财务科、医保办等行政科室",
|
|
"indicator_count": 6
|
|
},
|
|
{
|
|
"name": "后勤保障科室考核指标",
|
|
"dept_type": "logistics",
|
|
"description": "适用于总务科、采购科、基建科",
|
|
"indicator_count": 6
|
|
}
|
|
]
|