|
| 1 | +# ================================================== |
| 2 | +# CSES (Course Schedule Exchange Schema) 解析器 |
| 3 | +# ================================================== |
| 4 | +import yaml |
| 5 | +import json |
| 6 | +from datetime import datetime, time |
| 7 | +from typing import Dict, List, Optional |
| 8 | +from loguru import logger |
| 9 | + |
| 10 | + |
| 11 | +class CSESParser: |
| 12 | + """CSES格式课程表解析器""" |
| 13 | + |
| 14 | + def __init__(self): |
| 15 | + self.schedule_data = None |
| 16 | + |
| 17 | + def load_from_file(self, file_path: str) -> bool: |
| 18 | + """从文件加载CSES数据 |
| 19 | +
|
| 20 | + Args: |
| 21 | + file_path: CSES文件路径 |
| 22 | +
|
| 23 | + Returns: |
| 24 | + bool: 加载成功返回True,否则返回False |
| 25 | + """ |
| 26 | + try: |
| 27 | + with open(file_path, "r", encoding="utf-8") as f: |
| 28 | + self.schedule_data = yaml.safe_load(f) |
| 29 | + return self._validate_schedule() |
| 30 | + except Exception as e: |
| 31 | + logger.error(f"加载CSES文件失败: {e}") |
| 32 | + return False |
| 33 | + |
| 34 | + def load_from_content(self, content: str) -> bool: |
| 35 | + """从字符串内容加载CSES数据 |
| 36 | +
|
| 37 | + Args: |
| 38 | + content: YAML格式的CSES内容 |
| 39 | +
|
| 40 | + Returns: |
| 41 | + bool: 加载成功返回True,否则返回False |
| 42 | + """ |
| 43 | + try: |
| 44 | + self.schedule_data = yaml.safe_load(content) |
| 45 | + return self._validate_schedule() |
| 46 | + except Exception as e: |
| 47 | + logger.error(f"解析CSES内容失败: {e}") |
| 48 | + return False |
| 49 | + |
| 50 | + def _validate_schedule(self) -> bool: |
| 51 | + """验证课程表数据的有效性 |
| 52 | +
|
| 53 | + Returns: |
| 54 | + bool: 数据有效返回True,否则返回False |
| 55 | + """ |
| 56 | + if not self.schedule_data: |
| 57 | + logger.error("课程表数据为空") |
| 58 | + return False |
| 59 | + |
| 60 | + # 基本结构验证 |
| 61 | + if "schedule" not in self.schedule_data: |
| 62 | + logger.error("缺少'schedule'字段") |
| 63 | + return False |
| 64 | + |
| 65 | + schedule = self.schedule_data["schedule"] |
| 66 | + if not isinstance(schedule, dict): |
| 67 | + logger.error("'schedule'字段必须是字典类型") |
| 68 | + return False |
| 69 | + |
| 70 | + # 验证时间段配置 |
| 71 | + if "timeslots" not in schedule: |
| 72 | + logger.error("缺少'timeslots'字段") |
| 73 | + return False |
| 74 | + |
| 75 | + timeslots = schedule["timeslots"] |
| 76 | + if not isinstance(timeslots, list): |
| 77 | + logger.error("'timeslots'字段必须是列表类型") |
| 78 | + return False |
| 79 | + |
| 80 | + # 验证每个时间段 |
| 81 | + for i, timeslot in enumerate(timeslots): |
| 82 | + if not self._validate_timeslot(timeslot, i): |
| 83 | + return False |
| 84 | + |
| 85 | + return True |
| 86 | + |
| 87 | + def _validate_timeslot(self, timeslot: dict, index: int) -> bool: |
| 88 | + """验证单个时间段的配置 |
| 89 | +
|
| 90 | + Args: |
| 91 | + timeslot: 时间段配置字典 |
| 92 | + index: 时间段索引 |
| 93 | +
|
| 94 | + Returns: |
| 95 | + bool: 有效返回True,否则返回False |
| 96 | + """ |
| 97 | + required_fields = ["name", "start_time", "end_time"] |
| 98 | + for field in required_fields: |
| 99 | + if field not in timeslot: |
| 100 | + logger.error(f"时间段{index}缺少'{field}'字段") |
| 101 | + return False |
| 102 | + |
| 103 | + # 验证时间格式 |
| 104 | + try: |
| 105 | + start_time = self._parse_time(timeslot["start_time"]) |
| 106 | + end_time = self._parse_time(timeslot["end_time"]) |
| 107 | + |
| 108 | + if start_time >= end_time: |
| 109 | + logger.error(f"时间段{index}的开始时间必须早于结束时间") |
| 110 | + return False |
| 111 | + |
| 112 | + except ValueError as e: |
| 113 | + logger.error(f"时间段{index}时间格式错误: {e}") |
| 114 | + return False |
| 115 | + |
| 116 | + return True |
| 117 | + |
| 118 | + def _parse_time(self, time_str: str) -> time: |
| 119 | + """解析时间字符串 |
| 120 | +
|
| 121 | + Args: |
| 122 | + time_str: 时间字符串 (HH:MM 或 HH:MM:SS) |
| 123 | +
|
| 124 | + Returns: |
| 125 | + time: 时间对象 |
| 126 | +
|
| 127 | + Raises: |
| 128 | + ValueError: 时间格式错误 |
| 129 | + """ |
| 130 | + try: |
| 131 | + if ":" in time_str: |
| 132 | + parts = time_str.split(":") |
| 133 | + if len(parts) == 2: |
| 134 | + return time(int(parts[0]), int(parts[1])) |
| 135 | + elif len(parts) == 3: |
| 136 | + return time(int(parts[0]), int(parts[1]), int(parts[2])) |
| 137 | + raise ValueError(f"无效的时间格式: {time_str}") |
| 138 | + except (ValueError, IndexError): |
| 139 | + raise ValueError(f"无法解析时间: {time_str}") |
| 140 | + |
| 141 | + def get_non_class_times(self) -> Dict[str, str]: |
| 142 | + """获取非上课时间段配置 |
| 143 | +
|
| 144 | + 将CSES格式的时间段转换为SecRandom使用的非上课时间段格式 |
| 145 | +
|
| 146 | + Returns: |
| 147 | + Dict[str, str]: 非上课时间段字典,格式为 {"name": "HH:MM:SS-HH:MM:SS"} |
| 148 | + """ |
| 149 | + if not self.schedule_data: |
| 150 | + return {} |
| 151 | + |
| 152 | + non_class_times = {} |
| 153 | + schedule = self.schedule_data["schedule"] |
| 154 | + timeslots = schedule["timeslots"] |
| 155 | + |
| 156 | + # 按开始时间排序 |
| 157 | + sorted_timeslots = sorted(timeslots, key=lambda x: x["start_time"]) |
| 158 | + |
| 159 | + # 构建上课时间段列表 |
| 160 | + class_periods = [] |
| 161 | + for timeslot in sorted_timeslots: |
| 162 | + start_time = self._format_time_for_secrandom(timeslot["start_time"]) |
| 163 | + end_time = self._format_time_for_secrandom(timeslot["end_time"]) |
| 164 | + class_periods.append((start_time, end_time)) |
| 165 | + |
| 166 | + # 生成非上课时间段 |
| 167 | + # 1. 第一节课之前的时间 |
| 168 | + if class_periods: |
| 169 | + first_start = class_periods[0][0] |
| 170 | + if first_start != "00:00:00": |
| 171 | + non_class_times["before_first_class"] = f"00:00:00-{first_start}" |
| 172 | + |
| 173 | + # 2. 课间时间(两节课之间) |
| 174 | + for i in range(len(class_periods) - 1): |
| 175 | + current_end = class_periods[i][1] |
| 176 | + next_start = class_periods[i + 1][0] |
| 177 | + if current_end != next_start: |
| 178 | + period_name = f"break_{i + 1}" |
| 179 | + non_class_times[period_name] = f"{current_end}-{next_start}" |
| 180 | + |
| 181 | + # 3. 最后一节课之后的时间 |
| 182 | + if class_periods: |
| 183 | + last_end = class_periods[-1][1] |
| 184 | + if last_end != "23:59:59": |
| 185 | + non_class_times["after_last_class"] = f"{last_end}-23:59:59" |
| 186 | + |
| 187 | + logger.info(f"成功解析CSES课程表,生成{len(non_class_times)}个非上课时间段") |
| 188 | + return non_class_times |
| 189 | + |
| 190 | + def _format_time_for_secrandom(self, time_str: str) -> str: |
| 191 | + """将时间字符串格式化为SecRandom需要的格式 (HH:MM:SS) |
| 192 | +
|
| 193 | + Args: |
| 194 | + time_str: 原始时间字符串 (HH:MM 或 HH:MM:SS) |
| 195 | +
|
| 196 | + Returns: |
| 197 | + str: 格式化后的时间字符串 (HH:MM:SS) |
| 198 | + """ |
| 199 | + if time_str.count(":") == 1: # HH:MM 格式 |
| 200 | + return f"{time_str}:00" |
| 201 | + return time_str |
| 202 | + |
| 203 | + def get_class_info(self) -> List[Dict]: |
| 204 | + """获取课程信息列表 |
| 205 | +
|
| 206 | + Returns: |
| 207 | + List[Dict]: 课程信息列表 |
| 208 | + """ |
| 209 | + if not self.schedule_data: |
| 210 | + return [] |
| 211 | + |
| 212 | + schedule = self.schedule_data["schedule"] |
| 213 | + timeslots = schedule.get("timeslots", []) |
| 214 | + |
| 215 | + class_info = [] |
| 216 | + for timeslot in timeslots: |
| 217 | + info = { |
| 218 | + "name": timeslot.get("name", ""), |
| 219 | + "start_time": timeslot.get("start_time", ""), |
| 220 | + "end_time": timeslot.get("end_time", ""), |
| 221 | + "teacher": timeslot.get("teacher", ""), |
| 222 | + "location": timeslot.get("location", ""), |
| 223 | + "day_of_week": timeslot.get("day_of_week", ""), |
| 224 | + } |
| 225 | + class_info.append(info) |
| 226 | + |
| 227 | + return class_info |
| 228 | + |
| 229 | + def get_summary(self) -> str: |
| 230 | + """获取课程表摘要信息 |
| 231 | +
|
| 232 | + Returns: |
| 233 | + str: 摘要信息 |
| 234 | + """ |
| 235 | + if not self.schedule_data: |
| 236 | + return "未加载课程表" |
| 237 | + |
| 238 | + schedule = self.schedule_data["schedule"] |
| 239 | + timeslots = schedule.get("timeslots", []) |
| 240 | + |
| 241 | + if not timeslots: |
| 242 | + return "课程表为空" |
| 243 | + |
| 244 | + # 获取最早和最晚时间 |
| 245 | + start_times = [slot["start_time"] for slot in timeslots] |
| 246 | + end_times = [slot["end_time"] for slot in timeslots] |
| 247 | + |
| 248 | + summary = f"课程表包含{len(timeslots)}个时间段," |
| 249 | + summary += f"最早开始时间:{min(start_times)}," |
| 250 | + summary += f"最晚结束时间:{max(end_times)}" |
| 251 | + |
| 252 | + return summary |
0 commit comments