from typing import Any from sqlalchemy.orm import Session from app.core.exceptions import AppError from app.models.source_case import CaseBase from app.repositories.case_repository import CaseRepository from app.schemas.training_config import ( ConfigOption, PatientConfig, TrainingConfigOptionsResponse, TrainingConfigRecommendedResponse, ) class TrainingConfigService: """训练配置服务：提供训练页病人初始化配置，不写数据库。""" def __init__(self, db: Session) -> None: self.db = db self.case_repo = CaseRepository(db) def get_recommended(self, case_id: int) -> TrainingConfigRecommendedResponse: """推荐配置：根据病例内容返回训练页默认病人初始化配置。""" case = self._get_case(case_id) recommended = self.default_patient_config(case) return TrainingConfigRecommendedResponse( case_id=case_id, recommended=recommended, recommended_labels=self.config_labels(recommended), options=self.config_options(), ) def get_options(self, case_id: int) -> TrainingConfigOptionsResponse: """配置选项：返回训练页自定义病人初始化配置的全部可选项和病例推荐值。""" case = self._get_case(case_id) recommended = self.default_patient_config(case) return TrainingConfigOptionsResponse( case_id=case_id, recommended=recommended, recommended_labels=self.config_labels(recommended), options=self.config_options(), ) def default_patient_config(self, case: CaseBase | None = None) -> PatientConfig: """默认配置：按病例年龄、科室语义和病情关键词初始化 AI 病人沟通风格。""" if case is None: return PatientConfig( visit_environment="outpatient", age_group="youth", education_level="higher", personality="calm", ) text = self._case_text(case) patient_age = case.patient_age visit_environment = "outpatient" personality = "calm" if self._contains_any( text, ["急诊", "危重", "休克", "昏迷", "意识障碍", "呼吸困难", "低氧", "抢救", "SpO2 90", "血氧明显下降"], ): visit_environment = "emergency" personality = "anxious" elif self._contains_any(text, ["住院", "病房", "入院", "住院治疗"]): visit_environment = "ward" age_group = "youth" if (patient_age is not None and patient_age < 14) or self._contains_any( text, ["儿童", "患儿", "儿科", "小儿", "幼儿", "婴儿"], ): age_group = "child" elif patient_age is not None and patient_age >= 65: age_group = "elderly" elif patient_age is not None and patient_age >= 45: age_group = "middle_aged" return PatientConfig( visit_environment=visit_environment, age_group=age_group, education_level="higher", personality=personality, ) def normalize_patient_config(self, config: PatientConfig | None, case: CaseBase | None = None) -> dict: """配置归一：校验并补齐前端传入的病人初始化配置。""" selected = config or self.default_patient_config(case) values = selected.model_dump() allowed = {key: {item.value for item in items} for key, items in self.config_options().items()} for key, value in values.items(): if value not in allowed.get(key, set()): raise AppError("TRAINING_CONFIG_INVALID", f"invalid patient config field: {key}", 400) return { "values": values, "labels": self.config_labels(selected), } def config_labels(self, config: PatientConfig) -> dict[str, str]: """配置标签：把配置值转换为前端和提示词可读的中文标签。""" option_map = { key: {item.value: item.label for item in items} for key, items in self.config_options().items() } values = config.model_dump() return {key: option_map.get(key, {}).get(value, value) for key, value in values.items()} def config_options(self) -> dict[str, list[ConfigOption]]: """配置选项：训练页可选病人初始化配置。""" return { "visit_environment": [ ConfigOption(value="outpatient", label="门诊", description="适合常规问诊训练"), ConfigOption(value="emergency", label="急诊", description="病情紧急，沟通节奏更快"), ConfigOption(value="ward", label="病房", description="适合住院病情追踪和处置沟通"), ], "age_group": [ ConfigOption(value="child", label="儿童", description="由家属代述为主"), ConfigOption(value="youth", label="青年", description="表达清楚，能配合问诊"), ConfigOption(value="middle_aged", label="中年", description="关注工作、家庭和慢病背景"), ConfigOption(value="elderly", label="老年", description="表达较慢，需关注基础病和用药史"), ], "education_level": [ ConfigOption(value="primary_or_below", label="小学及以下", description="医学术语理解弱"), ConfigOption(value="secondary", label="中等教育", description="能理解常见健康解释"), ConfigOption(value="higher", label="高等教育", description="理解能力强，能描述细节"), ], "personality": [ ConfigOption(value="calm", label="平和", description="情绪稳定，按问题回答"), ConfigOption(value="anxious", label="焦虑", description="更担心病情和治疗风险"), ConfigOption(value="impatient", label="急躁", description="希望快速获得结论"), ConfigOption(value="cooperative", label="配合", description="愿意补充细节"), ConfigOption(value="suspicious", label="多疑", description="会追问检查和用药依据"), ], } def _get_case(self, case_id: int) -> CaseBase: """病例校验：读取配置请求对应的已发布病例。""" case = self.case_repo.get_active_case(case_id) if not case: raise AppError("CASE_NOT_FOUND", "case not found or inactive", 404) return case def _case_text(self, case: CaseBase) -> str: """病例文本：拼接用于推荐配置推断的病例字段。""" parts: list[str] = [ case.title or "", case.chief_complaint or "", case.description or "", case.tags or "", case.patient_gender or "", ] for field_name in ( "symptom_tags", "disease_tags", "competency_tags", "guideline_tags", "knowledge_points", "icd_codes", ): parts.extend(self._stringify(getattr(case, field_name, None))) if case.traditional_case: parts.extend(self._stringify(case.traditional_case.guideline_reference)) if case.teaching_case: parts.extend(self._stringify(case.teaching_case.teaching_goal)) return " ".join(item for item in parts if item) def _stringify(self, value: Any) -> list[str]: """字段展开：把 JSON、列表和普通文本统一转换为可检索字符串。""" if value is None: return [] if isinstance(value, str): return [value] if isinstance(value, dict): return [str(item) for pair in value.items() for item in pair if item is not None] if isinstance(value, list | tuple | set): return [str(item) for item in value if item is not None] return [str(value)] def _contains_any(self, text: str, keywords: list[str]) -> bool: """关键词判断：判断病例文本是否命中任一推荐配置关键词。""" return any(keyword in text for keyword in keywords)