Code refactoring and optimization: remove duplicate code, simplify grade processing logic, and improve code readability and maintainability.

This commit is contained in:
NianBroken
2024-05-06 19:04:10 +08:00 Unverified
parent c9f5433005
commit 91344e99ab
7 changed files with 254 additions and 210 deletions
+1
View File
@@ -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 }}
+60
View File
@@ -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 "获取成绩:参数缺失"
+30
View File
@@ -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 "获取个人信息:参数缺失"
+79
View File
@@ -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
+34 -206
View File
@@ -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,59 +90,16 @@ 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
)
# 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"------"
integrated_grade_info = get_grade(
student_client, output_type="integrated_grade_info"
)
else:
# 成绩为空时将成绩信息定义为"成绩为空"
@@ -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,6 +224,7 @@ if run_log:
# 将任意个数的换行替换为两个换行
github_step_summary_run_log = re.sub("\n+", "\n\n", 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)
+6 -1
View File
@@ -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
+41
View File
@@ -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