""" 该类负责整合`reader.py`以及`informer.py`完成读取与通知流程。 启动后先从reader读取完整课程信息,然后根据时间安排发送通知。 时间安排包括:当前属于第几周、今日的所有课程都会在何时开始。 其中课程的开始时间与持续时间需要根据reader中定义的CourseSchedule的timeslot: TimeRange字段来确定 比如:from=1,to=2,即对应第一节课到第二节课(包括第二节) 而时间详细安排如下: 上午: 08:00 - 08:45 第一节 08:55 - 09:40 第二节 09:55 - 10:40 第三节 10:50 - 11:35 第四节 11:45 - 12:30 第五节 下午: 14:00 - 14:45 第六节 14:55 - 15:40 第七节 15:55 - 16:40 第八节 16:50 - 17:35 第九节 17:45 - 18:30 第十节 晚上: 19:00 - 19:45 第十一节 19:55 - 20:40 第十二节 周数安排如下: 从指定日期(新学期将从2025-08-25开始,包括08-25当天),每七天为算作一周 reader中定义的CourseSchedule类中包括字段week: TimeRange,若该字段的from=1,to=3,即对应第1,2,3周均存在该课程 通知逻辑按照如下规则: 假设对应第x节存在课程的话: 如果为第一节存在课程: 在07:00发送通知 若为第六节: 在13:00发送通知 若为第十一节: 在18:30发送通知 其他节课程提前十分钟发送通知即可 通知的发送时间严格按照上述定义,不得提前发送 通知将在合适时间发送,可使用时间轮算法完成周、日的课程安排,但需要先对reader返回的课程时间进行解析。该脚本将持续运行并检测安排,在合适的时间发送通知 """ import time import datetime from typing import List import reader import informer # 课程时间安排 CLASS_SCHEDULE = { "上午": { 1: "08:00", 2: "08:55", 3: "09:55", 4: "10:50", 5: "11:45" }, "下午": { 6: "14:00", 7: "14:55", 8: "15:55", 9: "16:50", 10: "17:45" }, "晚上": { 11: "19:00", 12: "19:55" } } # 通知发送时间 NOTIFICATION_TIMES = { 1: "07:00", # 第一节课在07:00发送通知 6: "13:00", # 第六节课在13:00发送通知 11: "18:30" # 第十一节课在18:30发送通知 } WEEKDAYS = { "Monday": "星期一", "Tuesday": "星期二", "Wednesday": "星期三", "Thursday": "星期四", "Friday": "星期五", "Saturday": "星期六", "Sunday": "星期日" } def _get_current_week(start_date: datetime.date = datetime.date(2025, 8, 25)) -> int: """ 计算当前是第几周 Args: start_date: 学期开始日期,默认为2025-08-25 Returns: int: 当前周数 """ today = datetime.date.today() delta = today - start_date week = delta.days // 7 + 1 return max(week, 1) # 确保至少为第1周 def _get_notification_time(period: str, timeslot_from: int) -> str: """ 获取课程通知发送时间 Args: period: 课程时段(上午/下午/晚上) timeslot_from: 课程开始节次 Returns: str: 通知发送时间(HH:MM格式) """ # 特殊节次的通知时间 if timeslot_from in NOTIFICATION_TIMES: return NOTIFICATION_TIMES[timeslot_from] # 其他节次提前十分钟发送通知 class_time = CLASS_SCHEDULE.get(period, {}).get(timeslot_from, "08:00") hour, minute = map(int, class_time.split(":")) # 提前10分钟 total_minutes = hour * 60 + minute - 10 if total_minutes < 0: total_minutes += 24 * 60 # 处理跨天情况 notify_hour = total_minutes // 60 notify_minute = total_minutes % 60 return f"{notify_hour:02d}:{notify_minute:02d}" def _calculate_time_interval(class_time: str) -> str: """ 计算距离上课的时间间隔 Args: class_time: 课程开始时间(HH:MM格式) Returns: str: 时间间隔描述 """ now = datetime.datetime.now() class_hour, class_minute = map(int, class_time.split(":")) class_datetime = now.replace(hour=class_hour, minute=class_minute, second=0, microsecond=0) if class_datetime < now: class_datetime += datetime.timedelta(days=1) # 如果是今天已过的时间,计算明天的 delta = class_datetime - now hours = delta.seconds // 3600 minutes = (delta.seconds % 3600) // 60 if hours > 0: return f"{hours}小时{minutes}分钟" else: return f"{minutes}分钟" def _should_send_notification(course, current_week: int, current_day: str) -> bool: """ 判断是否应该发送课程通知 Args: course: CourseSchedule对象 current_week: 当前周数 current_day: 当前星期几 Returns: bool: 是否应该发送通知 """ # 检查当前周数是否在课程周数范围内 if not (course.week.from_value <= current_week <= course.week.to_value): return False # 检查当前星期是否匹配 if course.day != current_day: return False return True def _send_course_notification(course, period: str) -> None: """ 发送课程通知 Args: course: CourseSchedule对象 period: 课程时段 """ # 获取课程开始时间 class_time = CLASS_SCHEDULE.get(period, {}).get(course.timeslot.from_value, "08:00") # 计算距离上课的时间 time_interval = _calculate_time_interval(class_time) # 创建通知数据 info_data = informer.InfoData( date=f"{datetime.date.today()} {course.day}", time=datetime.datetime.now().strftime("%H:%M:%S"), course_time=class_time, course_name=course.course_name, course_teacher=course.instructor, course_location=course.location, course_time_interval=time_interval ) # 发送通知 informer.publish_info(info_data) def main(): """主函数:读取课程信息并根据时间安排发送通知""" print("课程通知系统启动...") try: # 读取课程信息 student_info, courses = reader.read_course_schedule() print(f"已读取 {len(courses)} 门课程信息") # 持续运行并检测课程安排 while True: # 获取当前周数和星期几 current_week = _get_current_week() current_day = datetime.date.today().strftime("%A") # 转换为中文星期几 current_day = WEEKDAYS.get(current_day, current_day) # 获取当前时间 now = datetime.datetime.now() current_time = now.strftime("%H:%M") # 检查今天的课程 for course in courses: # 判断是否应该发送通知 if _should_send_notification(course, current_week, current_day): # 获取通知发送时间 notify_time = _get_notification_time(course.period, course.timeslot.from_value) # 如果当前时间匹配通知时间,则发送通知 if current_time == notify_time: print(f"发送课程通知: {course.course_name}") _send_course_notification(course, course.period) time.sleep(60*40) # 每30s检查一次 time.sleep(30) except KeyboardInterrupt: print("\n课程通知系统已停止") except Exception as e: print(f"课程通知系统出错: {e}") def _test_day_notifications(courses, current_week: int, current_day: str, day_name: str): """测试指定日期的通知逻辑(私有方法)""" print(f"\n=== 测试{day_name} ===") # 测试不同时段的通知,从06:55开始到20:00结束,每分钟递增 test_times = [] start_hour, start_minute = 6, 55 end_hour, end_minute = 20, 0 current_hour, current_minute = start_hour, start_minute while current_hour < end_hour or (current_hour == end_hour and current_minute <= end_minute): test_times.append(f"{current_hour:02d}:{current_minute:02d}") # 递增分钟 current_minute += 1 if current_minute >= 60: current_minute = 0 current_hour += 1 for test_time in test_times: # 检查课程 for course in courses: # 判断是否应该发送通知 if _should_send_notification(course, current_week, current_day): # 获取通知发送时间 notify_time = _get_notification_time(course.period, course.timeslot.from_value) # 如果当前时间匹配通知时间,则发送通知 if test_time == notify_time: print(f"\n达到模拟时间: {test_time}") print(f" [通知] {course.course_name} - {course.instructor} - {course.location}") # 不实际发送通知,只模拟 # _send_course_notification(course, course.period) def test_notification_logic(): """测试方法:模拟第一周第一天、第二天、第三天的课程安排通知逻辑""" print("开始测试通知逻辑...") try: # 读取课程信息 student_info, courses = reader.read_course_schedule() print(f"已读取 {len(courses)} 门课程信息") # 模拟第一周 current_week = 1 print(f"模拟第 {current_week} 周") # 测试第一天(星期一) _test_day_notifications(courses, current_week, "星期一", "第一天(星期一)") # 测试第二天(星期二) _test_day_notifications(courses, current_week, "星期二", "第二天(星期二)") # 测试第三天(星期三) _test_day_notifications(courses, current_week, "星期三", "第三天(星期三)") print("\n测试完成") except Exception as e: print(f"测试过程中出错: {e}") if __name__ == "__main__": # 运行测试方法 test_notification_logic() # 如果要运行主方法,取消下面的注释 # main()