326 lines
10 KiB
Python
326 lines
10 KiB
Python
"""
|
||
该类负责整合`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() |