565 lines
22 KiB
Python
565 lines
22 KiB
Python
"""
|
||
测试数据初始化脚本
|
||
|
||
包含：
|
||
1. 科室类型BSC维度权重配置初始化
|
||
2. 满意度调查问卷示例数据
|
||
3. 评分方法示例数据
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from sqlalchemy import select
|
||
from app.core.database import async_session_maker
|
||
from app.models.models import (
|
||
DeptType, DeptTypeDimensionWeight, BSCDimension,
|
||
Survey, SurveyQuestion, SurveyStatus, SurveyType, QuestionType,
|
||
Indicator, IndicatorType, IndicatorTemplate, TemplateIndicator, TemplateType,
|
||
Department, Staff
|
||
)
|
||
from app.services.dimension_weight_service import DEFAULT_WEIGHTS
|
||
import json
|
||
|
||
|
||
async def init_dimension_weights():
|
||
"""初始化科室类型BSC维度权重配置"""
|
||
print("=" * 50)
|
||
print("初始化科室类型BSC维度权重配置...")
|
||
print("=" * 50)
|
||
|
||
async with async_session_maker() as db:
|
||
for dept_type, weights in DEFAULT_WEIGHTS.items():
|
||
# 检查是否已存在
|
||
result = await db.execute(
|
||
select(DeptTypeDimensionWeight)
|
||
.where(DeptTypeDimensionWeight.dept_type == dept_type)
|
||
)
|
||
existing = result.scalar_one_or_none()
|
||
|
||
if existing:
|
||
print(f" {dept_type.value}: 已存在，跳过")
|
||
continue
|
||
|
||
config = DeptTypeDimensionWeight(
|
||
dept_type=dept_type,
|
||
financial_weight=weights["financial"],
|
||
customer_weight=weights["customer"],
|
||
internal_process_weight=weights["internal_process"],
|
||
learning_growth_weight=weights["learning_growth"],
|
||
description=weights.get("description", "")
|
||
)
|
||
db.add(config)
|
||
print(f" {dept_type.value}: 财务{weights['financial']*100}%, 客户{weights['customer']*100}%, 流程{weights['internal_process']*100}%, 学习{weights['learning_growth']*100}%")
|
||
|
||
await db.commit()
|
||
print("科室类型BSC维度权重配置初始化完成！\n")
|
||
|
||
|
||
async def init_sample_surveys():
|
||
"""初始化满意度调查问卷示例数据"""
|
||
print("=" * 50)
|
||
print("初始化满意度调查问卷示例数据...")
|
||
print("=" * 50)
|
||
|
||
async with async_session_maker() as db:
|
||
# 检查是否已存在问卷
|
||
result = await db.execute(select(Survey))
|
||
if result.scalars().first():
|
||
print(" 已存在问卷数据，跳过初始化\n")
|
||
return
|
||
|
||
# 创建住院患者满意度问卷
|
||
survey1 = Survey(
|
||
survey_name="住院患者满意度调查问卷",
|
||
survey_code="INPATIENT_001",
|
||
survey_type=SurveyType.INPATIENT,
|
||
description="用于评估住院患者对医院服务的满意度",
|
||
status=SurveyStatus.PUBLISHED,
|
||
is_anonymous=True,
|
||
total_questions=10
|
||
)
|
||
db.add(survey1)
|
||
await db.flush()
|
||
|
||
# 添加问卷题目
|
||
questions1 = [
|
||
{
|
||
"question_text": "您对入院手续办理的便捷程度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 1
|
||
},
|
||
{
|
||
"question_text": "您对病房环境的整洁舒适程度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 2
|
||
},
|
||
{
|
||
"question_text": "您对医生的服务态度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 3
|
||
},
|
||
{
|
||
"question_text": "您对护士的护理服务是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 4
|
||
},
|
||
{
|
||
"question_text": "您对医生的技术水平是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 5
|
||
},
|
||
{
|
||
"question_text": "您对医院餐饮服务质量是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 6
|
||
},
|
||
{
|
||
"question_text": "您对检查检验等候时间是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 7
|
||
},
|
||
{
|
||
"question_text": "您对出院结算流程是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 8
|
||
},
|
||
{
|
||
"question_text": "您对医院整体服务是否满意？",
|
||
"question_type": QuestionType.SINGLE_CHOICE,
|
||
"options": json.dumps([
|
||
{"label": "非常满意", "value": "5", "score": 5},
|
||
{"label": "满意", "value": "4", "score": 4},
|
||
{"label": "一般", "value": "3", "score": 3},
|
||
{"label": "不满意", "value": "2", "score": 2},
|
||
{"label": "非常不满意", "value": "1", "score": 1}
|
||
], ensure_ascii=False),
|
||
"score_max": 5,
|
||
"sort_order": 9
|
||
},
|
||
{
|
||
"question_text": "您对医院有什么意见或建议？",
|
||
"question_type": QuestionType.TEXT,
|
||
"is_required": False,
|
||
"sort_order": 10
|
||
}
|
||
]
|
||
|
||
for q in questions1:
|
||
question = SurveyQuestion(
|
||
survey_id=survey1.id,
|
||
question_text=q["question_text"],
|
||
question_type=q["question_type"],
|
||
options=q.get("options"),
|
||
score_max=q.get("score_max", 5),
|
||
is_required=q.get("is_required", True),
|
||
sort_order=q["sort_order"]
|
||
)
|
||
db.add(question)
|
||
|
||
print(f" 创建问卷: {survey1.survey_name}")
|
||
|
||
# 创建门诊患者满意度问卷
|
||
survey2 = Survey(
|
||
survey_name="门诊患者满意度调查问卷",
|
||
survey_code="OUTPATIENT_001",
|
||
survey_type=SurveyType.OUTPATIENT,
|
||
description="用于评估门诊患者对医院服务的满意度",
|
||
status=SurveyStatus.PUBLISHED,
|
||
is_anonymous=True,
|
||
total_questions=8
|
||
)
|
||
db.add(survey2)
|
||
await db.flush()
|
||
|
||
questions2 = [
|
||
{
|
||
"question_text": "您对挂号流程的便捷程度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 1
|
||
},
|
||
{
|
||
"question_text": "您对候诊时间是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 2
|
||
},
|
||
{
|
||
"question_text": "您对医生的诊疗服务是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 3
|
||
},
|
||
{
|
||
"question_text": "您对药房取药流程是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 4
|
||
},
|
||
{
|
||
"question_text": "您对收费处服务是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 5
|
||
},
|
||
{
|
||
"question_text": "您对门诊环境是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 6
|
||
},
|
||
{
|
||
"question_text": "您对门诊整体服务是否满意？",
|
||
"question_type": QuestionType.SINGLE_CHOICE,
|
||
"options": json.dumps([
|
||
{"label": "非常满意", "value": "5", "score": 5},
|
||
{"label": "满意", "value": "4", "score": 4},
|
||
{"label": "一般", "value": "3", "score": 3},
|
||
{"label": "不满意", "value": "2", "score": 2},
|
||
{"label": "非常不满意", "value": "1", "score": 1}
|
||
], ensure_ascii=False),
|
||
"score_max": 5,
|
||
"sort_order": 7
|
||
},
|
||
{
|
||
"question_text": "您对医院有什么意见或建议？",
|
||
"question_type": QuestionType.TEXT,
|
||
"is_required": False,
|
||
"sort_order": 8
|
||
}
|
||
]
|
||
|
||
for q in questions2:
|
||
question = SurveyQuestion(
|
||
survey_id=survey2.id,
|
||
question_text=q["question_text"],
|
||
question_type=q["question_type"],
|
||
options=q.get("options"),
|
||
score_max=q.get("score_max", 5),
|
||
is_required=q.get("is_required", True),
|
||
sort_order=q["sort_order"]
|
||
)
|
||
db.add(question)
|
||
|
||
print(f" 创建问卷: {survey2.survey_name}")
|
||
|
||
# 创建科室间满意度问卷
|
||
survey3 = Survey(
|
||
survey_name="科室间协作满意度调查",
|
||
survey_code="INTERNAL_001",
|
||
survey_type=SurveyType.DEPARTMENT,
|
||
description="用于评估科室之间的协作配合满意度",
|
||
status=SurveyStatus.DRAFT,
|
||
is_anonymous=False,
|
||
total_questions=6
|
||
)
|
||
db.add(survey3)
|
||
await db.flush()
|
||
|
||
questions3 = [
|
||
{
|
||
"question_text": "您对该科室的工作响应速度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 1
|
||
},
|
||
{
|
||
"question_text": "您对该科室的工作配合度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 2
|
||
},
|
||
{
|
||
"question_text": "您对该科室的服务态度是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 3
|
||
},
|
||
{
|
||
"question_text": "您对该科室的工作质量是否满意？",
|
||
"question_type": QuestionType.SCORE,
|
||
"score_max": 5,
|
||
"sort_order": 4
|
||
},
|
||
{
|
||
"question_text": "您对该科室整体协作是否满意？",
|
||
"question_type": QuestionType.SINGLE_CHOICE,
|
||
"options": json.dumps([
|
||
{"label": "非常满意", "value": "5", "score": 5},
|
||
{"label": "满意", "value": "4", "score": 4},
|
||
{"label": "一般", "value": "3", "score": 3},
|
||
{"label": "不满意", "value": "2", "score": 2},
|
||
{"label": "非常不满意", "value": "1", "score": 1}
|
||
], ensure_ascii=False),
|
||
"score_max": 5,
|
||
"sort_order": 5
|
||
},
|
||
{
|
||
"question_text": "您对该科室有什么建议？",
|
||
"question_type": QuestionType.TEXT,
|
||
"is_required": False,
|
||
"sort_order": 6
|
||
}
|
||
]
|
||
|
||
for q in questions3:
|
||
question = SurveyQuestion(
|
||
survey_id=survey3.id,
|
||
question_text=q["question_text"],
|
||
question_type=q["question_type"],
|
||
options=q.get("options"),
|
||
score_max=q.get("score_max", 5),
|
||
is_required=q.get("is_required", True),
|
||
sort_order=q["sort_order"]
|
||
)
|
||
db.add(question)
|
||
|
||
print(f" 创建问卷: {survey3.survey_name}")
|
||
|
||
await db.commit()
|
||
print("满意度调查问卷示例数据初始化完成！\n")
|
||
|
||
|
||
async def init_sample_survey_responses():
|
||
"""初始化满意度调查回答示例数据"""
|
||
print("=" * 50)
|
||
print("初始化满意度调查回答示例数据...")
|
||
print("=" * 50)
|
||
|
||
async with async_session_maker() as db:
|
||
# 获取问卷
|
||
result = await db.execute(
|
||
select(Survey).where(Survey.survey_code == "INPATIENT_001")
|
||
)
|
||
survey = result.scalar_one_or_none()
|
||
if not survey:
|
||
print(" 未找到住院患者满意度问卷，跳过\n")
|
||
return
|
||
|
||
# 获取科室列表
|
||
dept_result = await db.execute(select(Department))
|
||
departments = dept_result.scalars().all()
|
||
|
||
if not departments:
|
||
print(" 未找到科室数据，跳过\n")
|
||
return
|
||
|
||
# 获取问卷题目
|
||
q_result = await db.execute(
|
||
select(SurveyQuestion)
|
||
.where(SurveyQuestion.survey_id == survey.id)
|
||
.order_by(SurveyQuestion.sort_order)
|
||
)
|
||
questions = q_result.scalars().all()
|
||
|
||
# 为每个科室生成示例回答
|
||
from app.models.models import SurveyResponse, SurveyAnswer
|
||
import random
|
||
|
||
for dept in departments[:5]: # 只为前5个科室生成
|
||
# 生成5-10条回答
|
||
num_responses = random.randint(5, 10)
|
||
for _ in range(num_responses):
|
||
total_score = 0
|
||
max_score = 0
|
||
|
||
response = SurveyResponse(
|
||
survey_id=survey.id,
|
||
department_id=dept.id,
|
||
respondent_type="patient"
|
||
)
|
||
db.add(response)
|
||
await db.flush()
|
||
|
||
# 为每个题目生成回答
|
||
for q in questions:
|
||
if q.question_type == QuestionType.TEXT:
|
||
continue
|
||
|
||
# 随机生成评分（倾向高分）
|
||
score = random.choices(
|
||
[5, 4, 3, 2, 1],
|
||
weights=[50, 30, 15, 3, 2]
|
||
)[0]
|
||
|
||
answer = SurveyAnswer(
|
||
response_id=response.id,
|
||
question_id=q.id,
|
||
answer_value=str(score),
|
||
score=score
|
||
)
|
||
db.add(answer)
|
||
total_score += score
|
||
max_score += q.score_max
|
||
|
||
# 更新回答记录得分
|
||
response.total_score = total_score
|
||
response.max_score = max_score
|
||
response.satisfaction_rate = (total_score / max_score * 100) if max_score > 0 else 0
|
||
|
||
await db.commit()
|
||
print(f" 为{min(len(departments), 5)}个科室生成了满意度调查回答数据")
|
||
print("满意度调查回答示例数据初始化完成！\n")
|
||
|
||
|
||
async def init_scoring_method_indicators():
|
||
"""初始化评分方法示例指标"""
|
||
print("=" * 50)
|
||
print("初始化评分方法示例指标...")
|
||
print("=" * 50)
|
||
|
||
async with async_session_maker() as db:
|
||
# 检查是否已存在评分方法相关指标
|
||
result = await db.execute(
|
||
select(Indicator).where(Indicator.code.in_(["SCORE001", "SCORE002", "SCORE003", "SCORE004"]))
|
||
)
|
||
if result.scalars().first():
|
||
print(" 已存在评分方法示例指标，跳过初始化\n")
|
||
return
|
||
|
||
# 创建示例指标，展示各种评分方法
|
||
scoring_indicators = [
|
||
# 目标参照法示例
|
||
Indicator(
|
||
name="业务收支结余率",
|
||
code="SCORE001",
|
||
indicator_type=IndicatorType.QUANTITY,
|
||
bs_dimension=BSCDimension.FINANCIAL,
|
||
weight=12.6,
|
||
max_score=100,
|
||
target_value=15.0, # 目标值15%
|
||
target_unit="%",
|
||
calculation_method="(业务收入 - 业务支出)/业务收入 × 100%",
|
||
assessment_method="target_reference", # 目标参照法
|
||
data_source="财务系统",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward"]),
|
||
is_active=True
|
||
),
|
||
# 区间法-趋高指标示例
|
||
Indicator(
|
||
name="人均收支结余",
|
||
code="SCORE002",
|
||
indicator_type=IndicatorType.QUANTITY,
|
||
bs_dimension=BSCDimension.FINANCIAL,
|
||
weight=16.8,
|
||
max_score=100,
|
||
target_value=5000, # 基准值
|
||
target_unit="元",
|
||
calculation_method="业务收支结余/平均职工人数",
|
||
assessment_method="interval_high", # 区间法-趋高
|
||
data_source="财务系统",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward"]),
|
||
is_active=True
|
||
),
|
||
# 区间法-趋低指标示例
|
||
Indicator(
|
||
name="百元收入耗材率",
|
||
code="SCORE003",
|
||
indicator_type=IndicatorType.COST,
|
||
bs_dimension=BSCDimension.FINANCIAL,
|
||
weight=12.6,
|
||
max_score=100,
|
||
target_value=25.0, # 目标值≤25元
|
||
target_unit="元",
|
||
calculation_method="耗材支出/业务收入 × 100",
|
||
assessment_method="interval_low", # 区间法-趋低
|
||
data_source="物资系统",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward"]),
|
||
is_active=True
|
||
),
|
||
# 扣分法示例
|
||
Indicator(
|
||
name="病历质量考核",
|
||
code="SCORE004",
|
||
indicator_type=IndicatorType.QUALITY,
|
||
bs_dimension=BSCDimension.INTERNAL_PROCESS,
|
||
weight=3.2,
|
||
max_score=100,
|
||
target_value=0, # 目标值0（无乙级病历）
|
||
calculation_method="乙级病历每份扣5分，丙级病历不得分",
|
||
assessment_method="deduction", # 扣分法
|
||
deduction_standard="乙级病历每份扣5分",
|
||
data_source="病案质控",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward"]),
|
||
is_active=True
|
||
),
|
||
# 加分法示例
|
||
Indicator(
|
||
name="科研教学工作",
|
||
code="SCORE005",
|
||
indicator_type=IndicatorType.QUANTITY, # 使用QUANTITY代替
|
||
bs_dimension=BSCDimension.LEARNING_GROWTH,
|
||
weight=4.5,
|
||
max_score=100,
|
||
target_value=0,
|
||
calculation_method="科研教学成果加分",
|
||
assessment_method="bonus", # 加分法
|
||
deduction_standard="科研项目立项：市级加5分，省级加10分，国家级加20分\n论文发表：核心期刊加5分，SCI加10分",
|
||
data_source="科教科",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward", "medical_tech"]),
|
||
is_active=True
|
||
),
|
||
# 区间法-趋中指标示例
|
||
Indicator(
|
||
name="平均住院日",
|
||
code="SCORE006",
|
||
indicator_type=IndicatorType.EFFICIENCY,
|
||
bs_dimension=BSCDimension.INTERNAL_PROCESS,
|
||
weight=2.0,
|
||
max_score=100,
|
||
target_value=10.0, # 目标值10天
|
||
target_unit="天",
|
||
calculation_method="出院患者平均住院天数",
|
||
assessment_method="interval_center", # 区间法-趋中
|
||
data_source="病案统计",
|
||
applicable_dept_types=json.dumps(["clinical_surgical", "clinical_nonsurgical_ward"]),
|
||
is_active=True
|
||
),
|
||
]
|
||
|
||
for indicator in scoring_indicators:
|
||
db.add(indicator)
|
||
print(f" 创建指标: {indicator.name} ({indicator.assessment_method})")
|
||
|
||
await db.commit()
|
||
print("评分方法示例指标初始化完成！\n")
|
||
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
print("\n" + "=" * 60)
|
||
print("开始初始化测试数据...")
|
||
print("=" * 60 + "\n")
|
||
|
||
# 初始化科室类型BSC维度权重配置
|
||
await init_dimension_weights()
|
||
|
||
# 初始化满意度调查问卷示例数据
|
||
await init_sample_surveys()
|
||
|
||
# 初始化满意度调查回答示例数据
|
||
await init_sample_survey_responses()
|
||
|
||
# 初始化评分方法示例指标
|
||
await init_scoring_method_indicators()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试数据初始化完成！")
|
||
print("=" * 60 + "\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|