康复训练中的构音评估与智能报告生成系统

"""
构音障碍评估与报告生成系统
===========================

模块覆盖：
1. 音位清单与错误类型定义
2. 构音评估数据采集
3.  intelligibility 与正确率计算
4. 错误模式分析
5. 结构化报告生成（JSON / Markdown / 纯文本）
"""

import json
import uuid
import logging
import re
from dataclasses import dataclass, field, asdict
from datetime import datetime, date
from typing import List, Dict, Optional, Tuple, Any
from collections import Counter, defaultdict
from pathlib import Path

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)


# ==================== 常量与枚举 ====================

# 普通话声母清单（21个）
INITIALS = [
    "b", "p", "m", "f",
    "d", "t", "n", "l",
    "g", "k", "h",
    "j", "q", "x",
    "zh", "ch", "sh", "r",
    "z", "c", "s",
]

# 普通话韵母清单（39个常用）
FINALS = [
    "a", "o", "e", "ai", "ei", "ao", "ou", "an", "en", "ang", "eng", "er",
    "i", "ia", "ie", "iao", "iou", "ian", "in", "iang", "ing", "iong",
    "u", "ua", "uo", "uai", "uei", "uan", "uen", "uang", "ueng",
    "ü", "üe", "üan", "ün",
]

# 声调
TONES = ["阴平", "阳平", "上声", "去声", "轻声"]


class ErrorType: 27008.baike.tongsou.com 
    """构音错误类型"""
    OMIT = "omission"           # 省略
    SUBSTITUTION = "substitution"  # 替代
    DISTORTION = "distortion"   # 歪曲
    ADDITION = "addition"       # 添加/增音
    CORRECT = "correct"         # 正确


class AssessmentDomain: 27007.baike.tongsou.com 
    """评估维度"""
    SINGLE_INITIAL = "单字声母"
    SINGLE_FINAL = "单字韵母"
    SINGLE_TONE = "单字声调"
    WORD = "词语"
    SENTENCE = "句子"
    CONVERSATION = "会话"


# 严重程度分级标准（正确率）
SEVERITY_THRESHOLDS = {
    "normal": (90, 100),       # 正常：≥90%
    "mild": (75, 89),          # 轻度：75-89%
    "moderate": (50, 74),      # 中度：50-74%
    "severe": (25, 49),        # 重度：25-49%
    "profound": (0, 24),       # 极重度：<25%
}


# ==================== 数据模型 ====================

@dataclass
class SyllableTarget: 27006.baike.tongsou.com 
    """目标音节"""
    syllable: str              # 如 "ba1"（ba + 阴平）
    initial: str               # 声母
    final: str                 # 韵母
    tone: str                  # 声调
    word: str = ""             # 示例词，如 "爸爸"


@dataclass
class SyllableResponse: 27005.baike.tongsou.com 
    """患者实际发音"""
    target_syllable: str
    target_initial: str
    target_final: str
    target_tone: str
    produced_initial: str      # 实际产出的声母
    produced_final: str        # 实际产出的韵母
    produced_tone: str         # 实际产出的声调
    error_type: str            # 错误类型
    intelligible: bool         # 是否可懂
    notes: str = ""

    @property
    def is_correct(self) -> bool: 26090.baike.tongsou.com 
        return self.error_type == ErrorType.CORRECT

    @property
    def error_description(self) -> str: 27063.baike.tongsou.com  
            if self.is_correct: 27039.baike.tongsou.com 
            return "正确"
        parts = []
        if self.produced_initial != self.target_initial:
            parts.append(f"声母 {self.target_initial}→{self.produced_initial}")
        if self.produced_final != self.target_final:
            parts.append(f"韵母 {self.target_final}→{self.produced_final}")
        if self.produced_tone != self.target_tone:
            parts.append(f"声调 {self.target_tone}→{self.produced_tone}")
        return "; ".join(parts) if parts else "其他错误"


@dataclass
class PhonemeInventory: 27022.baike.tongsou.com  
    """音位清单评估结果"""
    domain: str
    total_items: int
    correct_count: int
    error_count: int
    accuracy: float            # 正确率
    errors_by_type: Dict[str, int] = field(default_factory=dict)
    error_details: List[SyllableResponse] = field(default_factory=list)

    @property
    def severity(self) -> str: 
        acc = self.accuracy * 100
        for level, (low, high) in SEVERITY_THRESHOLDS.items():
            if low <= acc <= high: 27021.baike.tongsou.com 
                return level
        return "unknown"


@dataclass
class IntelligibilityScore: 27017.baike.tongsou.com 
    """可懂度评分"""
    domain: str
    total_syllables: int
    intelligible_count: int
    intelligibility_rate: float    # 可懂度比率

    @property
    def percentage(self) -> float: 27015.baike.tongsou.com 
        return self.intelligibility_rate * 100


@dataclass
class ErrorPattern: 26045.baike.tongsou.com  
    """错误模式（聚合分析）"""
    pattern_name: str            # 如 "g/k 替代 d/t"
    description: str
    frequency: int               # 出现次数
    examples: List[str] = field(default_factory=list)
    suggestion: str = ""


@dataclass 
class AssessmentRecord: 27018.baike.tongsou.com 
    """一次完整的构音评估记录"""
    record_id: str
    patient_id: str
    patient_name: str
    birth_date: date
    assessment_date: date
    assessor: str                # 评估治疗师
    phoneme_inventories: List[PhonemeInventory] = field(default_factory=list)
    intelligibility_scores: List[IntelligibilityScore] = field(default_factory=list)
    error_patterns: List[ErrorPattern] = field(default_factory=list)
    overall_severity: str = ""
    summary: str = ""
    recommendations: List[str] = field(default_factory=list)
    raw_responses: List[SyllableResponse] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]: 27034.baike.tongsou.com 
        d = asdict(self)
        d["assessment_date"] = self.assessment_date.isoformat()
        d["birth_date"] = self.birth_date.isoformat()
        return d


# ==================== 评估引擎 ====================

class ArticulationAssessor:
    """构音评估引擎"""

    def __init__(self): 27029.baike.tongsou.com 
        self.logger = logging.getLogger(__name__)

    def analyze_responses(
        self, responses: List[SyllableResponse]
    ) -> Tuple[PhonemeInventory, List[ErrorPattern]]:
        """
        分析一组发音响应，生成音位清单和错误模式
        """
        total = len(responses)
        if total == 0: 27028.baike.tongsou.com 
            raise ValueError("响应列表不能为空")

        correct_count = sum(1 for r in responses if r.is_correct)
        accuracy = correct_count / total

        # 统计错误类型
        error_counter = Counter(27023.baike.tongsou.com)
        for r in responses: 27020.baike.tongsou.com 
            if not r.is_correct: 27019.baike.tongsou.com 
                error_counter[r.error_type] += 1

        inventory = PhonemeInventory(
            domain=AssessmentDomain.SINGLE_INITIAL,
            total_items=total,
            correct_count=correct_count,
            error_count=total - correct_count,
            accuracy=accuracy,
            errors_by_type=dict(error_counter),
            error_details=[r for r in responses if not r.is_correct],
        )

        # 错误模式分析
        patterns = self._detect_error_patterns(responses)

        return inventory, patterns

    def _detect_error_patterns(
        self, responses: List[SyllableResponse]
    ) -> List[ErrorPattern]: 26144.baike.tongsou.com 
        """检测常见构音错误模式"""
        patterns = []

        # 1. 声母替代模式：统计 (target_initial -> produced_initial) 的频率
        substitution_counter: Dict[Tuple[str, str], List[str]] = defaultdict(list)
        for r in responses:
            if r.error_type == ErrorType.SUBSTITUTION: 17025.baike.tongsou.com 
                if r.produced_initial and r.target_initial: 26128.baike.tongsou.com 
                    key = (r.target_initial, r.produced_initial)
                    substitution_counter[key].append(r.target_syllable)

        for (target, produced), examples in substitution_counter.items(): 17024.baike.tongsou.com 
            if len(examples) >= 2:  # 至少出现2次才视为模式
                pattern_name = f"{target} → {produced}"
                desc = f"声母 {target} 被系统性地替代为 {produced}"
                suggestion = self._get_suggestion(target, produced)
                patterns.append(ErrorPattern(
                    pattern_name=pattern_name,
                    description=desc,
                    frequency=len(examples),
                    examples=examples[:5],
                    suggestion=suggestion,
                ))

        # 2. 省略模式
        omission_examples = [
            r.target_syllable for r in responses
            if r.error_type == ErrorType.OMIT
        ]
        if omission_examples: 17023.baike.tongsou.com 
            patterns.append(ErrorPattern(
                pattern_name="声母/韵母省略",
                description="存在音位省略现象",
                frequency=len(omission_examples),
                examples=omission_examples[:5],
                suggestion="加强目标音位的感知训练与产出练习",
            ))

        # 3. 歪曲模式
        distortion_examples = [
            r.target_syllable for r in responses
            if r.error_type == ErrorType.DISTORTION
        ]
        if distortion_examples: 17021.baike.tongsou.com 
            patterns.append(ErrorPattern(
                pattern_name="音位歪曲",
                description="存在音位歪曲现象，发音位置或方法不准确",
                frequency=len(distortion_examples),
                examples=distortion_examples[:5],
                suggestion="进行构音器官运动训练与目标音位定位练习",
            ))

        # 按频率降序排列
        patterns.sort(key=lambda p: p.frequency, reverse=True)
        return patterns

    def _get_suggestion(self, target: str, produced: str) -> str: 17019.baike.tongsou.com 
        """根据替代类型给出训练建议"""
        # 常见构音错误建议库
        suggestions = {
            ("g", "d"): "进行舌根音/g/的定位训练，可使用压舌板辅助感知舌根抬起",
            ("k", "t"): "进行舌根送气音/k/的定位训练，感受气流从舌根与软腭间通过",
            ("d", "g"): "注意舌尖与上齿龈的接触，避免舌根过早抬起",
            ("zh", "z"): "进行翘舌音/zh/的训练，舌尖上翘抵住硬腭前部",
            ("ch", "c"): "进行翘舌送气音/ch/的训练，注意舌尖位置",
            ("sh", "s"): "进行翘舌擦音/sh/的训练，舌尖上翘接近硬腭",
            ("r", "l"): "进行/r/音的训练，舌尖上翘不接触硬腭，声带振动",
            ("j", "z"): "注意舌面前部与硬腭的接触，避免舌尖前伸",
        }
        return suggestions.get((target, produced),
                               f"针对 {target}→{produced} 的替代进行专项构音训练")

    def compute_intelligibility(
        self, responses: List[SyllableResponse], domain: str = AssessmentDomain.SENTENCE
    ) -> IntelligibilityScore: 17018.baike.tongsou.com 
        """计算可懂度"""
        total = len(responses)
        intelligible = sum(1 for r in responses if r.intelligible)
        rate = intelligible / total if total > 0 else 0.0
        return IntelligibilityScore(
            domain=domain,
            total_syllables=total,
            intelligible_count=intelligible,
            intelligibility_rate=rate,
        )

    def determine_overall_severity(
        self, inventories: List[PhonemeInventory]
    ) -> str:
        """综合多个维度的评估结果，确定整体严重程度"""
        if not inventories: 17017.baike.tongsou.com 
            return "unknown"
        avg_accuracy = sum(inv.accuracy for inv in inventories) / len(inventories)
        acc_pct = avg_accuracy * 100
        for level, (low, high) in SEVERITY_THRESHOLDS.items(): 17016.baike.tongsou.com 
            if low <= acc_pct <= high: 17015.baike.tongsou.com 
                return level
        return "unknown"

    def generate_summary(
        self,
        patient_name: str,
        birth_date: date,
        assessment_date: date,
        inventories: List[PhonemeInventory],
        intelligibility_scores: List[IntelligibilityScore],
        error_patterns: List[ErrorPattern],
        overall_severity: str,
    ) -> str: 17014.baike.tongsou.com 
        """生成评估总结文本"""
        age_years = (assessment_date - birth_date).days // 365
        severity_cn = {
            "normal": "正常范围",
            "mild": "轻度",
            "moderate": "中度",
            "severe": "重度",
            "profound": "极重度",
        }.get(overall_severity, "未知")

        lines = [
            f"患者 {patient_name}，{age_years}岁，于 {assessment_date.isoformat()} 接受构音评估。",
            "",
            "【评估结果】",
        ]
        for inv in inventories: 17013.baike.tongsou.com 
            lines.append(
                f"- {inv.domain}：共 {inv.total_items} 项，"
                f"正确 {inv.correct_count} 项，正确率 {inv.accuracy:.1%}，"
                f"严重程度：{inv.severity}"
            )

        if intelligibility_scores: 17012.baike.tongsou.com  
            lines.append("")
            lines.append("【可懂度】")
            for score in intelligibility_scores: 17011.baike.tongsou.com 
                lines.append(
                    f"- {score.domain}：{score.intelligible_count}/{score.total_syllables} "
                    f"({score.percentage:.1f}%)"
                )

        if error_patterns: 17009.baike.tongsou.com 
            lines.append("")
            lines.append("【主要错误模式】")
            for i, pat in enumerate(error_patterns, 1): 17010.baike.tongsou.com 
                lines.append(
                    f"{i}. {pat.pattern_name}（出现 {pat.frequency} 次）"
                )
                lines.append(f"   说明：{pat.description}")
                if pat.examples: 17008.baike.tongsou.com 
                    lines.append(f"   示例：{', '.join(pat.examples[:3])}")

        lines.append("")
        lines.append(f"【综合结论】构音障碍严重程度：{severity_cn}。")
        return "\n".join(lines)

    def generate_recommendations(
        self, error_patterns: List[ErrorPattern], overall_severity: str
    ) -> List[str]: 17007.baike.tongsou.com 
        """根据评估结果生成训练建议"""
        recs = []

        # 根据严重程度给出总体建议
        severity_rec = {
            "mild": "建议每周1-2次个别训练，配合家庭练习。",
            "moderate": "建议每周2-3次个别训练，重点突破主要错误模式。",
            "severe": "建议每周3次以上密集训练，同时加强构音器官运动能力。",
            "profound": "建议密集个别训练，必要时结合辅助沟通系统。",
        }
        if overall_severity in severity_rec:
            recs.append(severity_rec[overall_severity])

        # 针对每个错误模式给出建议
        for pat in error_patterns[:3]:  # 最多取前3个
            recs.append(f"针对「{pat.pattern_name}」：{pat.suggestion}")

        # 通用建议
        recs.append("建议家长/照护者在家中每日进行10-15分钟的目标音位练习。")
        recs.append("每8-12周进行一次复评，监测进展。")
        return recs


# ==================== 报告生成器 ====================

class ReportGenerator: 17006.baike.tongsou.com 
    """报告生成器"""

    def __init__(self): 17005.baike.tongsou.com 
        self.logger = logging.getLogger(__name__)

    def generate_json_report(self, record: AssessmentRecord) -> str: 17004.baike.tongsou.com 
        """生成JSON格式报告"""
        return json.dumps(record.to_dict(), ensure_ascii=False, indent=2)

    def generate_markdown_report(self, record: AssessmentRecord) -> str:
        """生成Markdown格式报告"""
        lines = [
            f"# 构音障碍评估报告",
            "",
            f"**患者姓名**：{record.patient_name}  ",
            f"**患者ID**：{record.patient_id}  ",
            f"**出生日期**：{record.birth_date.isoformat()}  ",
            f"**评估日期**：{record.assessment_date.isoformat()}  ",
            f"**评估治疗师**：{record.assessor}  ",
            f"**报告ID**：{record.record_id}  ",
            "",
            "---",
            "",
            "## 一、评估结果",
            "",
        ]

        for inv in record.phoneme_inventories:
            lines.append(f"### {inv.domain}")
            lines.append("")
            lines.append(f"| 项目 | 数值 |")
            lines.append(f"|------|------|")
            lines.append(f"| 总项数 | {inv.total_items} |")
            lines.append(f"| 正确数 | {inv.correct_count} |")
            lines.append(f"| 错误数 | {inv.error_count} |")
            lines.append(f"| 正确率 | {inv.accuracy:.1%} |")
            lines.append(f"| 严重程度 | {inv.severity} |")
            lines.append("")

            if inv.errors_by_type: 17003.baike.tongsou.com 
                lines.append("**错误类型分布**：")
                for etype, count in inv.errors_by_type.items():
                    lines.append(f"- {etype}：{count} 次")
                lines.append("")

        if record.intelligibility_scores:
            lines.append("## 二、言语可懂度")
            lines.append("")
            for score in record.intelligibility_scores:
                lines.append(
                    f"- **{score.domain}**：{score.intelligible_count}/{score.total_syllables} "
                    f"({score.percentage:.1f}%)"
                )
            lines.append("")

        if record.error_patterns:
            lines.append("## 三、主要错误模式")
            lines.append("")
            for i, pat in enumerate(record.error_patterns, 1): 17002.baike.tongsou.com 
                lines.append(f"### {i}. {pat.pattern_name}")
                lines.append("")
                lines.append(f"- **出现次数**：{pat.frequency}")
                lines.append(f"- **说明**：{pat.description}")
                if pat.examples:
                    lines.append(f"- **示例**：{', '.join(pat.examples[:5])}")
                lines.append(f"- **训练建议**：{pat.suggestion}")
                lines.append("")

        lines.append("## 四、综合结论")
        lines.append("")
        lines.append(f"构音障碍整体严重程度：**{record.overall_severity}**")
        lines.append("")
        lines.append(record.summary)
        lines.append("")

        if record.recommendations: 17001.baike.tongsou.com  
         lines.append("## 五、训练建议")
            lines.append("")
            for i, rec in enumerate(record.recommendations, 1):
                lines.append(f"{i}. {rec}")
            lines.append("")

        lines.append("---")
        lines.append(f"*报告生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
        return "\n".join(lines)

    def generate_text_report(self, record: AssessmentRecord) -> str:
        """生成纯文本报告"""
        lines = [
            "=" * 60,
            "           构音障碍评估报告",
            "=" * 60,
            "",
            f"患者姓名：{record.patient_name}",
            f"患者ID：  {record.patient_id}",
            f"出生日期：{record.birth_date.isoformat()}",
            f"评估日期：{record.assessment_date.isoformat()}",
            f"评估治疗师：{record.assessor}",
            f"报告ID：  {record.record_id}",
            "",
            "-" * 60,
            "",
        ]

        # 评估结果
        lines.append("【评估结果】")
        lines.append("")
        for inv in record.phoneme_inventories: 16064.baike.tongsou.com 
            lines.append(f"  {inv.domain}：")
            lines.append(f"    总项数：{inv.total_items}")
            lines.append(f"    正确数：{inv.correct_count}")
            lines.append(f"    错误数：{inv.error_count}")
            lines