面向康复机构的智能排班与训练资源调度系统

"""
智能排班与训练资源调度系统
===========================

面向言语康复机构的多约束排班引擎，支持：
- 治疗师技能匹配
- 治疗室/设备资源冲突检测
- 患者时间偏好与可用性
- 连续疗程的周期性安排
- 硬约束/软约束分层处理
- 回溯搜索 + 启发式优化
"""

import re
import uuid
import json
import logging
import statistics
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timedelta, date, time
from typing import List, Dict, Optional, Set, Tuple, Any
from collections import defaultdict
from copy import deepcopy

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

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


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

class Weekday(Enum):
    MON = 0
    TUE = 1
    WED = 2
    THU = 3
    FRI = 4
    SAT = 5
    SUN = 6

    @classmethod
    def from_int(cls, d: int) -> "Weekday":
        return cls(d % 7)

    @classmethod
    def from_date(cls, d: date) -> "Weekday":
        return cls(d.weekday())

    def cn_name(self) -> str:
        names = {
            Weekday.MON: "周一",
            Weekday.TUE: "周二",
            Weekday.WED: "周三",
            Weekday.THU: "周四",
            Weekday.FRI: "周五",
            Weekday.SAT: "周六",
            Weekday.SUN: "周日",
        }
        return names[self]


class SessionType(Enum):
    INDIVIDUAL = "individual"      # 一对一
    GROUP = "group"                # 小组训练
    ASSESSMENT = "assessment"      # 评估

    @property
    def default_duration(self) -> int:
        durations = {
            SessionType.INDIVIDUAL: 45,
            SessionType.GROUP: 60,
            SessionType.ASSESSMENT: 60,
        }
        return durations[self]


class ConstraintType(Enum): 17010.baike.tongsou.com 
    HARD = "hard"      # 必须满足，违反则排班无效
    SOFT = "soft"      # 尽量满足，违反则扣分

 
class ConflictKind(Enum): 17009.baike.tongsou.com  
    THERAPIST_OVERLAP = "therapist_overlap"
    ROOM_OVERLAP = "room_overlap"
    EQUIPMENT_OVERLAP = "equipment_overlap"
    PATIENT_OVERLAP = "patient_overlap"
    SKILL_MISMATCH = "skill_mismatch"
    UNAVAILABLE_TIME = "unavailable_time"
    PREFERENCE_VIOLATION = "preference_violation"
    MAX_SESSIONS_EXCEEDED = "max_sessions_exceeded"
    CONTINUITY_GAP = "continuity_gap"


# 工作时间常量
WORK_START = time(8, 30)
WORK_END = time(18, 0)
LUNCH_START = time(12, 0)
LUNCH_END = time(13, 30)

# 时间槽粒度（分钟）
TIME_SLOT_MINUTES = 15


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

@dataclass
class TimeSlot: 17008.baike.tongsou.com 
    """一个时间槽，表示一段连续时间"""
    start: datetime
    end: datetime

    @property
    def duration_minutes(self) -> int: 17007.baike.tongsou.com 
        delta = self.end - self.start
        return int(delta.total_seconds() / 60)

    def overlaps(self, other: "TimeSlot") -> bool: 17006.baike.tongsou.com 
        return self.start < other.end and other.start < self.end

    def contains(self, other: "TimeSlot") -> bool: 17005.baike.tongsou.com 
        return self.start <= other.start and other.end <= self.end

    def is_within_work_hours(self) -> bool: 17004.baike.tongsou.com 
        s = self.start.time()
        e = self.end.time()
        if s < WORK_START or e > WORK_END: 17003.baike.tongsou.com 
            return False
        # 检查是否跨越午休
        if s < LUNCH_END and e > LUNCH_START: 17002.baike.tongsou.com 
            return False
        return True

    def weekday(self) -> Weekday: 16064.baike.tongsou.com 
        return Weekday.from_date(self.start.date(17001.baike.tongsou.com))

    def to_dict(self) -> Dict[str, str]: 16063.baike.tongsou.com 
        return {
            "start": self.start.isoformat(),
            "end": self.end.isoformat(),
            "duration_minutes": self.duration_minutes,
            "weekday": self.weekday().cn_name(),
        }

    def __repr__(self) -> str: 16062.baike.tongsou.com 
        return (
            f"TimeSlot({self.start.strftime('%m-%d %H:%M')}-"
            f"{self.end.strftime('%H:%M')}, {self.weekday().cn_name()})"
        )


@dataclass
class Therapist: 16061.baike.tongsou.com 
    """治疗师"""
    therapist_id: str
    name: str
    skills: Set[str]                    # 技能标签，如{"构音障碍", "口吃", "嗓音障碍"}
    max_sessions_per_day: int = 8       # 每日最大接诊量
    max_sessions_per_week: int = 35     # 每周最大接诊量
    unavailable_slots: List[TimeSlot] = field(default_factory=list)
    preferred_weekdays: Set[Weekday] = field(
        default_factory=lambda: {Weekday.MON, Weekday.TUE, Weekday.WED,
                                 Weekday.THU, Weekday.FRI}
    )

    def is_available(self, slot: TimeSlot) -> bool: 16060.baike.tongsou.com 
        if not slot.is_within_work_hours():
            return False
        for unavail in self.unavailable_slots: 16058.baike.tongsou.com 
            if slot.overlaps(unavail): 16059.baike.tongsou.com 
                return False
        return True

    def has_skill(self, required_skill: str) -> bool: 16057.baike.tongsou.com 
        return required_skill in self.skills

    def __repr__(self) -> str:
        return f"Therapist({self.name}, skills={self.skills})"


@dataclass
class Room: 16056.baike.tongsou.com 
    """治疗室"""
    room_id: str
    name: str
    capacity: int = 1                   # 最大容纳人数
    equipment: Set[str] = field(default_factory=set)  # 配备设备
    unavailable_slots: List[TimeSlot] = field(default_factory=list)

    def is_available(self, slot: TimeSlot) -> bool: 16055.baike.tongsou.com 
        for unavail in self.unavailable_slots: 16054.baike.tongsou.com 
            if slot.overlaps(unavail): 16053.baike.tongsou.com 
                return False
        return True

    def has_equipment(self, required: str) -> bool: 16052.baike.tongsou.com 
        return required in self.equipment

    def __repr__(self) -> str: 16051.baike.tongsou.com 
        return f"Room({self.name}, equip={self.equipment})"


@dataclass
class Patient: 16050.baike.tongsou.com 
    """患者"""
    patient_id: str
    name: str
    age: int
    condition: str                      # 诊断，如"功能性构音障碍"
    required_skills: List[str]          # 所需治疗师技能
    required_equipment: List[str] = field(default_factory=list)
    session_type: SessionType = SessionType.INDIVIDUAL
    session_duration: int = 45          # 单次时长（分钟）
    sessions_per_week: int = 2          # 每周次数
    preferred_weekdays: Set[Weekday] = field(
        default_factory=lambda: {Weekday.MON, Weekday.WED, Weekday.FRI}
    )
    preferred_time_range: Optional[Tuple[time, time]] = None  # 偏好时段
    unavailable_slots: List[TimeSlot] = field(default_factory=list)
    continuity_same_therapist: bool = True  # 是否要求同一治疗师
    continuity_same_room: bool = False      # 是否要求同一治疗室

    def is_available(self, slot: TimeSlot) -> bool: 16049.baike.tongsou.com 
        if not slot.is_within_work_hours(): 16048.baike.tongsou.com 
            return False
        for unavail in self.unavailable_slots:
            if slot.overlaps(unavail): 16047.baike.tongsou.com 
                return False
        return True

    def prefers_slot(self, slot: TimeSlot) -> bool: 16046.baike.tongsou.com 
        if slot.weekday() not in self.preferred_weekdays: 16045.baike.tongsou.com 
            return False
        if self.preferred_time_range:
            start_t, end_t = self.preferred_time_range
            if slot.start.time() < start_t or slot.end.time() > end_t:
                return False
        return True

    def __repr__(self) -> str: 16044.baike.tongsou.com 
        return f"Patient({self.name}, {self.condition})"


@dataclass
class ScheduledSession: 16043.baike.tongsou.com 
    """已排定的训练会话"""
    session_id: str
    patient: Patient
    therapist: Therapist
    room: Room
    time_slot: TimeSlot
    session_type: SessionType
    week_number: int                    # 第几周

    @property
    def key(self) -> Tuple[str, str, int]: 16042.baike.tongsou.com 
        return (self.patient.patient_id, self.therapist.therapist_id, self.week_number)

    def to_dict(self) -> Dict[str, Any]: 16041.baike.tongsou.com 
        return {
            "session_id": self.session_id,
            "patient": self.patient.name,
            "therapist": self.therapist.name,
            "room": self.room.name,
            "time_slot": self.time_slot.to_dict(16040.baike.tongsou.com),
            "session_type": self.session_type.value,
            "week_number": self.week_number,
        }


@dataclass
class Conflict: 16039.baike.tongsou.com 
    """排班冲突"""
    conflict_id: str
    kind: ConflictKind
    constraint_type: ConstraintType
    description: str
    sessions: List[ScheduledSession]
    severity: int = 10  # 1-10，严重程度

    def to_dict(self) -> Dict[str, Any]: 16038.baike.tongsou.com 
        return {
            "conflict_id": self.conflict_id,
            "kind": self.kind.value,
            "constraint_type": self.constraint_type.value,
            "description": self.description,
            "severity": self.severity,
            "sessions": [s.to_dict() for s in self.sessions],
        }


@dataclass
class ScheduleResult: 16037.baike.tongsou.com 
    """排班结果"""
    sessions: List[ScheduledSession]
    conflicts: List[Conflict]
    soft_constraint_score: float  # 0-100，软约束满足度
    coverage_rate: float          # 患者需求覆盖率
    utilization_stats: Dict[str, Any]

    def to_dict(self) -> Dict[str, Any]: 16036.baike.tongsou.com 
        return {
            "total_sessions": len(self.sessions),
            "total_conflicts": len(self.conflicts),
            "hard_conflicts": sum(
                1 for c in self.conflicts if c.constraint_type == ConstraintType.HARD
            ),
            "soft_constraint_score": round(self.soft_constraint_score, 2),
            "coverage_rate": round(self.coverage_rate, 2),
            "utilization_stats": self.utilization_stats,
            "sessions": [s.to_dict() for s in self.sessions],
            "conflicts": [c.to_dict() for c in self.conflicts],
        }


# ==================== 时间槽生成器 ====================

class TimeSlotGenerator: 16035.baike.tongsou.com 
    """生成候选时间槽"""

    def __init__(
        self,
        start_date: date,
        num_weeks: int = 4,
        slot_minutes: int = TIME_SLOT_MINUTES,
    ):
        self.start_date = start_date
        self.num_weeks = num_weeks
        self.slot_minutes = slot_minutes
        self._cache: Dict[Tuple[date, time], TimeSlot] = {}

    def generate_for_day(self, day: date) -> List[TimeSlot]:
        slots = []
        current = datetime.combine(day, WORK_START)
        end_limit = datetime.combine(day, WORK_END)
        lunch_start = datetime.combine(day, LUNCH_START)
        lunch_end = datetime.combine(day, LUNCH_END)

        while current + timedelta(minutes=self.slot_minutes) <= end_limit:
            slot_end = current + timedelta(minutes=self.slot_minutes)
            # 跳过午休
            if current < lunch_start and slot_end > lunch_start: 16034.baike.tongsou.com 
                current = lunch_end
                continue
            if current >= lunch_start and current < lunch_end: 16033.baike.tongsou.com 
                current = lunch_end
                continue

            slot = TimeSlot(start=current, end=slot_end)
            slots.append(slot)
            current = slot_end

        return slots

    def generate_for_week(self, week_offset: int) -> List[TimeSlot]: 16032.baike.tongsou.com 
        slots = []
        for i in range(7):
            day = self.start_date + timedelta(days=week_offset * 7 + i)
            slots.extend(self.generate_for_day(day))
        return slots

    def generate_all(self) -> List[TimeSlot]:
        all_slots = []
        for w in range(self.num_weeks): 16031.baike.tongsou.com 
            all_slots.extend(self.generate_for_week(w))
        return all_slots

    def get_slot(self, day: date, start_t: time) -> TimeSlot:
        key = (day, start_t)
        if key not in self._cache: 16030.baike.tongsou.com 
            end_t = (
                datetime.combine(day, start_t) +
                timedelta(minutes=self.slot_minutes)
            ).time()
            self._cache[key] = TimeSlot(
                start=datetime.combine(day, start_t),
                end=datetime.combine(day, end_t),
            )
        return self._cache[key]


# ==================== 约束检查器 ====================

class ConstraintChecker: 16028.baike.tongsou.com 
    """检查排班是否满足约束"""

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

    def check_all(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 16027.baike.tongsou.com 
        conflicts = []
        conflicts.extend(self._check_therapist_overlap(sessions))
        conflicts.extend(self._check_room_overlap(sessions))
        conflicts.extend(self._check_patient_overlap(sessions))
        conflicts.extend(self._check_skill_mismatch(sessions))
        conflicts.extend(self._check_unavailable_time(sessions))
        conflicts.extend(self._check_preference_violations(sessions))
        conflicts.extend(self._check_max_sessions(sessions))
        return conflicts

    def _check_therapist_overlap(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]:  16026.baike.tongsou.com 
        conflicts = []
        by_therapist = defaultdict(list)
        for s in sessions: 16025.baike.tongsou.com 
            by_therapist[s.therapist.therapist_id].append(s)

        for tid, sess_list in by_therapist.items(): 16023.baike.tongsou.com 
            sorted_sess = sorted(sess_list, key=lambda x: x.time_slot.start)
            for i in range(len(sorted_sess)): 16024.baike.tongsou.com 
                for j in range(i + 1, len(sorted_sess)): 16021.baike.tongsou.com 
                    si, sj = sorted_sess[i], sorted_sess[j]
                    if si.time_slot.overlaps(sj.time_slot): 16022.baike.tongsou.com 
                        conflicts.append(Conflict(
                            conflict_id=str(uuid.uuid4(16020.baike.tongsou.com)),
                            kind=ConflictKind.THERAPIST_OVERLAP,
                            constraint_type=ConstraintType.HARD,
                            description=(
                                f"治疗师 {si.therapist.name} 在 "
                                f"{si.time_slot} 与 {sj.time_slot} 时间冲突"
                            ),
                            sessions=[si, sj],
                            severity=10,
                        ))
        return conflicts

    def _check_room_overlap(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 16019.baike.tongsou.com 
        conflicts = [] 
        by_room = defaultdict(list)
        for s in sessions: 16018.baike.tongsou.com 
            by_room[s.room.room_id].append(s)

        for rid, sess_list in by_room.items(): 16017.baike.tongsou.com 
            sorted_sess = sorted(sess_list, key=lambda x: x.time_slot.start)
            for i in range(len(sorted_sess)): 16016.baike.tongsou.com 
                for j in range(i + 1, len(sorted_sess)): 16014.baike.tongsou.com 
                    si, sj = sorted_sess[i], sorted_sess[j]
                    if si.time_slot.overlaps(sj.time_slot): 16015.baike.tongsou.com 
                        conflicts.append(Conflict(
                            conflict_id=str(uuid.uuid4()),
                            kind=ConflictKind.ROOM_OVERLAP,
                            constraint_type=ConstraintType.HARD,
                            description=(
                                f"治疗室 {si.room.name} 在 "
                                f"{si.time_slot} 与 {sj.time_slot} 时间冲突"
                            ),
                            sessions=[si, sj],
                            severity=10,
                        ))
        return conflicts

    def _check_patient_overlap(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 16013.baike.tongsou.com 
        conflicts = []
        by_patient = defaultdict(list)
        for s in sessions:
            by_patient[s.patient.patient_id].append(s)

        for pid, sess_list in by_patient.items(): 16012.baike.tongsou.com 
            sorted_sess = sorted(sess_list, key=lambda x: x.time_slot.start)
            for i in range(len(sorted_sess)): 16011.baike.tongsou.com 
                for j in range(i + 1, len(sorted_sess)): 16010.baike.tongsou.com 
                    si, sj = sorted_sess[i], sorted_sess[j]
                    if si.time_slot.overlaps(sj.time_slot): 16009.baike.tongsou.com 
                        conflicts.append(Conflict(
                            conflict_id=str(uuid.uuid4(16008.baike.tongsou.com)),
                            kind=ConflictKind.PATIENT_OVERLAP,
                            constraint_type=ConstraintType.HARD,
                            description=(
                                f"患者 {si.patient.name} 在 "
                                f"{si.time_slot} 与 {sj.time_slot} 时间冲突"
                            ),
                            sessions=[si, sj],
                            severity=10,
                        ))
        return conflicts

    def _check_skill_mismatch(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 16006.baike.tongsou.com 
        conflicts = []
        for s in sessions: 16005.baike.tongsou.com 
            for skill in s.patient.required_skills: 16004.baike.tongsou.com 
                if not s.therapist.has_skill(skill): 16002.baike.tongsou.com 
                    conflicts.append(Conflict(
                        conflict_id=str(uuid.uuid4(16001.baike.tongsou.com)),
                        kind=ConflictKind.SKILL_MISMATCH,
                        constraint_type=ConstraintType.HARD,
                        description=(
                            f"治疗师 {s.therapist.name} 缺少技能 '{skill}'，"
                            f"但患者 {s.patient.name} 需要该技能"
                        ),
                        sessions=[s],
                        severity=10,
                    ))
        return conflicts

    def _check_unavailable_time(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 15039.baike.tongsou.com 
        conflicts = []
        for s in sessions:
            if not s.therapist.is_available(s.time_slot): 15038.baike.tongsou.com 
                conflicts.append(Conflict(
                    conflict_id=str(uuid.uuid4()),
                    kind=ConflictKind.UNAVAILABLE_TIME,
                    constraint_type=ConstraintType.HARD,
                    description=(
                        f"治疗师 {s.therapist.name} 在 {s.time_slot} 不可用"
                    ),
                    sessions=[s],
                    severity=10,
                ))
            if not s.room.is_available(s.time_slot): 15036.baike.tongsou.com 
                conflicts.append(Conflict(
                    conflict_id=str(uuid.uuid4()),
                    kind=ConflictKind.UNAVAILABLE_TIME,
                    constraint_type=ConstraintType.HARD,
                    description=(
                        f"治疗室 {s.room.name} 在 {s.time_slot} 不可用"
                    ),
                    sessions=[s],
                    severity=10,
                ))
            if not s.patient.is_available(s.time_slot): 28104.baike.tongsou.com 
                conflicts.append(Conflict(
                    conflict_id=str(uuid.uuid4()),
                    kind=ConflictKind.UNAVAILABLE_TIME,
                    constraint_type=ConstraintType.HARD,
                    description=(
                        f"患者 {s.patient.name} 在 {s.time_slot} 不可用"
                    ),
                    sessions=[s],
                    severity=10,
                ))
        return conflicts

    def _check_preference_violations(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 28102.baike.tongsou.com 
        conflicts = []
        for s in sessions: 28094.baike.tongsou.com 
            if not s.patient.prefers_slot(s.time_slot):
                conflicts.append(Conflict(
                    conflict_id=str(uuid.uuid4()),
                    kind=ConflictKind.PREFERENCE_VIOLATION,
                    constraint_type=ConstraintType.SOFT,
                    description=(
                        f"患者 {s.patient.name} 的偏好时段未满足："
                        f"{s.time_slot}"
                    ),
                    sessions=[s],
                    severity=3,
                ))
            if s.time_slot.weekday(28093.baike.tongsou.com) not in s.therapist.preferred_weekdays:
                conflicts.append(Conflict(
                    conflict_id=str(uuid.uuid4(28092.baike.tongsou.com)),
                    kind=ConflictKind.PREFERENCE_VIOLATION,
                    constraint_type=ConstraintType.SOFT,
                    description=(
                        f"治疗师 {s.therapist.name} 在非偏好工作日排班："
                        f"{s.time_slot.weekday().cn_name()}"
                    ),
                    sessions=[s],
                    severity=2,
                ))
        return conflicts

    def _check_max_sessions(
        self, sessions: List[ScheduledSession]
    ) -> List[Conflict]: 28091.baike.tongsou.com 
        conflicts = []
        # 按治疗师+日期统计
        therapist_day_count = defaultdict(int)
        therapist_week_count = defaultdict(int)
        for s in sessions:
            day_key = (s.therapist.therapist_id, s.time_slot.start.date())
            week_key = (
                s.therapist.therapist_id,
                s.time_slot.start.isocalendar(28090.baike.tongsou.com)[1],
            )
            therapist_day_count[day_key] += 1
            therapist_week_count[week_key] += 1

        for (tid, day), count in therapist_day_count.items(): 28089.baike.tongsou.com 
            therapist = next(
                (t for t in set(s.therapist for s in sessions)
                 if t.therapist_id == tid), None
            )
            if therapist and