biss/ZFScoresCheck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimized code

Browse Source
This commit is contained in:
NianBroken
2024-07-26 23:02:23 +08:00 Unverified
parent 91e353c901
commit b2ee1c9ea8
4 changed files with 109 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/ad.py
+26 -27
View File
@@ -1,51 +1,50 @@
import os import os
import requests import requests
# 定义常量URL # 定义常量URL，广告内容的URL
AD_URL = "https://raw.githubusercontent.com/NianBroken/ZFCheckScores/main/data/advertise.txt" AD_URL = "https://raw.githubusercontent.com/NianBroken/ZFCheckScores/main/data/advertise.txt"
# 定义常量URL，免广告用户列表的URL
NO_AD_USERS_LIST_URL = "https://raw.githubusercontent.com/NianBroken/ZFCheckScores/main/data/no_ad_users_list.txt" NO_AD_USERS_LIST_URL = "https://raw.githubusercontent.com/NianBroken/ZFCheckScores/main/data/no_ad_users_list.txt"
# 定义本地数据存储路径 # 定义本地数据存储的文件夹路径
FOLDER_PATH = "data" FOLDER_PATH = "data"
# 定义本地信息文件的路径
INFO_FILE_PATH = os.path.join(FOLDER_PATH, "info.txt") INFO_FILE_PATH = os.path.join(FOLDER_PATH, "info.txt")
# 从文件中读取当前用户 # 从文件中读取当前用户
try: try:
with open(INFO_FILE_PATH, "r") as info_file: with open(INFO_FILE_PATH, "r") as info_file: # 尝试打开本地信息文件并读取内容
CURRENT_USER = info_file.read().strip() CURRENT_USER = info_file.read().strip() # 读取文件内容并去除首尾空格，赋值给CURRENT_USER
except Exception: except Exception: # 如果读取文件发生异常
CURRENT_USER = "None" CURRENT_USER = "None" # 将CURRENT_USER设为"None"
# 发送HTTP GET请求获取指定URL的内容 # 发送HTTP GET请求获取指定URL的内容
def fetch_content(url): def fetch_content(url): # 定义函数fetch_content，参数为url
try: try:
response = requests.get(url) response = requests.get(url) # 发送HTTP GET请求
return response.text if response.status_code == 200 else None return response.text if response.status_code == 200 else None # 如果响应状态码是200，返回响应内容，否则返回None
except Exception: except Exception: # 如果请求过程中发生异常
return None return None # 返回None
# 获取免广告用户列表 # 获取免广告用户列表
def get_no_ad_users_list(): def get_no_ad_users_list(): # 定义函数get_no_ad_users_list
try: try:
return fetch_content(NO_AD_USERS_LIST_URL) or [] return fetch_content(NO_AD_USERS_LIST_URL) or [] # 尝试获取免广告用户列表内容，如果失败返回空列表
except Exception: except Exception: # 如果过程中发生异常
return [] return [] # 返回空列表
# 获取广告内容 # 获取广告内容
def get_advertise(): def get_advertise(): # 定义函数get_advertise
try: try:
# 获取免广告用户列表 no_ad_users_list = get_no_ad_users_list() # 获取免广告用户列表
no_ad_users_list = get_no_ad_users_list()
# 如果当前用户不在免广告用户列表中，获取广告内容 if CURRENT_USER not in no_ad_users_list: # 如果当前用户不在免广告用户列表中
if CURRENT_USER not in no_ad_users_list: ad_content = fetch_content(AD_URL) # 获取广告内容
ad_content = fetch_content(AD_URL) return ad_content # 返回广告内容
return ad_content else: # 当前用户在免广告用户列表中
# 当前用户在免广告用户列表中，不推送广告 return None # 不推送广告，返回None
else: except Exception: # 如果过程中发生异常
return None return None # 返回None
except Exception:
return None
scripts/delete_old_runs.py
+61 -49
View File
@@ -1,77 +1,89 @@
# 必要的依赖库 # 必要的依赖库
import requests import requests # 导入requests库，用于发送HTTP请求
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone # 导入datetime模块中的datetime, timedelta, timezone类，用于处理日期和时间
import os import os # 导入os模块，用于处理环境变量
class GitHubActionsManager: class GitHubActionsManager:
def __init__(self, repo_url, token): def __init__(self, repo_url, token, run_id, hour_count):
self.repo_url = repo_url # GitHub仓库URL self.repo_url = repo_url # GitHub仓库的URL
self.token = token # GitHub个人访问令牌 self.token = token # GitHub个人访问令牌
self.runs_url = f"{self.repo_url}/actions/runs" # 获取运行记录的URL self.run_id = run_id # 当前工作流运行记录的ID
self.current_time = datetime.now(timezone.utc) # 获取当前时间 self.hour_count = hour_count # 要删除的记录时间间隔，以小时为单位
self.runs_url = f"{self.repo_url}/actions/runs" # 用于获取工作流运行记录的URL
self.current_time = datetime.now(timezone.utc) # 获取当前UTC时间
self.deleted_count = 0 # 初始化删除计数器，记录成功删除的运行记录数量
def get_workflow_runs(self, url): def get_workflow_runs(self, url):
# 发送GET请求获取工作流运行记录 # 发送GET请求以获取工作流运行记录
response = requests.get(url, headers={"Authorization": f"token {self.token}"}) response = requests.get(url, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证
if response.status_code == 200: # 如果请求成功 if response.status_code == 200: # 如果响应状态码为200（成功）
return response.json()["workflow_runs"] # 返回运行记录列表 return response.json()["workflow_runs"] # 返回响应中的工作流运行记录列表
else: # 如果请求失败 else: # 如果响应状态码不是200（失败）
self.log(f"Failed to fetch runs from {url}. Status code: {response.status_code}") # 打印错误信息 self.log(f"Failed to fetch runs from {url}. Status code: {response.status_code}") # 打印失败日志
return [] # 返回空列表 return [] # 返回空列表
def delete_run(self, run_id): def delete_run(self, run_id):
delete_url = f"{self.runs_url}/{run_id}" # 构建删除运行记录的URL delete_url = f"{self.runs_url}/{run_id}" # 构建删除指定运行记录的URL
# 发送DELETE请求删除指定ID的运行记录 # 发送DELETE请求以删除指定ID的工作流运行记录
response = requests.delete(delete_url, headers={"Authorization": f"token {self.token}"}) response = requests.delete(delete_url, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证
if response.status_code == 204: # 如果删除成功 if response.status_code == 204: # 如果响应状态码为204（成功删除）
self.log(f"Deleted run with ID {run_id}") # 打印删除成功的消息 self.log(f"Deleted run with ID {run_id}") # 打印删除成功日志
else: # 如果删除失败 self.deleted_count += 1 # 成功删除计数器加1
self.log(f"Failed to delete run with ID {run_id}. Status code: {response.status_code}") # 打印错误信息 else: # 如果响应状态码不是204（删除失败）
self.log(f"Failed to delete run with ID {run_id}. Status code: {response.status_code}") # 打印删除失败日志
def delete_old_runs(self): def delete_old_runs(self):
while True: # 无限循环，直到没有符合条件的运行记录为止 while True: # 无限循环，直到没有符合条件的运行记录
next_page = self.runs_url # 初始化下一页的URL为第一页 next_page = self.runs_url # 初始化下一页的URL为第一页
page_number = 1 # 初始化页数计数器
any_deleted = False # 标记是否有记录被删除 any_deleted = False # 标记是否有记录被删除
while next_page: # 循环直到没有下一页 while next_page: # 循环直到没有下一页
self.log(f"Fetching page {page_number}") # 输出当前的页数 # 发送GET请求以获取一页工作流运行记录
# 发送GET请求获取一页工作流运行记录 response = requests.get(next_page, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证
response = requests.get(next_page, headers={"Authorization": f"token {self.token}"}) if response.status_code == 200: # 如果响应状态码为200（成功）
if response.status_code == 200: # 如果请求成功 data = response.json() # 将响应数据解析为JSON格式
data = response.json() # 将响应数据转换为JSON格式 runs = data["workflow_runs"] # 获取工作流运行记录列表
runs = data["workflow_runs"] # 获取运行记录列表 next_page = response.links.get("next", {}).get("url") # 获取下一页的URL，如果没有下一页则为None
next_page = response.links.get("next", {}).get("url") # 获取下一页的URL
page_number += 1 # 增加页数计数器
for run in runs: # 遍历每条运行记录 for run in runs: # 遍历每一条运行记录
# 将运行记录的创建时间转换为UTC时间 if self.hour_count == 0: # 如果设置的时间间隔为0小时
run_time = datetime.strptime(run["created_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) if run["id"] != self.run_id: # 如果运行记录ID不等于当前运行记录ID
# 计算当前时间与运行记录创建时间的差值 self.delete_run(run["id"]) # 删除该运行记录
time_difference = self.current_time - run_time
if time_difference > timedelta(hours=hour_count): # 如果差值超过指定小时 默认168小时
self.delete_run(run["id"]) # 删除运行记录
any_deleted = True # 标记有记录被删除 any_deleted = True # 标记有记录被删除
else: # 如果请求失败 else: # 如果设置的时间间隔大于0小时
self.log(f"Failed to fetch runs. Status code: {response.status_code}") # 打印错误信息 # 将运行记录的创建时间字符串转换为datetime对象，并设置为UTC时间
run_time = datetime.strptime(run["created_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
# 计算当前时间与运行记录创建时间的时间差
time_difference = self.current_time - run_time
if time_difference > timedelta(hours=self.hour_count): # 如果时间差大于设定的小时数
self.delete_run(run["id"]) # 删除该运行记录
any_deleted = True # 标记有记录被删除
else: # 如果响应状态码不是200（失败）
self.log(f"Failed to fetch runs. Status code: {response.status_code}") # 打印获取运行记录失败日志
break # 退出循环 break # 退出循环
if not any_deleted: # 如果没有任何记录被删除，退出循环 if not any_deleted: # 如果没有任何记录被删除
break break # 退出循环
self.log(f"Total number of deleted runs: {self.deleted_count}") # 输出总共成功删除的运行记录数量
def log(self, message): def log(self, message):
beijing_time = datetime.now(timezone.utc).astimezone(timezone(timedelta(hours=8))) # 获取北京时间 # 获取当前时间的北京时间（UTC+8）
current_time_str = beijing_time.strftime("%Y-%m-%d %H:%M:%S:%f")[:-3] # 获取当前时间，精确到三位毫秒 beijing_time = datetime.now(timezone.utc).astimezone(timezone(timedelta(hours=8)))
# 将当前时间格式化为字符串，精确到三位毫秒
current_time_str = beijing_time.strftime("%Y-%m-%d %H:%M:%S:%f")[:-3]
print(f"{current_time_str} {message}") # 打印日志消息 print(f"{current_time_str} {message}") # 打印日志消息
if __name__ == "__main__": if __name__ == "__main__":
repository_name = os.environ.get("REPOSITORY_NAME") repository_name = os.environ.get("REPOSITORY_NAME") # 从环境变量获取GitHub仓库名称
github_token = os.environ.get("GITHUB_TOKEN") github_token = os.environ.get("GITHUB_TOKEN") # 从环境变量获取GitHub个人访问令牌
hour_count = int(os.environ.get("HOUR_COUNT")) github_run_id = int(os.environ.get("GITHUB_RUN_ID")) # 从环境变量获取当前工作流运行记录的ID，并转换为整数
repo_url = f"https://api.github.com/repos/{repository_name}" # GitHub仓库URL hour_count = int(os.environ.get("HOUR_COUNT")) # 从环境变量获取要删除的记录时间间隔（小时），并转换为整数
repo_url = f"https://api.github.com/repos/{repository_name}" # 构建GitHub仓库的API URL
token = f"{github_token}" # GitHub个人访问令牌 token = f"{github_token}" # GitHub个人访问令牌
manager = GitHubActionsManager(repo_url, token) # 创建GitHubActionsManager实例 # 创建GitHubActionsManager实例
manager.delete_old_runs() # 删除指定小时前的运行记录 manager = GitHubActionsManager(repo_url, token, github_run_id, hour_count)
manager.delete_old_runs() # 删除指定时间间隔（小时）之前的运行记录
scripts/push.py
+20 -7
View File
@@ -4,27 +4,40 @@ import re
from scripts.ad import get_advertise from scripts.ad import get_advertise
# 定义函数send_message，用于发送消息
def send_message(token, title, content): def send_message(token, title, content):
# 推送消息的url，格式化字符串以包含token
url = f"https://push.showdoc.com.cn/server/api/push/{token}" url = f"https://push.showdoc.com.cn/server/api/push/{token}"
# 调用get_advertise函数获取广告内容
advertise = get_advertise() advertise = get_advertise()
# 如果广告内容不为空，则将广告内容添加到消息内容前
content = f"{advertise}{content}" if advertise else content content = f"{advertise}{content}" if advertise else content
# 定义正则表达式模式，用于匹配包含“教学班ID”的行
pattern = re.compile(r"^.*教学班ID.*$\n?", re.MULTILINE) pattern = re.compile(r"^.*教学班ID.*$\n?", re.MULTILINE)
# 使用正则表达式替换匹配的行为空字符串，并去除前后的空白字符
content = re.sub(pattern, "", content).strip() content = re.sub(pattern, "", content).strip()
# 定义一个字典，用于存储需要替换的字符串及其对应的替换值
replacements = { replacements = {
"------": "\n------\n", "------": "\n------\n", # 替换多连字符为换行后的连字符
"个人信息：": "<h1>个人信息</h1>\n", "个人信息：": "<h1>个人信息</h1>\n", # 替换“个人信息：”为HTML标题标签
"成绩信息：": "<h1>成绩信息</h1>\n", "成绩信息：": "<h1>成绩信息</h1>\n", # 替换“成绩信息：”为HTML标题标签
"未公布成绩的课程：": "<h1>未公布成绩的课程</h1>\n", "未公布成绩的课程：": "<h1>未公布成绩的课程</h1>\n", # 替换“未公布成绩的课程：”为HTML标题标签
"工作流信息：": "<h1>工作流信息</h1>\n", "工作流信息：": "<h1>工作流信息</h1>\n", # 替换“工作流信息：”为HTML标题标签
"Copyright © 2024 NianBroken. All rights reserved.": "Copyright © 2024 <a href='https://www.nianbroken.top/' target='_blank'>NianBroken</a>. All rights reserved.", "Copyright © 2024 NianBroken. All rights reserved.": "Copyright © 2024 <a href='https://www.nianbroken.top/' target='_blank'>NianBroken</a>. All rights reserved.", # 替换版权信息为带有超链接的HTML
} }
# 遍历字典中的所有键值对，进行替换操作
for old, new in replacements.items(): for old, new in replacements.items():
content = content.replace(old, new) content = content.replace(old, new)
# 创建一个字典，包含标题和内容
data = {"title": title, "content": content} data = {"title": title, "content": content}
# 将字典转换为JSON字符串，并编码为UTF-8
body = json.dumps(data).encode(encoding="utf-8") body = json.dumps(data).encode(encoding="utf-8")
# 定义HTTP请求头，设置内容类型为JSON
headers = {"Content-Type": "application/json"} headers = {"Content-Type": "application/json"}
# 发送POST请求，携带JSON数据和请求头
response = requests.post(url, data=body, headers=headers) response = requests.post(url, data=body, headers=headers)
# 解析 JSON 数据 # 解析响应的JSON数据，转换为字典
response_dict = json.loads(response.text) response_dict = json.loads(response.text)
# 返回响应字典
return response_dict return response_dict
scripts/user_login.py
+1 -1
View File
@@ -37,6 +37,6 @@ def login(url, username, password):
""" """
elif lgn["code"] != 1000: elif lgn["code"] != 1000:
pprint(lgn) pprint(lgn)
sys.exit() sys.exit("登录失败")
return student_client return student_client