commit 074a2ee56c879cf419ae04f295664cb5fa2d812e Author: slhafzjw Date: Wed Aug 20 10:55:11 2025 +0800 feat(CourseInformer): 添加课程通知系统 - 实现了读取教务系统导出的xls课程安排以及通过Napcat实例发送对应通知的功能 - 添加了通知配置文件和测试脚本 - 创建了项目结构和必要的配置文件 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ebc5ca9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +/resources/教学安排表.xls +/CLAUDE.md +/.idea/misc.xml diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..7ec258c --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,9 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml +/.name diff --git a/.idea/CourseInformer.iml b/.idea/CourseInformer.iml new file mode 100644 index 0000000..ad5b163 --- /dev/null +++ b/.idea/CourseInformer.iml @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..205225c --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..94a25f7 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/informer.py b/informer.py new file mode 100644 index 0000000..6d001e3 --- /dev/null +++ b/informer.py @@ -0,0 +1,161 @@ +""" +通知类定义: +``` +InfoData: + date: 当前日期（包括星期几） + time: 当前时间 + course_time: 课程开始时间 + course_name: 课程名称 + course_teacher: 教师名称 + course_location: 上课地点 + course_time_interval: 距上课剩余时间，根据上文计算 +``` +通知格式参考InfoData内容 + +通知形式：参考下方bash脚本，将ip换为192.168.12.1,uid、路径均不变。抽取为方法 +``` +#!/bin/bash +msg=$1 +json=$(jq -n --arg msg "$msg" --argjson uid 2998813882 '{user_id: $uid, message: $msg}') +curl -s -o /dev/null -X POST http://127.0.0.1:3000/send_private_msg \ + -H "Content-Type: application/json" \ + -d "$json" +``` + +定义方法publish_info负责推送，该类只暴露publish_info方法，如果涉及其他方法以及字段，均需要以`_`开头进行私有化 +""" + +from dataclasses import dataclass +from datetime import datetime +import json +import requests +import os + + +@dataclass +class InfoData: + """Data class for notification information""" + date: str # 当前日期（包括星期几） + time: str # 当前时间 + course_time: str # 课程开始时间 + course_name: str # 课程名称 + course_teacher: str # 教师名称 + course_location: str # 上课地点 + course_time_interval: str # 距上课剩余时间，根据上文计算 + + +def publish_info(info_data: InfoData) -> bool: + """ + Publish course information notification + + Args: + info_data: InfoData object containing notification information + + Returns: + bool: True if notification was sent successfully, False otherwise + """ + message = f"""==========课程提醒========== + +课程: {info_data.course_name} +地点: {info_data.course_location} +教师: {info_data.course_teacher} +课程开始时间: {info_data.course_time} +距离上课: {info_data.course_time_interval} + +当前日期: {info_data.date} +当前时间: {info_data.time}""" + try: + # Format the message based on InfoData + + # Send the notification using curl + _send_notification(message) + return True + except Exception as e: + print(f"Failed to publish info: {e}") + return False + + +def _load_config(): + """ + Load notification configuration from JSON file (private method) + + Returns: + dict: Configuration dictionary + """ + config_path = os.path.join(os.path.dirname(__file__), 'notification_config.json') + with open(config_path, 'r', encoding='utf-8') as f: + return json.load(f) + + +def _send_notification(message: str) -> None: + """ + Send notification via HTTP POST request (private method) + + Args: + message: The message to send + """ + # Load configuration + config = _load_config() + + # Extract server and user configuration + host = config['server']['host'] + port = config['server']['port'] + endpoint = config['server']['endpoint'] + uid = config['user']['uid'] + + # Create URL + url = f"http://{host}:{port}{endpoint}" + + # Create JSON payload + payload = { + "user_id": uid, + "message": message + } + + # Send POST request + headers = {"Content-Type": "application/json"} + response = requests.post(url, json=payload, headers=headers) + + # Raise an exception if the request was unsuccessful + response.raise_for_status() + + +def _get_test_info() -> InfoData: + """ + Get current course information (private method) + + Returns: + InfoData: Current course information + """ + # Get current date and time + now = datetime.now() + date_str = now.strftime("%Y-%m-%d") + time_str = now.strftime("%H:%M:%S") + + # For demonstration, we'll use sample data + # In a real implementation, this would fetch from reader.py + info_data = InfoData( + date=date_str+" 星期一", + time=time_str, + course_time="08:00", + course_name="示例课程", + course_teacher="示例教师", + course_location="示例地点", + course_time_interval="30分钟" + ) + + return info_data + + +# Example usage (for testing purposes) +if __name__ == "__main__": + # Get current info + info = _get_test_info() + + # Publish the info + success = publish_info(info) + + if success: + print("Notification sent successfully") + else: + print("Failed to send notification") \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..31c25ae --- /dev/null +++ b/main.py @@ -0,0 +1,326 @@ +""" +该类负责整合`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() \ No newline at end of file diff --git a/notification_config.json b/notification_config.json new file mode 100644 index 0000000..b6c8f26 --- /dev/null +++ b/notification_config.json @@ -0,0 +1,10 @@ +{ + "server": { + "host": "192.168.12.1", + "port": 3000, + "endpoint": "/send_private_msg" + }, + "user": { + "uid": 2998813882 + } +} \ No newline at end of file diff --git a/reader.py b/reader.py new file mode 100644 index 0000000..8c9f75d --- /dev/null +++ b/reader.py @@ -0,0 +1,247 @@ +""" +Module for reading course schedule information from HTML file +""" + +from dataclasses import dataclass +from typing import List +import os +import re +from bs4 import BeautifulSoup + + +@dataclass +class TimeRange: + """Data class representing a time range with from and to values""" + from_value: int + to_value: int + + +@dataclass +class CourseSchedule: + """Data class representing a course schedule entry""" + course_name: str + instructor: str + week: TimeRange # e.g., "1-8", "11-18", etc. + timeslot: TimeRange # e.g., "[1-2]", "[6-7]", "[8-10]", etc. + location: str # e.g., "[龙]二号楼2301" + day: str # e.g., "星期一", "星期二", etc. + period: str # e.g., "上午", "下午", "晚上" + + +@dataclass +class StudentInfo: + """Data class representing student information""" + student_id: str + student_name: str + class_name: str + total_credits: float + + +def read_course_schedule(file_path: str = "resources/教学安排表.xls") -> tuple[StudentInfo, List[CourseSchedule]]: + """ + Read _course schedule information from HTML file + + Args: + file_path: Path to the HTML file containing _course schedule + + Returns: + Tuple of (StudentInfo, List of CourseSchedule objects) + """ + if not os.path.exists(file_path): + raise FileNotFoundError(f"Course schedule file not found: {file_path}") + + # Read the HTML file with GBK encoding + with open(file_path, 'r', encoding='GBK') as file: + content = file.read() + + # Parse HTML content + soup = BeautifulSoup(content, 'html.parser') + + # Extract student information + student_id = "Unknown" + student_name = "Unknown" + class_name = "Unknown" + total_credits = 0.0 + + # Get student info from hidden inputs + xh_input = soup.find('input', {'id': 'xh'}) + if xh_input and xh_input.get('value'): + student_id = xh_input['value'] + + # Get student name and class from the first table + tables = soup.find_all('table') + if tables: + first_table = tables[0] + table_rows = first_table.find_all('tr') + for row in table_rows: + cells = row.find_all('td') + for cell in cells: + text = cell.get_text().strip() + if text.startswith("姓名："): + student_name = text.replace("姓名：", "") + elif text.startswith("所在班级："): + class_name = text.replace("所在班级：", "") + + _student_info = StudentInfo( + student_id=student_id, + student_name=student_name, + class_name=class_name, + total_credits=total_credits + ) + + # Extract _course schedule information + _courses = [] + + # Get day names from the header row + days = ["星期一", "星期二", "星期三", "星期四", "星期五"] # Default days + header_row = soup.find('tr', class_='H') + if header_row: + day_cells = header_row.find_all('td', class_='td0') + days = [] + for cell in day_cells: + day_text = cell.get_text().strip() + if day_text and "星期" in day_text: # Only add actual day names + days.append(day_text) + + # Get _course information from the schedule table + course_divs = soup.find_all('div', class_='div1') + for div in course_divs: + # Extract _course information from each div + xkinfo = div.find('span', class_='xkinfo') + if xkinfo: + # Each div may contain multiple _courses + course_blocks = xkinfo.find_all('div', style=lambda x: x and 'padding-bottom:5px' in x) + for block in course_blocks: + # Get the raw text + raw_text = block.get_text() + + # Initialize variables + course_name = "Unknown Course" + instructor = "Unknown" + location = "Unknown" + week_range = TimeRange(1, 1) + timeslot_range = TimeRange(1, 1) + + # Parse the concatenated text based on the format: + # course_name + instructor + weeks[timeslot] + location + # e.g., "企业资源计划（ERP）黄伟 1-8[3-4][龙]一号楼1307" + + # Look for the pattern: numbers followed by brackets (timeslot) + time_pattern = r'([0-9\-]+)(\[[0-9\-]+\])' + time_match = re.search(time_pattern, raw_text) + + if time_match: + # Extract time information + weeks_str = time_match.group(1) + timeslot_str = time_match.group(2) + + # Parse week range + week_range = TimeRange(1, 1) + week_range = time_parser(weeks_str,week_range) + + # Parse timeslot range (remove brackets) + timeslot_range = TimeRange(1, 1) + timeslot_clean = timeslot_str.strip('[]') + timeslot_range = time_parser(timeslot_clean, timeslot_range) + + # Extract location (everything after the time info) + time_end = time_match.end() + if time_end < len(raw_text): + location = raw_text[time_end:].strip() + + # Extract the part before time info + time_start = time_match.start() + before_time = raw_text[:time_start].strip() + + # Split the part before time to get _course name and instructor + # Look for Chinese characters to identify the instructor + instructor_pattern = r'([\u4e00-\u9fff]+)$' + instructor_match = re.search(instructor_pattern, before_time) + if instructor_match: + instructor = instructor_match.group(1) + # Course name is everything before the instructor + instructor_start = instructor_match.start() + course_name = before_time[:instructor_start].strip() + else: + # If we can't find instructor, use the whole part as _course name + course_name = before_time + else: + # Fallback: try to extract at least the _course name + # Assume the first part is the _course name + parts = raw_text.split() + if parts: + course_name = parts[0] + + # Determine day and period based on div id + day = "Unknown" + period = "Unknown" + div_id = div.get('id', '') + if div_id.startswith('k') and len(div_id) >= 3: + # Extract day from div id (k11, k21, etc.) + # First digit after 'k' represents the day (1=Monday, 2=Tuesday, etc.) + try: + day_index = int(div_id[1]) - 1 + if 0 <= day_index < len(days): + day = days[day_index] + except (ValueError, IndexError): + pass + + # Extract period from div id (last digit represents the period) + # 1,2 = 上午, 3,4 = 下午, 5 = 晚上 + try: + period_index = int(div_id[2]) + if period_index in [1, 2]: + period = "上午" + elif period_index in [3, 4]: + period = "下午" + elif period_index == 5: + period = "晚上" + except (ValueError, IndexError): + pass + + _course = CourseSchedule( + course_name=course_name, + instructor=instructor, + week=week_range, + timeslot=timeslot_range, + location=location, + day=day, + period=period + ) + _courses.append(_course) + + return _student_info, _courses + + +def time_parser(timeslot_clean, timeslot_range): + if '-' in timeslot_clean: + timeslot_parts = timeslot_clean.split('-') + if len(timeslot_parts) == 2: + try: + timeslot_range = TimeRange(int(timeslot_parts[0]), int(timeslot_parts[1])) + except ValueError: + pass + else: + try: + timeslot_value = int(timeslot_clean) + timeslot_range = TimeRange(timeslot_value, timeslot_value) + except ValueError: + pass + return timeslot_range + + +if __name__ == "__main__": + # Test the function + try: + student_info, courses = read_course_schedule() + print(f"Student: {student_info.student_name} ({student_info.student_id})") + print(f"Class: {student_info.class_name}") + print("\nCourses:") + for course in courses: # Print first 5 courses + print(f"- {course.course_name} by {course.instructor}") + print(f" Time: {course.week.from_value}-{course.week.to_value} [{course.timeslot.from_value}-{course.timeslot.to_value}]") + print(f" Location: {course.location}") + print(f" Day: {course.day}, Period: {course.period}") + print() + except Exception as e: + print(f"Error reading course schedule: {e}") \ No newline at end of file