diff --git a/scripts/ad.py b/scripts/ad.py index 05b00b0..24acc7e 100644 --- a/scripts/ad.py +++ b/scripts/ad.py @@ -1,51 +1,50 @@ import os import requests -# 定义常量URL +# 定义常量URL,广告内容的URL 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" -# 定义本地数据存储路径 +# 定义本地数据存储的文件夹路径 FOLDER_PATH = "data" +# 定义本地信息文件的路径 INFO_FILE_PATH = os.path.join(FOLDER_PATH, "info.txt") # 从文件中读取当前用户 try: - with open(INFO_FILE_PATH, "r") as info_file: - CURRENT_USER = info_file.read().strip() -except Exception: - CURRENT_USER = "None" + with open(INFO_FILE_PATH, "r") as info_file: # 尝试打开本地信息文件并读取内容 + CURRENT_USER = info_file.read().strip() # 读取文件内容并去除首尾空格,赋值给CURRENT_USER +except Exception: # 如果读取文件发生异常 + CURRENT_USER = "None" # 将CURRENT_USER设为"None" # 发送HTTP GET请求获取指定URL的内容 -def fetch_content(url): +def fetch_content(url): # 定义函数fetch_content,参数为url try: - response = requests.get(url) - return response.text if response.status_code == 200 else None - except Exception: - return None + response = requests.get(url) # 发送HTTP GET请求 + return response.text if response.status_code == 200 else None # 如果响应状态码是200,返回响应内容,否则返回None + except Exception: # 如果请求过程中发生异常 + return None # 返回None # 获取免广告用户列表 -def get_no_ad_users_list(): +def get_no_ad_users_list(): # 定义函数get_no_ad_users_list try: - return fetch_content(NO_AD_USERS_LIST_URL) or [] - except Exception: - return [] + return fetch_content(NO_AD_USERS_LIST_URL) or [] # 尝试获取免广告用户列表内容,如果失败返回空列表 + except Exception: # 如果过程中发生异常 + return [] # 返回空列表 # 获取广告内容 -def get_advertise(): +def get_advertise(): # 定义函数get_advertise 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: - ad_content = fetch_content(AD_URL) - return ad_content - # 当前用户在免广告用户列表中,不推送广告 - else: - return None - except Exception: - return None + if CURRENT_USER not in no_ad_users_list: # 如果当前用户不在免广告用户列表中 + ad_content = fetch_content(AD_URL) # 获取广告内容 + return ad_content # 返回广告内容 + else: # 当前用户在免广告用户列表中 + return None # 不推送广告,返回None + except Exception: # 如果过程中发生异常 + return None # 返回None diff --git a/scripts/delete_old_runs.py b/scripts/delete_old_runs.py index f3365ed..fcc0355 100644 --- a/scripts/delete_old_runs.py +++ b/scripts/delete_old_runs.py @@ -1,77 +1,89 @@ # 必要的依赖库 -import requests -from datetime import datetime, timedelta, timezone -import os +import requests # 导入requests库,用于发送HTTP请求 +from datetime import datetime, timedelta, timezone # 导入datetime模块中的datetime, timedelta, timezone类,用于处理日期和时间 +import os # 导入os模块,用于处理环境变量 class GitHubActionsManager: - def __init__(self, repo_url, token): - self.repo_url = repo_url # GitHub仓库URL + def __init__(self, repo_url, token, run_id, hour_count): + self.repo_url = repo_url # GitHub仓库的URL self.token = token # GitHub个人访问令牌 - self.runs_url = f"{self.repo_url}/actions/runs" # 获取运行记录的URL - self.current_time = datetime.now(timezone.utc) # 获取当前时间 + self.run_id = run_id # 当前工作流运行记录的ID + 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): - # 发送GET请求获取工作流运行记录 - response = requests.get(url, headers={"Authorization": f"token {self.token}"}) - if response.status_code == 200: # 如果请求成功 - return response.json()["workflow_runs"] # 返回运行记录列表 - else: # 如果请求失败 - self.log(f"Failed to fetch runs from {url}. Status code: {response.status_code}") # 打印错误信息 + # 发送GET请求以获取工作流运行记录 + response = requests.get(url, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证 + if response.status_code == 200: # 如果响应状态码为200(成功) + return response.json()["workflow_runs"] # 返回响应中的工作流运行记录列表 + else: # 如果响应状态码不是200(失败) + self.log(f"Failed to fetch runs from {url}. Status code: {response.status_code}") # 打印失败日志 return [] # 返回空列表 def delete_run(self, run_id): - delete_url = f"{self.runs_url}/{run_id}" # 构建删除运行记录的URL - # 发送DELETE请求删除指定ID的运行记录 - response = requests.delete(delete_url, headers={"Authorization": f"token {self.token}"}) - if response.status_code == 204: # 如果删除成功 - self.log(f"Deleted run with ID {run_id}") # 打印删除成功的消息 - else: # 如果删除失败 - self.log(f"Failed to delete run with ID {run_id}. Status code: {response.status_code}") # 打印错误信息 + delete_url = f"{self.runs_url}/{run_id}" # 构建删除指定运行记录的URL + # 发送DELETE请求以删除指定ID的工作流运行记录 + response = requests.delete(delete_url, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证 + if response.status_code == 204: # 如果响应状态码为204(成功删除) + self.log(f"Deleted run with ID {run_id}") # 打印删除成功日志 + self.deleted_count += 1 # 成功删除计数器加1 + else: # 如果响应状态码不是204(删除失败) + self.log(f"Failed to delete run with ID {run_id}. Status code: {response.status_code}") # 打印删除失败日志 def delete_old_runs(self): - while True: # 无限循环,直到没有符合条件的运行记录为止 + while True: # 无限循环,直到没有符合条件的运行记录 next_page = self.runs_url # 初始化下一页的URL为第一页 - page_number = 1 # 初始化页数计数器 any_deleted = False # 标记是否有记录被删除 while next_page: # 循环直到没有下一页 - self.log(f"Fetching page {page_number}") # 输出当前的页数 - # 发送GET请求获取一页工作流运行记录 - response = requests.get(next_page, headers={"Authorization": f"token {self.token}"}) - if response.status_code == 200: # 如果请求成功 - data = response.json() # 将响应数据转换为JSON格式 - runs = data["workflow_runs"] # 获取运行记录列表 - next_page = response.links.get("next", {}).get("url") # 获取下一页的URL - page_number += 1 # 增加页数计数器 + # 发送GET请求以获取一页工作流运行记录 + response = requests.get(next_page, headers={"Authorization": f"token {self.token}"}) # 设置请求头以进行身份验证 + if response.status_code == 200: # 如果响应状态码为200(成功) + data = response.json() # 将响应数据解析为JSON格式 + runs = data["workflow_runs"] # 获取工作流运行记录列表 + next_page = response.links.get("next", {}).get("url") # 获取下一页的URL,如果没有下一页则为None - for run in runs: # 遍历每条运行记录 - # 将运行记录的创建时间转换为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=hour_count): # 如果差值超过指定小时 默认168小时 - self.delete_run(run["id"]) # 删除运行记录 - any_deleted = True # 标记有记录被删除 - else: # 如果请求失败 - self.log(f"Failed to fetch runs. Status code: {response.status_code}") # 打印错误信息 + for run in runs: # 遍历每一条运行记录 + if self.hour_count == 0: # 如果设置的时间间隔为0小时 + if run["id"] != self.run_id: # 如果运行记录ID不等于当前运行记录ID + self.delete_run(run["id"]) # 删除该运行记录 + any_deleted = True # 标记有记录被删除 + else: # 如果设置的时间间隔大于0小时 + # 将运行记录的创建时间字符串转换为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 # 退出循环 - if not any_deleted: # 如果没有任何记录被删除,退出循环 - break + if not any_deleted: # 如果没有任何记录被删除 + break # 退出循环 + + self.log(f"Total number of deleted runs: {self.deleted_count}") # 输出总共成功删除的运行记录数量 def log(self, message): - 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] # 获取当前时间,精确到三位毫秒 + # 获取当前时间的北京时间(UTC+8) + 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}") # 打印日志消息 if __name__ == "__main__": - repository_name = os.environ.get("REPOSITORY_NAME") - github_token = os.environ.get("GITHUB_TOKEN") - hour_count = int(os.environ.get("HOUR_COUNT")) - repo_url = f"https://api.github.com/repos/{repository_name}" # GitHub仓库URL + repository_name = os.environ.get("REPOSITORY_NAME") # 从环境变量获取GitHub仓库名称 + github_token = os.environ.get("GITHUB_TOKEN") # 从环境变量获取GitHub个人访问令牌 + github_run_id = int(os.environ.get("GITHUB_RUN_ID")) # 从环境变量获取当前工作流运行记录的ID,并转换为整数 + 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个人访问令牌 - manager = GitHubActionsManager(repo_url, token) # 创建GitHubActionsManager实例 - manager.delete_old_runs() # 删除指定小时前的运行记录 + # 创建GitHubActionsManager实例 + manager = GitHubActionsManager(repo_url, token, github_run_id, hour_count) + manager.delete_old_runs() # 删除指定时间间隔(小时)之前的运行记录 diff --git a/scripts/push.py b/scripts/push.py index 177086b..fa9e0a6 100644 --- a/scripts/push.py +++ b/scripts/push.py @@ -4,27 +4,40 @@ import re from scripts.ad import get_advertise +# 定义函数send_message,用于发送消息 def send_message(token, title, content): + # 推送消息的url,格式化字符串以包含token url = f"https://push.showdoc.com.cn/server/api/push/{token}" + # 调用get_advertise函数获取广告内容 advertise = get_advertise() + # 如果广告内容不为空,则将广告内容添加到消息内容前 content = f"{advertise}{content}" if advertise else content + # 定义正则表达式模式,用于匹配包含“教学班ID”的行 pattern = re.compile(r"^.*教学班ID.*$\n?", re.MULTILINE) + # 使用正则表达式替换匹配的行为空字符串,并去除前后的空白字符 content = re.sub(pattern, "", content).strip() + # 定义一个字典,用于存储需要替换的字符串及其对应的替换值 replacements = { - "------": "\n------\n", - "个人信息:": "

个人信息

\n", - "成绩信息:": "

成绩信息

\n", - "未公布成绩的课程:": "

未公布成绩的课程

\n", - "工作流信息:": "

工作流信息

\n", - "Copyright © 2024 NianBroken. All rights reserved.": "Copyright © 2024 NianBroken. All rights reserved.", + "------": "\n------\n", # 替换多连字符为换行后的连字符 + "个人信息:": "

个人信息

\n", # 替换“个人信息:”为HTML标题标签 + "成绩信息:": "

成绩信息

\n", # 替换“成绩信息:”为HTML标题标签 + "未公布成绩的课程:": "

未公布成绩的课程

\n", # 替换“未公布成绩的课程:”为HTML标题标签 + "工作流信息:": "

工作流信息

\n", # 替换“工作流信息:”为HTML标题标签 + "Copyright © 2024 NianBroken. All rights reserved.": "Copyright © 2024 NianBroken. All rights reserved.", # 替换版权信息为带有超链接的HTML } + # 遍历字典中的所有键值对,进行替换操作 for old, new in replacements.items(): content = content.replace(old, new) + # 创建一个字典,包含标题和内容 data = {"title": title, "content": content} + # 将字典转换为JSON字符串,并编码为UTF-8 body = json.dumps(data).encode(encoding="utf-8") + # 定义HTTP请求头,设置内容类型为JSON headers = {"Content-Type": "application/json"} + # 发送POST请求,携带JSON数据和请求头 response = requests.post(url, data=body, headers=headers) - # 解析 JSON 数据 + # 解析响应的JSON数据,转换为字典 response_dict = json.loads(response.text) + # 返回响应字典 return response_dict diff --git a/scripts/user_login.py b/scripts/user_login.py index 4e65ee8..df3fa91 100644 --- a/scripts/user_login.py +++ b/scripts/user_login.py @@ -37,6 +37,6 @@ def login(url, username, password): """ elif lgn["code"] != 1000: pprint(lgn) - sys.exit() + sys.exit("登录失败") return student_client