From 91344e99ab4ef124c5d9d1cdb35f4d312bd8a4d8 Mon Sep 17 00:00:00 2001 From: NianBroken Date: Mon, 6 May 2024 19:04:10 +0800 Subject: [PATCH] Code refactoring and optimization: remove duplicate code, simplify grade processing logic, and improve code readability and maintainability. --- .github/workflows/main.yml | 1 + get_grade.py | 60 +++++++++ get_info.py | 30 +++++ get_selected_courses.py | 79 ++++++++++++ main.py | 246 ++++++------------------------------- pushplus.py | 7 +- user_login.py | 41 +++++++ 7 files changed, 254 insertions(+), 210 deletions(-) create mode 100644 get_grade.py create mode 100644 get_info.py create mode 100644 get_selected_courses.py create mode 100644 user_login.py diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index d2f782d..616d61a 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -57,6 +57,7 @@ jobs: id: run_main_program env: FORCE_PUSH_MESSAGE: ${{ github.event.inputs.force_push_message }} + GITHUB_ACTIONS: ${{github.actions}} URL: ${{ secrets.URL }} USERNAME: ${{ secrets.USERNAME }} PASSWORD: ${{ secrets.PASSWORD }} diff --git a/get_grade.py b/get_grade.py new file mode 100644 index 0000000..a69adab --- /dev/null +++ b/get_grade.py @@ -0,0 +1,60 @@ +def get_grade(student_client, output_type="none"): + # 获取成绩信息 + grade_data = student_client.get_grade("").get("data", {}) + grade = grade_data.get("courses", []) + + # 遍历 grade 中的每个字典，将 title 中的中文括号替换为英文括号 + for course_data_grade in grade: + course_data_grade["title"] = ( + course_data_grade["title"].replace("（", "(").replace("）", ")") + ) + + # 按照提交时间降序排序 + sorted_grade = sorted(grade, key=lambda x: x["submission_time"], reverse=True) + + # 学分总和 + total_credit = sum(float(course["credit"]) for course in grade) + + # 学分绩点总和 + total_xfjd = sum(float(course["xfjd"]) for course in grade) + + # (百分制成绩*学分)的总和 + sum_of_percentage_grades_multiplied_by_credits = sum( + float(course["percentage_grades"]) * float(course["credit"]) for course in grade + ) + + # GPA计算 (学分*绩点)的总和/学分总和 + gpa = "{:.2f}".format(total_xfjd / total_credit) + + # 百分制GPA计算 (百分制成绩*学分)的总和/学分总和 + percentage_gpa = "{:.2f}".format( + sum_of_percentage_grades_multiplied_by_credits / total_credit + ) + + # 初始化输出成绩信息字符串 + integrated_grade_info = "成绩信息：" + + # 遍历前8条成绩信息 + for _, course in enumerate(sorted_grade[:8]): + # 整合成绩信息 + integrated_grade_info += ( + f"\n" + f"教学班ID：{course['class_id']}\n" + f"课程名称：{course['title']}\n" + f"任课教师：{course['teacher']}\n" + f"成绩：{course['grade']}\n" + f"提交时间：{course['submission_time']}\n" + f"提交人姓名：{course['name_of_submitter']}\n" + f"------" + ) + + if output_type == "grade": + return grade + elif output_type == "gpa": + return gpa + elif output_type == "percentage_gpa": + return percentage_gpa + elif output_type == "integrated_grade_info": + return integrated_grade_info + else: + return "获取成绩：参数缺失" diff --git a/get_info.py b/get_info.py new file mode 100644 index 0000000..00c462d --- /dev/null +++ b/get_info.py @@ -0,0 +1,30 @@ +from get_grade import get_grade + + +def get_info(student_client, output_type="none"): + # 获取个人信息 + info = student_client.get_info()["data"] + + # 整合个人信息 + info = ( + f"个人信息：\n" + f"学号：{info['sid']}\n" + f"班级：{info['class_name']}\n" + f"姓名：{info['name']}" + ) + + grade = get_grade(student_client, output_type="grade") + gpa = get_grade(student_client, output_type="gpa") + percentage_gpa = get_grade(student_client, output_type="percentage_gpa") + + if grade: + # 整合个人信息 + gpa_info = f"\n当前GPA：{gpa}\n" f"当前百分制GPA：{percentage_gpa}\n" f"------" + integrated_info = f"{info}{gpa_info}" + + if output_type == "info": + return info + elif output_type == "integrated_info": + return integrated_info + else: + return "获取个人信息：参数缺失" diff --git a/get_selected_courses.py b/get_selected_courses.py new file mode 100644 index 0000000..dee6897 --- /dev/null +++ b/get_selected_courses.py @@ -0,0 +1,79 @@ +from get_grade import get_grade + + +def get_selected_courses(student_client): + + # 获取成绩信息 + grade = get_grade(student_client, output_type="grade") + + # 获取已选课程信息 + selected_courses_data = student_client.get_selected_courses().get("data", {}) + selected_courses = selected_courses_data.get("courses", []) + + # 已选课程信息不为空时,处理未公布成绩的课程和异常课程 + if selected_courses: + # 初始化空字典用于存储未公布成绩的课程,按学年学期分组 + ungraded_courses_by_semester = {} + # 初始化空字典用于存储异常的课程,按学年学期分组 + abnormal_courses_by_semester = {} + + # 获取成绩列表中的class_id集合 + grade_class_ids = {course["class_id"] for course in grade} + + # 初始化输出内容 + selected_courses_filtering = "" + + # 遍历selected_courses和grade中的每个课程 + for course in selected_courses + grade: + # 获取课程的class_id和学年学期 + yearsemester_id = course["class_name"].split("(")[1].split(")")[0] + year, semester, seq = yearsemester_id.split("-") + + # 构建年学期名称,例如 "a至b学年第c学期" + yearsemester_name = f"{year}至{semester}学年第{seq}学期" + + # 判断课程是否未公布成绩或为异常课程 + if course["class_id"] not in grade_class_ids: + # 未公布成绩 + ungraded_courses_by_semester.setdefault(yearsemester_name, []).append( + f"{course['title'].replace('（', '(').replace('）', ')')} - {course['teacher']}" + ) + elif course["class_id"] not in { + course["class_id"] for course in selected_courses + }: + # 异常课程 + abnormal_courses_by_semester.setdefault(yearsemester_name, []).append( + f"{course['title'].replace('（', '(').replace('）', ')')} - {course['teacher']}" + ) + + # 构建输出内容 + if ungraded_courses_by_semester: + # 存在未公布成绩的课程 + selected_courses_filtering += "------\n未公布成绩的课程：" + for i, (semester, courses) in enumerate( + ungraded_courses_by_semester.items() + ): + if i > 0: + selected_courses_filtering += "\n------" + selected_courses_filtering += f"\n{semester}：" + for course in courses: + selected_courses_filtering += f"\n{course}" + + if abnormal_courses_by_semester: + # 存在异常的课程 + if ungraded_courses_by_semester: + # 如果存在课程,添加分隔线 + selected_courses_filtering += "\n" + selected_courses_filtering += "------\n异常的课程：" + for i, (semester, courses) in enumerate( + abnormal_courses_by_semester.items() + ): + if i > 0: + selected_courses_filtering += "\n------" + selected_courses_filtering += f"\n{semester}：" + for course in courses: + selected_courses_filtering += f"\n{course}" + else: + selected_courses_filtering = "------\n已选课程信息为空" + + return selected_courses_filtering diff --git a/main.py b/main.py index d6c3730..3f86d31 100644 --- a/main.py +++ b/main.py @@ -1,17 +1,23 @@ # 必要的依赖库 import re -import base64 import hashlib import os -import sys import shutil -import json -from pprint import pprint -from zfn_api import Client +from user_login import login +from get_info import get_info +from get_grade import get_grade +from get_selected_courses import get_selected_courses from pushplus import send_message + +# MD5加密 +def md5_encrypt(string): + return hashlib.md5(string.encode()).hexdigest() + + # 从环境变量中提取教务系统的URL、用户名、密码和TOKEN等信息 force_push_message = os.environ.get("FORCE_PUSH_MESSAGE") +github_actions = os.environ.get("GITHUB_ACTIONS") url = os.environ.get("URL") username = os.environ.get("USERNAME") password = os.environ.get("PASSWORD") @@ -41,71 +47,21 @@ run_count = 2 # 初始化运行日志 run_log = "" - -# MD5加密 -def md5_encrypt(string): - return hashlib.md5(string.encode()).hexdigest() - - -# 初始化变量 -cookies = {} -base_url = url -raspisanie = [] -ignore_type = [] -detail_category_type = [] -timeout = 5 - -# 创建教务系统客户端对象 -student_client = Client( - cookies=cookies, - base_url=base_url, - raspisanie=raspisanie, - ignore_type=ignore_type, - detail_category_type=detail_category_type, - timeout=timeout, -) - # 登录 -if not cookies: - login_result = student_client.login(username, password) - if login_result["code"] == 1001: - # 如果需要验证码,获取验证码并进行登录 - verify_data = login_result["data"] - # 将验证码图片写入文件 - with open(os.path.abspath("kaptcha.png"), "wb") as pic: - pic.write(base64.b64decode(verify_data.pop("kaptcha_pic"))) - # 输入验证码 - verify_data["kaptcha"] = input("输入验证码：") - # 使用验证码进行登录 - login_result = student_client.login_with_kaptcha(**verify_data) - - if login_result["code"] != 1000: - pprint(login_result) - sys.exit() - pprint(login_result) - - elif login_result["code"] != 1000: - pprint(login_result) - sys.exit() +student_client = login(url, username, password) # 获取个人信息 -info = student_client.get_info()["data"] +info = get_info(student_client, output_type="info") -# 整合个人信息 -integrated_info = ( - f"个人信息：\n" - f"学号：{info['sid']}\n" - f"班级：{info['class_name']}\n" - f"姓名：{info['name']}" -) +# 获取完整个人信息 +integrated_info = get_info(student_client, output_type="integrated_info") # 加密个人信息 -encrypted_info = md5_encrypt(integrated_info) - +encrypted_info = md5_encrypt(info) # 判断info.txt文件是否存在 if not os.path.exists(info_file_path): - # 如果文件不存在,创建并写入encrypted_info的内容 + # 如果文件不存在,创建并写入加密后的个人信息 with open(info_file_path, "w") as info_file: info_file.write(encrypted_info) else: @@ -134,60 +90,17 @@ for _ in range(run_count): old_grade_file.write(grade_file.read()) # 获取成绩信息 - grade_data = student_client.get_grade("").get("data", {}) - grade = grade_data.get("courses", []) + grade = get_grade(student_client, output_type="grade") - # 成绩不为空时则对成绩信息进行处理 + # 成绩不为空时 if grade: - # 遍历 grade 中的每个字典，将 title 中的中文括号替换为英文括号 - for course_data_grade in grade: - course_data_grade["title"] = ( - course_data_grade["title"].replace("（", "(").replace("）", ")") - ) - # 清空grade.txt文件内容 with open(grade_file_path, "w") as grade_file: grade_file.truncate() - # 按照提交时间降序排序 - sorted_grade = sorted(grade, key=lambda x: x["submission_time"], reverse=True) - - # 学分总和 - total_credit = sum(float(course["credit"]) for course in grade) - - # 学分绩点总和 - total_xfjd = sum(float(course["xfjd"]) for course in grade) - - # (百分制成绩*学分)的总和 - sum_of_percentage_grades_multiplied_by_credits = sum( - float(course["percentage_grades"]) * float(course["credit"]) - for course in grade + integrated_grade_info = get_grade( + student_client, output_type="integrated_grade_info" ) - - # GPA计算 (学分*绩点)的总和/学分总和 - gpa = "{:.2f}".format(total_xfjd / total_credit) - - # 百分制GPA计算 (百分制成绩*学分)的总和/学分总和 - percentage_gpa = "{:.2f}".format( - sum_of_percentage_grades_multiplied_by_credits / total_credit - ) - - # 初始化输出成绩信息字符串 - integrated_grade_info = "成绩信息：" - - # 遍历前8条成绩信息 - for i, course in enumerate(sorted_grade[:8]): - # 整合成绩信息 - integrated_grade_info += ( - f"\n" - f"教学班ID：{course['class_id']}\n" - f"课程名称：{course['title']}\n" - f"任课教师：{course['teacher']}\n" - f"成绩：{course['grade']}\n" - f"提交时间：{course['submission_time']}\n" - f"提交人姓名：{course['name_of_submitter']}\n" - f"------" - ) else: # 成绩为空时将成绩信息定义为"成绩为空" integrated_grade_info = "------\n成绩为空\n------" @@ -199,13 +112,6 @@ for _ in range(run_count): with open(grade_file_path, "w") as grade_file: grade_file.write(encrypted_integrated_grade_info) -# 成绩信息不为空时整合GPA信息 -if grade: - # 整合个人信息 - integrated_info += ( - f"\n当前GPA：{gpa}\n" f"当前百分制GPA：{percentage_gpa}\n" f"------" - ) - # 读取grade.txt和old_grade.txt文件的内容 with open(grade_file_path, "r") as grade_file, open( old_grade_file_path, "r" @@ -216,71 +122,8 @@ with open(grade_file_path, "r") as grade_file, open( # 整合MD5值 integrated_grade_info += f"\n" f"MD5：{encrypted_integrated_grade_info}" -# 获取已选课程信息 -selected_courses_data = student_client.get_selected_courses().get("data", {}) -selected_courses = selected_courses_data.get("courses", []) - -# 已选课程信息不为空时,处理未公布成绩的课程和异常课程 -if selected_courses: - # 初始化空字典用于存储未公布成绩的课程,按学年学期分组 - ungraded_courses_by_semester = {} - # 初始化空字典用于存储异常的课程,按学年学期分组 - abnormal_courses_by_semester = {} - - # 获取成绩列表中的class_id集合 - grade_class_ids = {course["class_id"] for course in grade} - - # 初始化输出内容 - selected_courses_filtering = "" - - # 遍历selected_courses和grade中的每个课程 - for course in selected_courses + grade: - # 获取课程的class_id和学年学期 - yearsemester_id = course["class_name"].split("(")[1].split(")")[0] - year, semester, seq = yearsemester_id.split("-") - - # 构建年学期名称,例如 "a至b学年第c学期" - yearsemester_name = f"{year}至{semester}学年第{seq}学期" - - # 判断课程是否未公布成绩或为异常课程 - if course["class_id"] not in grade_class_ids: - # 未公布成绩 - ungraded_courses_by_semester.setdefault(yearsemester_name, []).append( - f"{course['title'].replace('（', '(').replace('）', ')')} - {course['teacher']}" - ) - elif course["class_id"] not in { - course["class_id"] for course in selected_courses - }: - # 异常课程 - abnormal_courses_by_semester.setdefault(yearsemester_name, []).append( - f"{course['title'].replace('（', '(').replace('）', ')')} - {course['teacher']}" - ) - - # 构建输出内容 - if ungraded_courses_by_semester: - # 存在未公布成绩的课程 - selected_courses_filtering += "------\n未公布成绩的课程：" - for i, (semester, courses) in enumerate(ungraded_courses_by_semester.items()): - if i > 0: - selected_courses_filtering += "\n------" - selected_courses_filtering += f"\n{semester}：" - for course in courses: - selected_courses_filtering += f"\n{course}" - - if abnormal_courses_by_semester: - # 存在异常的课程 - if ungraded_courses_by_semester: - # 如果存在课程,添加分隔线 - selected_courses_filtering += "\n" - selected_courses_filtering += "------\n异常的课程：" - for i, (semester, courses) in enumerate(abnormal_courses_by_semester.items()): - if i > 0: - selected_courses_filtering += "\n------" - selected_courses_filtering += f"\n{semester}：" - for course in courses: - selected_courses_filtering += f"\n{course}" -else: - selected_courses_filtering = "------\n已选课程信息为空" +# 获取未公布成绩的课程和异常的课程 +selected_courses_filtering = get_selected_courses(student_client) # 工作流信息 workflow_info = ( @@ -297,6 +140,14 @@ workflow_info = ( f"Beijing Time：{beijing_time}" ) +# 第一次运行时的提示文本 +first_run_text = ( + "你的程序运行成功\n" + "从现在开始,程序将会每隔 30 分钟自动检测一次成绩是否有更新\n" + "若有更新,将通过微信推送及时通知你\n" + "------" +) + # 整合所有信息 # 注意此处integrated_send_info保存的是未加密的信息,仅用于信息推送 # 若是在 Github Actions 等平台运行,请不要使用print(integrated_send_info) @@ -307,15 +158,6 @@ integrated_send_info = ( f"{workflow_info}" ) -# 第一次运行时的提示文本 -first_run_text = ( - "你的程序运行成功\n" - "从现在开始,程序将会每隔 30 分钟自动检测一次成绩是否有更新\n" - "若有更新,将通过微信推送及时通知你\n" - "------" -) - - # 整合首次运行时需要使用到的所有信息 first_time_run_integrated_send_info = f"{first_run_text}\n" f"{integrated_send_info}" @@ -337,15 +179,8 @@ if run_count == 2: first_time_run_integrated_send_info, ) - # 解析 JSON 数据 - first_run_text_response_dict = json.loads(first_run_text_response_text) - - # 删除 "data" 字段 - if "data" in first_run_text_response_dict: - first_run_text_response_dict.pop("data") - # 输出响应内容 - run_log += f"{first_run_text_response_dict}\n" + run_log += f"{first_run_text_response_text}\n" else: # 如果非第一次运行,则输出成绩信息 if grade: @@ -366,16 +201,8 @@ else: "正方教务管理系统成绩推送", grades_updated_push_integrated_send_info, ) - - # 解析 JSON 数据 - response_dict = json.loads(response_text) - - # 删除 "data" 字段 - if "data" in response_dict: - response_dict.pop("data") - # 输出响应内容 - run_log += f"{response_dict}\n" + run_log += f"{response_text}\n" else: run_log += "成绩未更新" @@ -397,9 +224,10 @@ if run_log: # 将任意个数的换行替换为两个换行 github_step_summary_run_log = re.sub("\n+", "\n\n", github_step_summary_run_log) - # 将 github_step_summary_run_log 写入到 GitHub Actions 的环境文件中 - with open(github_step_summary, "w", encoding="utf-8") as file: - file.write(github_step_summary_run_log) + if github_actions: + # 将 github_step_summary_run_log 写入到 GitHub Actions 的环境文件中 + with open(github_step_summary, "w", encoding="utf-8") as file: + file.write(github_step_summary_run_log) # 删除 __pycache__ 缓存目录及其内容 current_directory = os.getcwd() diff --git a/pushplus.py b/pushplus.py index 2155067..782b4f5 100644 --- a/pushplus.py +++ b/pushplus.py @@ -8,5 +8,10 @@ def send_message(token, title, content): body = json.dumps(data).encode(encoding="utf-8") headers = {"Content-Type": "application/json"} response = requests.post(url, data=body, headers=headers) + # 解析 JSON 数据 + response_dict = json.loads(response.text) + # 删除 "data" 字段 + if "data" in response_dict: + response_dict.pop("data") - return response.text + return response_dict diff --git a/user_login.py b/user_login.py new file mode 100644 index 0000000..6add6c6 --- /dev/null +++ b/user_login.py @@ -0,0 +1,41 @@ +import base64 +import os +import sys +from pprint import pprint +from zfn_api import Client + + +def login(url, username, password): + cookies = {} + base_url = url + raspisanie = [] + ignore_type = [] + detail_category_type = [] + timeout = 5 + + student_client = Client( + cookies=cookies, + base_url=base_url, + raspisanie=raspisanie, + ignore_type=ignore_type, + detail_category_type=detail_category_type, + timeout=timeout, + ) + + if cookies == {}: + lgn = student_client.login(username, password) + if lgn["code"] == 1001: + verify_data = lgn["data"] + with open(os.path.abspath("kaptcha.png"), "wb") as pic: + pic.write(base64.b64decode(verify_data.pop("kaptcha_pic"))) + verify_data["kaptcha"] = input("输入验证码：") + ret = student_client.login_with_kaptcha(**verify_data) + if ret["code"] != 1000: + pprint(ret) + sys.exit() + pprint(ret) + elif lgn["code"] != 1000: + pprint(lgn) + sys.exit() + + return student_client