commit 69430da9650f92476bdd0835886aa02f6f891a3e Author: NianBroken Date: Sun Jan 21 22:31:43 2024 +0800 add diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml new file mode 100644 index 0000000..d480f13 --- /dev/null +++ b/.github/workflows/main.yml @@ -0,0 +1,69 @@ +name: CheckScores + +on: + schedule: + - cron: "*/30 * * * *" # 每30分钟执行一次 + workflow_dispatch: # 允许手动触发workflow + +jobs: + CheckScores: + runs-on: ubuntu-latest # 在最新版的Ubuntu系统上运行 + + steps: + - name: Checkout Repository # 使用GitHub Actions提供的动作来检出代码库 + uses: actions/checkout@v2 + + - name: Set up Python # 配置Python环境 + uses: actions/setup-python@v2 + with: + python-version: 3.x # 使用Python 3.x版本 + + - name: Install dependencies # 使用pip安装项目的依赖项 + run: | + pip install requests rsa pyquery + + - name: Configure Git # 配置Git用户信息 + run: | + git config --global user.email "actions@github.com" + git config --global user.name "GitHub Actions" + + - name: Fetch upstream changes # 添加上游仓库作为远程仓库，并获取上游仓库的main分支的最新更改 + run: | + git remote add upstream https://github.com/NianBroken/ZFCheckScores.git + git fetch upstream main + + - name: Backup local .txt files # 创建一个备份目录，并将当前目录中的所有.txt文件复制到备份目录 + run: | + mkdir backup + cp *.txt backup/ + + - name: Force push changes from upstream to current main branch # 将上游main分支的更改强制推送到当前main分支 + run: | + git checkout main + git reset --hard upstream/main + git push origin main --force + + - name: Restore local .txt files and delete backup # 从备份目录将.txt文件还原到当前目录，然后删除备份目录 + run: | + cp backup/*.txt . + rm -rf backup + + - name: Run main.py # 运行主程序main.py + run: python main.py + env: + URL: ${{ secrets.URL }} + USERNAME: ${{ secrets.USERNAME }} + PASSWORD: ${{ secrets.PASSWORD }} + TOKEN: ${{ secrets.TOKEN }} + + - name: Add changes # 将当前目录中的所有更改添加到Git的暂存区 + run: | + git add . + + - name: Commit changes # 提交更改到Git仓库 + run: | + git commit -m "Update from GitHub Actions" || true # 如果没有更改，则允许提交失败，但不中断workflow + + - name: Force push changes to main branch # 将更改强制推送到main分支 + run: | + git push origin main --force diff --git a/README.md b/README.md new file mode 100644 index 0000000..a8e5882 --- /dev/null +++ b/README.md @@ -0,0 +1,95 @@ +# 正方教务系统成绩推送 + +## 作用 + +**使用本项目前：** + +睡醒看一遍教务系统、上厕所看一遍教务系统、刷牙看一遍教务系统、洗脸看一遍教务系统、吃早餐看一遍教务系统、吃午饭看一遍教务系统、睡午觉前看一遍教务系统、午觉醒来看一遍教务系统、吃完晚饭看一遍教务系统、洗澡看一遍教务系统、睡觉之前看一遍教务系统。 + +**使用本项目后：** + +成绩更新后自动发通知到微信 以节省您宝贵的时间 + +## 测试环境 + +正方教务系统 版本 V8.0.0 + +## 目前支持的功能 + +1. 每隔 30 分钟自动检测成绩是否更新 若更新则向微信推送通知 + +2. 显示成绩提交时间（何时将成绩录入教务系统） + +3. 显示成绩提交人姓名（何人将成绩录入教务系统） + +## 使用方法 + +### 1. Fork 本仓库 + + + + +### 2. 开启工作流读写权限 + + + + + +### 3. 添加 Secrets + +| Name | 例子 | 说明 | +| -------- | -------------------------- | ----------------- | +| URL | https://www.nianbroken.top | 教务系统地址 | +| USERNAME | 2971802058 | 教务系统用户名 | +| PASSWORD | Y3xhaCkb5PZ4 | 教务系统密码 | +| TOKEN | J65KWMBfyDh3YPLpcvm8 | Pushplus 的 token | + + + + +获取 TOKEN： + +- [登录 Pushplus ](https://www.pushplus.plus/login.html) + + +- [获取 Token](https://www.pushplus.plus/api/open/user/token) + +打开页面后，你会得到一个类似下列所示的 Json 代码块，`data`中的值就是 TOKEN + +```json +{ + "code": 200, + "msg": "请求成功", + "data": "cd735c356aa14d16b1452aa932ac89cc", + "count": null +} +``` + +- [相关文档](https://www.pushplus.plus/doc/guide/openApi.html#_1-%E8%8E%B7%E5%8F%96token) + +当你的 Secrets 添加完成后，你的页面应该是类似下图所示 + + +### 4. 开启 Actions + + + + +### 5. 运行程序 + + + + +当你的页面应该是类似下图所示时则表示程序运行结束且未报错 + +在此之后 程序将会每隔 30 分钟自动运行一次 + +### 6. 查看运行结果 + + + + +### 7. 成绩更新通知 + +当检测到成绩更新时，你的页面应该是类似下图所示 + diff --git a/img/1.png b/img/1.png new file mode 100644 index 0000000..669d89b Binary files /dev/null and b/img/1.png differ diff --git a/img/10.png b/img/10.png new file mode 100644 index 0000000..15aae3a Binary files /dev/null and b/img/10.png differ diff --git a/img/11.png b/img/11.png new file mode 100644 index 0000000..348a392 Binary files /dev/null and b/img/11.png differ diff --git a/img/12.png b/img/12.png new file mode 100644 index 0000000..fbf3078 Binary files /dev/null and b/img/12.png differ diff --git a/img/13.png b/img/13.png new file mode 100644 index 0000000..aa48948 Binary files /dev/null and b/img/13.png differ diff --git a/img/14.png b/img/14.png new file mode 100644 index 0000000..17c26c8 Binary files /dev/null and b/img/14.png differ diff --git a/img/15.png b/img/15.png new file mode 100644 index 0000000..59a5bb2 Binary files /dev/null and b/img/15.png differ diff --git a/img/16.png b/img/16.png new file mode 100644 index 0000000..655f07d Binary files /dev/null and b/img/16.png differ diff --git a/img/17.png b/img/17.png new file mode 100644 index 0000000..a8ee3e0 Binary files /dev/null and b/img/17.png differ diff --git a/img/2.png b/img/2.png new file mode 100644 index 0000000..09856fc Binary files /dev/null and b/img/2.png differ diff --git a/img/3.png b/img/3.png new file mode 100644 index 0000000..38252f3 Binary files /dev/null and b/img/3.png differ diff --git a/img/4.png b/img/4.png new file mode 100644 index 0000000..6fcf405 Binary files /dev/null and b/img/4.png differ diff --git a/img/5.png b/img/5.png new file mode 100644 index 0000000..e614291 Binary files /dev/null and b/img/5.png differ diff --git a/img/6.png b/img/6.png new file mode 100644 index 0000000..005722a Binary files /dev/null and b/img/6.png differ diff --git a/img/7.png b/img/7.png new file mode 100644 index 0000000..24ef630 Binary files /dev/null and b/img/7.png differ diff --git a/img/8.png b/img/8.png new file mode 100644 index 0000000..c0e831c Binary files /dev/null and b/img/8.png differ diff --git a/img/9.png b/img/9.png new file mode 100644 index 0000000..19ea27a Binary files /dev/null and b/img/9.png differ diff --git a/main.py b/main.py new file mode 100644 index 0000000..a2a5e7a --- /dev/null +++ b/main.py @@ -0,0 +1,143 @@ +import base64 +import hashlib +import os +import sys +from pprint import pprint +from zfn_api import Client +from pushplus import send_message + +# 教务系统的URL、用户名和密码 +url = os.environ.get("URL") +username = os.environ.get("USERNAME") +password = os.environ.get("PASSWORD") + + +# 定义一个md5加密的封装函数 +def md5_encrypt(string): + # 将字符串转换为字节 + byte_string = string.encode() + # 创建一个MD5对象 + md5_object = hashlib.md5(byte_string) + # 返回MD5加密后的十六进制字符串 + return md5_object.hexdigest() + + +# 初始化相关变量 +cookies = {} # 初始化cookies为空字典 +base_url = url +raspisanie = [] +ignore_type = [] +detail_category_type = [] +timeout = 5 +stu = Client( + cookies=cookies, + base_url=base_url, + raspisanie=raspisanie, + ignore_type=ignore_type, + detail_category_type=detail_category_type, + timeout=timeout, +) + +# 如果cookies为空字典，则进行登录 +if not cookies: + lgn = stu.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 = stu.login_with_kaptcha(**verify_data) + if ret["code"] != 1000: + pprint(ret) + sys.exit() + pprint(ret) + elif lgn["code"] != 1000: + pprint(lgn) + sys.exit() + +# 获取个人信息 +info = stu.get_info()["data"] +# 整合个人信息 +integrate_info = ( + f"个人信息：\n" f"学号：{info['sid']}\n" f"班级：{info['class_name']}\n" f"姓名：{info['name']}" +) + +# 获取成绩信息 +firstrun_file_path = "firstrun.txt" + +# 如果firstrun.txt文件不存在，创建并写入true +if not os.path.exists(firstrun_file_path): + with open(firstrun_file_path, "w") as firstrun_file: + firstrun_file.write("true") + +# 第一次运行则运行两遍，否则运行一遍 +run_count = 2 if open(firstrun_file_path).read() == "true" else 1 +for _ in range(run_count): + # 如果grade.txt文件不存在，创建文件 + if not os.path.exists("grade.txt"): + open("grade.txt", "w").close() + + # 清空old_grade.txt文件内容 + with open("old_grade.txt", "w"): + pass + + # 将grade.txt文件中的内容写入到old_grade.txt文件内。 + with open("grade.txt", "r") as grade_file, open( + "old_grade.txt", "w" + ) as old_grade_file: + old_grade_file.write(grade_file.read()) + + # 清空grade.txt文件内容 + with open("grade.txt", "w"): + pass + + # 将下列代码输出的所有内容都写入到grade.txt文件内。 + integrated_grade_info = "成绩信息：" + grade = stu.get_grade(2023)["data"]["courses"] + for course in grade: + # 整合成绩信息 + integrated_grade_info += ( + f"\n课程ID: {course['course_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"------" + ) + + # 加密保存成绩 + encrypt_integrated_grade_info = md5_encrypt(integrated_grade_info) + + # 将加密后的成绩信息写入grade.txt文件 + with open("grade.txt", "w") as grade_file: + grade_file.write(encrypt_integrated_grade_info) + +# 对grade.txt和old_grade.txt两个文件的内容进行比对 +with open("grade.txt", "r") as grade_file, open("old_grade.txt", "r") as old_grade_file: + # 读取grade.txt和old_grade.txt文件的内容 + grade_content = grade_file.read() + old_grade_content = old_grade_file.read() + +# 输出成绩信息 +print(f"新成绩：{encrypt_integrated_grade_info}") +print(f"旧成绩：{old_grade_content}") +print("------") + +# 整合所有信息 +integrate_send_info = f"{integrate_info}\n------\n{integrated_grade_info}" + +# 输出成绩是否更新 +if grade_content == old_grade_content: + print("成绩未更新") +else: + print("成绩已更新") + response_text = send_message("教务成绩已更新", integrate_send_info) + print(response_text) + +# 更新firstrun.txt文件内容为false +if open(firstrun_file_path).read() == "true": + with open(firstrun_file_path, "w") as firstrun_file: + # 将firstrun.txt文件内容更新为false + firstrun_file.write("false") diff --git a/pushplus.py b/pushplus.py new file mode 100644 index 0000000..6ae6bdf --- /dev/null +++ b/pushplus.py @@ -0,0 +1,14 @@ +import requests +import json +import os + + +def send_message(title, content): + token = os.environ.get("TOKEN") + url = "http://www.pushplus.plus/send" + data = {"token": token, "title": title, "content": content} + body = json.dumps(data).encode(encoding="utf-8") + headers = {"Content-Type": "application/json"} + response = requests.post(url, data=body, headers=headers) + + return response.text diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e69de29 diff --git a/zfn_api.py b/zfn_api.py new file mode 100644 index 0000000..a164f03 --- /dev/null +++ b/zfn_api.py @@ -0,0 +1,1537 @@ +import base64 +import binascii +import json +import re +import time +import traceback +import unicodedata +from urllib.parse import urljoin + +import requests +import rsa +from pyquery import PyQuery as pq +from requests import exceptions + +RASPIANIE = [ + ["8:00", "8:40"], + ["8:45", "9:25"], + ["9:30", "10:10"], + ["10:30", "11:10"], + ["11:15", "11:55"], + ["14:30", "15:10"], + ["15:15", "15:55"], + ["16:05", "16:45"], + ["16:50", "17:30"], + ["18:40", "19:20"], + ["19:25", "20:05"], + ["20:10", "20:50"], + ["20:55", "21:35"], +] + + +class Client: + raspisanie = [] + ignore_type = [] + + def __init__(self, cookies={}, **kwargs): + # 基础配置 + self.base_url = kwargs.get("base_url") + self.raspisanie = kwargs.get("raspisanie", RASPIANIE) + self.ignore_type = kwargs.get("ignore_type", []) + self.detail_category_type = kwargs.get("detail_category_type", []) + self.timeout = kwargs.get("timeout", 3) + Client.raspisanie = self.raspisanie + Client.ignore_type = self.ignore_type + + self.key_url = urljoin(self.base_url, "xtgl/login_getPublicKey.html") + self.login_url = urljoin(self.base_url, "xtgl/login_slogin.html") + self.kaptcha_url = urljoin(self.base_url, "kaptcha") + self.headers = requests.utils.default_headers() + self.headers["Referer"] = self.login_url + self.headers[ + "User-Agent" + ] = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36" + self.headers[ + "Accept" + ] = "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3" + self.sess = requests.Session() + self.sess.keep_alive = False + self.cookies = cookies + + def login(self, sid, password): + """登录教务系统""" + need_verify = False + try: + # 登录页 + req_csrf = self.sess.get( + self.login_url, headers=self.headers, timeout=self.timeout + ) + if req_csrf.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + # 获取csrf_token + doc = pq(req_csrf.text) + csrf_token = doc("#csrftoken").attr("value") + pre_cookies = self.sess.cookies.get_dict() + # 获取publicKey并加密密码 + req_pubkey = self.sess.get( + self.key_url, headers=self.headers, timeout=self.timeout + ).json() + modulus = req_pubkey["modulus"] + exponent = req_pubkey["exponent"] + if str(doc("input#yzm")) == "": + # 不需要验证码 + encrypt_password = self.encrypt_password(password, modulus, exponent) + # 登录数据 + login_data = { + "csrftoken": csrf_token, + "yhm": sid, + "mm": encrypt_password, + } + # 请求登录 + req_login = self.sess.post( + self.login_url, + headers=self.headers, + data=login_data, + timeout=self.timeout, + ) + doc = pq(req_login.text) + tips = doc("p#tips") + if str(tips) != "": + if "用户名或密码" in tips.text(): + return {"code": 1002, "msg": "用户名或密码不正确"} + return {"code": 998, "msg": tips.text()} + self.cookies = self.sess.cookies.get_dict() + return {"code": 1000, "msg": "登录成功", "data": {"cookies": self.cookies}} + # 需要验证码，返回相关页面验证信息给用户，TODO: 增加更多验证方式 + need_verify = True + req_kaptcha = self.sess.get( + self.kaptcha_url, headers=self.headers, timeout=self.timeout + ) + kaptcha_pic = base64.b64encode(req_kaptcha.content).decode() + return { + "code": 1001, + "msg": "获取验证码成功", + "data": { + "sid": sid, + "csrf_token": csrf_token, + "cookies": pre_cookies, + "password": password, + "modulus": modulus, + "exponent": exponent, + "kaptcha_pic": kaptcha_pic, + "timestamp": time.time(), + }, + } + except exceptions.Timeout: + msg = "获取验证码超时" if need_verify else "登录超时" + return {"code": 1003, "msg": msg} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + msg = "获取验证码时未记录的错误" if need_verify else "登录时未记录的错误" + return {"code": 999, "msg": f"{msg}：{str(e)}"} + + def login_with_kaptcha( + self, sid, csrf_token, cookies, password, modulus, exponent, kaptcha, **kwargs + ): + """需要验证码的登陆""" + try: + encrypt_password = self.encrypt_password(password, modulus, exponent) + login_data = { + "csrftoken": csrf_token, + "yhm": sid, + "mm": encrypt_password, + "yzm": kaptcha, + } + req_login = self.sess.post( + self.login_url, + headers=self.headers, + cookies=cookies, + data=login_data, + timeout=self.timeout, + ) + if req_login.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + # 请求登录 + doc = pq(req_login.text) + tips = doc("p#tips") + if str(tips) != "": + if "验证码" in tips.text(): + return {"code": 1004, "msg": "验证码输入错误"} + if "用户名或密码" in tips.text(): + return {"code": 1002, "msg": "用户名或密码不正确"} + return {"code": 998, "msg": tips.text()} + self.cookies = self.sess.cookies.get_dict() + # 不同学校系统兼容差异 + if not self.cookies.get("route"): + route_cookies = { + "JSESSIONID": self.cookies["JSESSIONID"], + "route": cookies["route"], + } + self.cookies = route_cookies + else: + return {"code": 1000, "msg": "登录成功", "data": {"cookies": self.cookies}} + except exceptions.Timeout: + return {"code": 1003, "msg": "登录超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "验证码登录时未记录的错误：" + str(e)} + + def get_info(self): + """获取个人信息""" + url = urljoin(self.base_url, "xsxxxggl/xsxxwh_cxCkDgxsxx.html?gnmkdm=N100801") + try: + req_info = self.sess.get( + url, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_info.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_info.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + info = req_info.json() + if info is None: + return self._get_info() + result = { + "sid": info.get("xh"), + "class_name": info.get("bh_id", info.get("xjztdm")), + "name": info.get("xm"), + } + return {"code": 1000, "msg": "获取个人信息成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取个人信息超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取个人信息时未记录的错误：" + str(e)} + + def _get_info(self): + """获取个人信息""" + url = urljoin(self.base_url, "xsxxxggl/xsgrxxwh_cxXsgrxx.html?gnmkdm=N100801") + try: + req_info = self.sess.get( + url, headers=self.headers, cookies=self.cookies, timeout=self.timeout + ) + if req_info.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_info.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + pending_result = {} + # 学生基本信息 + for ul_item in doc.find("div.col-sm-6").items(): + content = pq(ul_item).find("div.form-group") + # key = re.findall(r'^[\u4E00-\u9FA5A-Za-z0-9]+', pq(content).find('label.col-sm-4.control-label').text())[0] + key = pq(content).find("label.col-sm-4.control-label").text() + value = pq(content).find("div.col-sm-8 p.form-control-static").text() + # 到这一步，解析到的数据基本就是一个键值对形式的html数据了，比如"[学号：]:123456" + pending_result[key] = value + # 学生学籍信息，其他信息，联系方式 + for ul_item in doc.find("div.col-sm-4").items(): + content = pq(ul_item).find("div.form-group") + key = pq(content).find("label.col-sm-4.control-label").text() + value = pq(content).find("div.col-sm-8 p.form-control-static").text() + # 到这一步，解析到的数据基本就是一个键值对形式的html数据了，比如"[学号：]:123456" + pending_result[key] = value + if pending_result.get("学号：") == "": + return { + "code": 1014, + "msg": "当前学年学期无学生时盒数据，您可能已经毕业了。\n\n如果是专升本同学，请使用专升本后的新学号登录～", + } + result = { + "sid": pending_result["学号："], + "name": pending_result["姓名："], + # "birthday": "无" if pending_result.get("出生日期：") == '' else pending_result["出生日期："], + # "id_number": "无" if pending_result.get("证件号码：") == '' else pending_result["证件号码："], + # "candidate_number": "无" if pending_result.get("考生号：") == '' else pending_result["考生号："], + # "status": "无" if pending_result.get("学籍状态：") == '' else pending_result["学籍状态："], + # "entry_date": "无" if pending_result.get("入学日期：") == '' else pending_result["入学日期："], + # "graduation_school": "无" if pending_result.get("毕业中学：") == '' else pending_result["毕业中学："], + "domicile": "无" + if pending_result.get("籍贯：") == "" + else pending_result["籍贯："], + "phone_number": "无" + if pending_result.get("手机号码：") == "" + else pending_result["手机号码："], + "parents_number": "无", + "email": "无" + if pending_result.get("电子邮箱：") == "" + else pending_result["电子邮箱："], + "political_status": "无" + if pending_result.get("政治面貌：") == "" + else pending_result["政治面貌："], + "national": "无" + if pending_result.get("民族：") == "" + else pending_result["民族："], + # "education": "无" if pending_result.get("培养层次：") == '' else pending_result["培养层次："], + # "postal_code": "无" if pending_result.get("邮政编码：") == '' else pending_result["邮政编码："], + # "grade": int(pending_result["学号："][0:4]), + } + if pending_result.get("学院名称：") is not None: + # 如果在个人信息页面获取到了学院班级 + result.update( + { + "college_name": "无" + if pending_result.get("学院名称：") == "" + else pending_result["学院名称："], + "major_name": "无" + if pending_result.get("专业名称：") == "" + else pending_result["专业名称："], + "class_name": "无" + if pending_result.get("班级名称：") == "" + else pending_result["班级名称："], + } + ) + else: + # 如果个人信息页面获取不到学院班级，则此处需要请求另外一个地址以获取学院、专业、班级等信息 + _url = urljoin( + self.base_url, + "xszbbgl/xszbbgl_cxXszbbsqIndex.html?doType=details&gnmkdm=N106005", + ) + _req_info = self.sess.post( + _url, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + data={"offDetails": "1", "gnmkdm": "N106005", "czdmKey": "00"}, + ) + _doc = pq(_req_info.text) + if _doc("p.error_title").text() != "无功能权限，": + # 通过学生证补办申请入口，来补全部分信息 + for ul_item in _doc.find("div.col-sm-6").items(): + content = pq(ul_item).find("div.form-group") + key = ( + pq(content).find("label.col-sm-4.control-label").text() + + "：" + ) # 为了保持格式一致，这里加个冒号 + value = ( + pq(content).find("div.col-sm-8 label.control-label").text() + ) + # 到这一步，解析到的数据基本就是一个键值对形式的html数据了，比如"[学号：]:123456" + pending_result[key] = value + result.update( + { + "college_name": "无" + if pending_result.get("学院：") is None + else pending_result["学院："], + "major_name": "无" + if pending_result.get("专业：") is None + else pending_result["专业："], + "class_name": "无" + if pending_result.get("班级：") is None + else pending_result["班级："], + } + ) + return {"code": 1000, "msg": "获取个人信息成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取个人信息超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取个人信息时未记录的错误：" + str(e)} + + def get_grade(self, year: int, term: int = 0, use_personal_info: bool = False): + """ + 获取成绩 + use_personal_info: 是否使用获取个人信息接口获取成绩 + """ + url = urljoin( + self.base_url, + "cjcx/cjcx_cxDgXscj.html?doType=query&gnmkdm=N305005" + if use_personal_info + else "cjcx/cjcx_cxXsgrcj.html?doType=query&gnmkdm=N305005", + ) + temp_term = term + term = term**2 * 3 + term = "" if term == 0 else term + data = { + "xnm": str(year), # 学年数 + "xqm": str(term), # 学期数，第一学期为3，第二学期为12, 整个学年为空'' + "_search": "false", + "nd": int(time.time() * 1000), + "queryModel.showCount": "100", # 每页最多条数 + "queryModel.currentPage": "1", + "queryModel.sortName": "", + "queryModel.sortOrder": "asc", + "time": "0", # 查询次数 + } + try: + req_grade = self.sess.post( + url, + headers=self.headers, + data=data, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_grade.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_grade.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + grade = req_grade.json() + grade_items = grade.get("items") + if not grade_items: + return {"code": 1005, "msg": "获取内容为空"} + result = { + "sid": grade_items[0]["xh"], + "name": grade_items[0]["xm"], + "year": year, + "term": temp_term, + "count": len(grade_items), + "courses": [ + { + "course_id": i.get("kch_id"), + "title": i.get("kcmc"), + "teacher": i.get("jsxm"), + "grade": self.parse_int(i.get("cj")), + "submission_time": i.get("tjsj"), + "name_of_submitter": i.get("tjrxm"), + } + for i in grade_items + ], + } + return {"code": 1000, "msg": "获取成绩成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取成绩超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取成绩时未记录的错误：" + str(e)} + + def get_schedule(self, year: int, term: int): + """获取课程表信息""" + url = urljoin(self.base_url, "kbcx/xskbcx_cxXsKb.html?gnmkdm=N2151") + temp_term = term + term = term**2 * 3 + data = {"xnm": str(year), "xqm": str(term)} + try: + req_schedule = self.sess.post( + url, + headers=self.headers, + data=data, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_schedule.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_schedule.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + schedule = req_schedule.json() + if not schedule.get("kbList"): + return {"code": 1005, "msg": "获取内容为空"} + result = { + "sid": schedule["xsxx"].get("XH"), + "name": schedule["xsxx"].get("XM"), + "year": year, + "term": temp_term, + "count": len(schedule["kbList"]), + "courses": [ + { + "course_id": i.get("kch_id"), + "title": i.get("kcmc"), + "teacher": i.get("xm"), + "class_name": i.get("jxbmc"), + "credit": self.align_floats(i.get("xf")), + "weekday": self.parse_int(i.get("xqj")), + "time": self.display_course_time(i.get("jc")), + "sessions": i.get("jc"), + "list_sessions": self.list_sessions(i.get("jc")), + "weeks": i.get("zcd"), + "list_weeks": self.list_weeks(i.get("zcd")), + "evaluation_mode": i.get("khfsmc"), + "campus": i.get("xqmc"), + "place": i.get("cdmc"), + "hours_composition": i.get("kcxszc"), + "weekly_hours": self.parse_int(i.get("zhxs")), + "total_hours": self.parse_int(i.get("zxs")), + } + for i in schedule["kbList"] + ], + "extra_courses": [i.get("qtkcgs") for i in schedule.get("sjkList")], + } + result = self.split_merge_display(result) + return {"code": 1000, "msg": "获取课表成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取课表超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取课表时未记录的错误：" + str(e)} + + def get_academia(self): + """获取学业生涯情况""" + url_main = urljoin( + self.base_url, + "xsxy/xsxyqk_cxXsxyqkIndex.html?gnmkdm=N105515&layout=default", + ) + url_info = urljoin( + self.base_url, "xsxy/xsxyqk_cxJxzxjhxfyqKcxx.html?gnmkdm=N105515" + ) + try: + req_main = self.sess.get( + url_main, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + stream=True, + ) + if req_main.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc_main = pq(req_main.text) + if doc_main("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + if str(doc_main("div.alert-danger")) != "": + return {"code": 998, "msg": doc_main("div.alert-danger").text()} + sid = doc_main("form#form input#xh_id").attr("value") + display_statistics = ( + doc_main("div#alertBox").text().replace(" ", "").replace("\n", "") + ) + sid = doc_main("input#xh_id").attr("value") + statistics = self.get_academia_statistics(display_statistics) + type_statistics = self.get_academia_type_statistics(req_main.text) + details = {} + for type in type_statistics.keys(): + details[type] = self.sess.post( + url_info, + headers=self.headers, + data={"xfyqjd_id": type_statistics[type]["id"]}, + cookies=self.cookies, + timeout=self.timeout, + stream=True, + ).json() + result = { + "sid": sid, + "statistics": statistics, + "details": [ + { + "type": type, + "credits": type_statistics[type]["credits"], + "courses": [ + { + "course_id": i.get("KCH"), + "title": i.get("KCMC"), + "situation": self.parse_int(i.get("XDZT")), + "display_term": self.get_display_term( + sid, i.get("JYXDXNM"), i.get("JYXDXQMC") + ), + "credit": self.align_floats(i.get("XF")), + "category": self.get_course_category(type, i), + "nature": i.get("KCXZMC"), + "max_grade": self.parse_int(i.get("MAXCJ")), + "grade_point": self.align_floats(i.get("JD")), + } + for i in details[type] + ], + } + for type in type_statistics.keys() + if len(details[type]) > 0 + ], + } + return {"code": 1000, "msg": "获取学业情况成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取学业情况超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取学业情况时未记录的错误：" + str(e)} + + def get_academia_pdf(self): + """获取学业生涯（学生成绩总表）pdf""" + url_view = urljoin(self.base_url, "bysxxcx/xscjzbdy_dyXscjzbView.html") + url_window = urljoin(self.base_url, "bysxxcx/xscjzbdy_dyCjdyszxView.html") + url_policy = urljoin(self.base_url, "xtgl/bysxxcx/xscjzbdy_cxXsCount.html") + url_filetype = urljoin(self.base_url, "bysxxcx/xscjzbdy_cxGswjlx.html") + url_common = urljoin(self.base_url, "common/common_cxJwxtxx.html") + url_file = urljoin(self.base_url, "bysxxcx/xscjzbdy_dyList.html") + url_progress = urljoin(self.base_url, "xtgl/progress_cxProgressStatus.html") + data = { + "gsdygx": "10628-zw-mrgs", + "ids": "", + "bdykcxzDms": "", + "cytjkcxzDms": "", + "cytjkclbDms": "", + "cytjkcgsDms": "", + "bjgbdykcxzDms": "", + "bjgbdyxxkcxzDms": "", + "djksxmDms": "", + "cjbzmcDms": "", + "cjdySzxs": "", + "wjlx": "pdf", + } + + try: + data_view = {"time": str(round(time.time() * 1000)), "gnmkdm": "N558020"} + data_params = data_view + del data_params["time"] + # View接口 + req_view = self.sess.post( + url_view, + headers=self.headers, + data=data_view, + params=data_view, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_view.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_view.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + # Window接口 + data_window = {"xh": ""} + self.sess.post( + url_window, + headers=self.headers, + data=data_window, + params=data_params, + cookies=self.cookies, + timeout=self.timeout, + ) + # 许可接口 + data_policy = data + del data_policy["wjlx"] + self.sess.post( + url_policy, + headers=self.headers, + data=data_policy, + params=data_params, + cookies=self.cookies, + timeout=self.timeout, + ) + # 文件类型接口 + data_filetype = data_policy + self.sess.post( + url_filetype, + headers=self.headers, + data=data_filetype, + params=data_params, + cookies=self.cookies, + timeout=self.timeout, + ) + # Common接口 + self.sess.post( + url_common, + headers=self.headers, + data=data_params, + params=data_params, + cookies=self.cookies, + timeout=self.timeout, + ) + # 获取PDF文件URL + req_file = self.sess.post( + url_file, + headers=self.headers, + data=data, + params=data_params, + cookies=self.cookies, + timeout=self.timeout, + ) + doc = pq(req_file.text) + if "错误" in doc("title").text(): + error = doc("p.error_title").text() + return {"code": 998, "msg": error} + # 进度接口 + data_progress = { + "key": "score_print_processed", + "gnmkdm": "N558020", + } + self.sess.post( + url_progress, + headers=self.headers, + data=data_progress, + params=data_progress, + cookies=self.cookies, + timeout=self.timeout, + ) + # 生成PDF文件URL + pdf = ( + req_file.text.replace("#成功", "") + .replace('"', "") + .replace("/", "\\") + .replace("\\\\", "/") + ) + # 下载PDF文件 + req_pdf = self.sess.get( + urljoin(self.base_url, pdf), + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout + 2, + ) + result = req_pdf.content # 二进制内容 + return {"code": 1000, "msg": "获取学生成绩总表pdf成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取成绩总表pdf超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取成绩总表pdf时未记录的错误：" + str(e)} + + def get_schedule_pdf(self, year: int, term: int, name: str = "导出"): + """获取课表pdf""" + url_policy = urljoin(self.base_url, "kbdy/bjkbdy_cxXnxqsfkz.html") + url_file = urljoin(self.base_url, "kbcx/xskbcx_cxXsShcPdf.html") + origin_term = term + term = term**2 * 3 + data = { + "xm": name, + "xnm": str(year), + "xqm": str(term), + "xnmc": f"{year}-{year + 1}", + "xqmmc": str(origin_term), + "jgmc": "undefined", + "xxdm": "", + "xszd.sj": "true", + "xszd.cd": "true", + "xszd.js": "true", + "xszd.jszc": "false", + "xszd.jxb": "true", + "xszd.xkbz": "true", + "xszd.kcxszc": "true", + "xszd.zhxs": "true", + "xszd.zxs": "true", + "xszd.khfs": "true", + "xszd.xf": "true", + "xszd.skfsmc": "false", + "kzlx": "dy", + } + + try: + # 许可接口 + pilicy_params = {"gnmkdm": "N2151"} + req_policy = self.sess.post( + url_policy, + headers=self.headers, + data=data, + params=pilicy_params, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_policy.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_policy.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + # 获取PDF文件URL + file_params = {"doType": "table"} + req_file = self.sess.post( + url_file, + headers=self.headers, + data=data, + params=file_params, + cookies=self.cookies, + timeout=self.timeout, + ) + doc = pq(req_file.text) + if "错误" in doc("title").text(): + error = doc("p.error_title").text() + return {"code": 998, "msg": error} + result = req_file.content # 二进制内容 + return {"code": 1000, "msg": "获取课程表pdf成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取课程表pdf超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取课程表pdf时未记录的错误：" + str(e)} + + def get_notifications(self): + """获取通知消息""" + url = urljoin(self.base_url, "xtgl/index_cxDbsy.html?doType=query") + data = { + "sfyy": "0", # 是否已阅，未阅未1，已阅为2 + "flag": "1", + "_search": "false", + "nd": int(time.time() * 1000), + "queryModel.showCount": "1000", # 最多条数 + "queryModel.currentPage": "1", # 当前页数 + "queryModel.sortName": "cjsj", + "queryModel.sortOrder": "desc", # 时间倒序, asc正序 + "time": "0", + } + try: + req_notification = self.sess.post( + url, + headers=self.headers, + data=data, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_notification.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_notification.text) + if doc("h5").text() == "用户登录" or "错误" in doc("title").text(): + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + notifications = req_notification.json() + result = [ + {**self.split_notifications(i), "create_time": i.get("cjsj")} + for i in notifications.get("items") + ] + return {"code": 1000, "msg": "获取消息成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取消息超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": "获取消息时未记录的错误：" + str(e)} + + def get_selected_courses(self, year: int, term: int): + """获取已选课程信息""" + try: + url = urljoin( + self.base_url, + "xsxk/zzxkyzb_cxZzxkYzbChoosedDisplay.html?gnmkdm=N253512", + ) + temp_term = term + term = term**2 * 3 + data = {"xkxnm": str(year), "xkxqm": str(term)} + req_selected = self.sess.post( + url, + data=data, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_selected.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_selected.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + selected = req_selected.json() + result = { + "year": year, + "term": temp_term, + "count": len(selected), + "courses": [ + { + "course_id": i.get("kch"), + "class_id": i.get("jxb_id"), + "do_id": i.get("do_jxb_id"), + "title": i.get("kcmc"), + "teacher_id": (re.findall(r"(.*?\d+)/", i.get("jsxx")))[0], + "teacher": (re.findall(r"/(.*?)/", i.get("jsxx")))[0], + "credit": float(i.get("xf", 0)), + "category": i.get("kklxmc"), + "capacity": int(i.get("jxbrs", 0)), + "selected_number": int(i.get("yxzrs", 0)), + "place": self.get_place(i.get("jxdd")), + "time": self.get_course_time(i.get("sksj")), + "optional": int(i.get("zixf", 0)), + "waiting": i.get("sxbj"), + } + for i in selected + ], + } + return {"code": 1000, "msg": "获取已选课程成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取已选课程超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": f"获取已选课程时未记录的错误：{str(e)}"} + + def get_block_courses(self, year: int, term: int, block: int): + """获取板块课选课列表""" + # TODO: 优化代码 + try: + # 获取head_data + url_head = urljoin( + self.base_url, + "xsxk/zzxkyzb_cxZzxkYzbIndex.html?gnmkdm=N253512&layout=default", + ) + req_head_data = self.sess.get( + url_head, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_head_data.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_head_data.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + if str(doc("div.nodata")) != "": + return {"code": 998, "msg": doc("div.nodata").text()} + got_credit_list = [i for i in doc("font[color='red']").items()] + if len(got_credit_list) == 0: + return {"code": 1005, "msg": "板块课内容为空"} + head_data = {"got_credit": got_credit_list[2].string} + + kklxdm_list = [] + xkkz_id_list = [] + for tab_content in doc("a[role='tab']").items(): + onclick_content = tab_content.attr("onclick") + r = re.findall(r"'(.*?)'", str(onclick_content)) + kklxdm_list.append(r[0].strip()) + xkkz_id_list.append(r[1].strip()) + head_data["bkk1_kklxdm"] = kklxdm_list[0] + head_data["bkk2_kklxdm"] = kklxdm_list[1] + head_data["bkk3_kklxdm"] = kklxdm_list[2] + head_data["bkk1_xkkz_id"] = xkkz_id_list[0] + head_data["bkk2_xkkz_id"] = xkkz_id_list[1] + head_data["bkk3_xkkz_id"] = xkkz_id_list[2] + + for head_data_content in doc("input[type='hidden']"): + name = head_data_content.attr("name") + value = head_data_content.attr("value") + head_data[str(name)] = str(value) + + url_display = urljoin( + self.base_url, "xsxk/zzxkyzb_cxZzxkYzbDisplay.html?gnmkdm=N253512" + ) + display_req_data = { + "xkkz_id": head_data[f"bkk{block}_xkkz_id"], + "xszxzt": "1", + "kspage": "0", + } + req_display_data = self.sess.post( + url_display, + headers=self.headers, + data=display_req_data, + cookies=self.cookies, + timeout=self.timeout, + ) + doc_display = pq(req_display_data.text) + display_data = {} + for display_data_content in doc_display("input[type='hidden']").items(): + name = display_data_content.get("name") + value = display_data_content.get("value") + display_data[str(name)] = str(value) + head_data.update(display_data) + + # 获取课程列表 + url_kch = urljoin( + self.base_url, "xsxk/zzxkyzb_cxZzxkYzbPartDisplay.html?gnmkdm=N253512" + ) + url_bkk = urljoin( + self.base_url, "xsxk/zzxkyzb_cxJxbWithKchZzxkYzb.html?gnmkdm=N253512" + ) + term = term**2 * 3 + kch_data = { + "bklx_id": head_data["bklx_id"], + "xqh_id": head_data["xqh_id"], + "zyfx_id": head_data["zyfx_id"], + "njdm_id": head_data["njdm_id"], + "bh_id": head_data["bh_id"], + "xbm": head_data["xbm"], + "xslbdm": head_data["xslbdm"], + "ccdm": head_data["ccdm"], + "xsbj": head_data["xsbj"], + "xkxnm": str(year), + "xkxqm": str(term), + "kklxdm": head_data[f"bkk{block}_kklxdm"], + "kkbk": head_data["kkbk"], + "rwlx": head_data["rwlx"], + "kspage": "1", + "jspage": "10", + } + kch_res = self.sess.post( + url_kch, + headers=self.headers, + data=kch_data, + cookies=self.cookies, + timeout=self.timeout, + ) + jkch_res = kch_res.json() + bkk_data = { + "bklx_id": head_data["bklx_id"], + "xkxnm": str(year), + "xkxqm": str(term), + "xkkz_id": head_data[f"bkk{block}_xkkz_id"], + "xqh_id": head_data["xqh_id"], + "zyfx_id": head_data["zyfx_id"], + "njdm_id": head_data["njdm_id"], + "bh_id": head_data["bh_id"], + "xbm": head_data["xbm"], + "xslbdm": head_data["xslbdm"], + "ccdm": head_data["ccdm"], + "xsbj": head_data["xsbj"], + "kklxdm": head_data[f"bkk{block}_kklxdm"], + "kch_id": jkch_res["tmpList"][0]["kch_id"], + "kkbk": head_data["kkbk"], + "rwlx": head_data["rwlx"], + "zyh_id": head_data["zyh_id"], + } + bkk_res = self.sess.post( + url_bkk, + headers=self.headers, + data=bkk_data, + cookies=self.cookies, + timeout=self.timeout, + ) + jbkk_res = bkk_res.json() + if block != 3 and (len(jkch_res["tmpList"]) != len(jbkk_res)): + return {"code": 999, "msg": "板块课编号及长度错误"} + temp_list = jkch_res["tmpList"] + block_list = jbkk_res + for i in range(len(temp_list)): + temp_list[i].update(block_list[i]) + + result = { + "count": len(temp_list), + "courses": [ + { + "course_id": j["kch_id"], + "class_id": j.get("jxb_id"), + "do_id": j.get("do_jxb_id"), + "title": j.get("kcmc"), + "teacher_id": (re.findall(r"(.*?\d+)/", j.get("jsxx")))[0], + "teacher": (re.findall(r"/(.*?)/", j.get("jsxx")))[0], + "credit": float(j.get("xf"), 0), + "kklxdm": head_data[f"bkk{block}_kklxdm"], + "capacity": int(i.get("jxbrl", 0)), + "selected_number": int(i.get("yxzrs", 0)), + "place": self.get_place(i.get("jxdd")), + "time": self.get_course_time(i.get("sksj")), + } + for j in temp_list + ], + } + return {"code": 1000, "msg": "获取板块课信息成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "获取板块课信息超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": f"获取板块课信息时未记录的错误：{str(e)}"} + + def select_course( + self, + sid: str, + course_id: str, + do_id: str, + kklxdm: str, + year: int, + term: int, + ): + """选课""" + try: + url_select = urljoin( + self.base_url, "xsxk/zzxkyzb_xkBcZyZzxkYzb.html?gnmkdm=N253512" + ) + term = term**2 * 3 + select_data = { + "jxb_ids": do_id, + "kch_id": course_id, + # 'rwlx': '3', + # 'rlkz': '0', + # 'rlzlkz': '1', + # 'sxbj': '1', + # 'xxkbj': '0', + # 'cxbj': '0', + "qz": "0", + # 'xkkz_id': '9B247F4EFD6291B9E055000000000001', + "xkxnm": str(year), + "xkxqm": str(term), + "njdm_id": str(sid[0:2]), + "zyh_id": str(sid[2:6]), + "kklxdm": str(kklxdm), + # 'xklc': '1', + } + req_select = self.sess.post( + url_select, + headers=self.headers, + data=select_data, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_select.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_select.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + result = req_select.json() + return {"code": 1000, "msg": "选课成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "选课超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": f"选课时未记录的错误：{str(e)}"} + + def cancel_course(self, do_id: str, course_id: str, year: int, term: int): + """取消选课""" + try: + url_cancel = urljoin( + self.base_url, "xsxk/zzxkyzb_tuikBcZzxkYzb.html?gnmkdm=N253512" + ) + term = term**2 * 3 + cancel_data = { + "jxb_ids": do_id, + "kch_id": course_id, + "xkxnm": str(year), + "xkxqm": str(term), + } + req_cancel = self.sess.post( + url_cancel, + headers=self.headers, + data=cancel_data, + cookies=self.cookies, + timeout=self.timeout, + ) + if req_cancel.status_code != 200: + return {"code": 2333, "msg": "教务系统挂了"} + doc = pq(req_cancel.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + result = {"status": re.findall(r"(\d+)", req_cancel.text)[0]} + return {"code": 1000, "msg": "退课成功", "data": result} + except exceptions.Timeout: + return {"code": 1003, "msg": "选课超时"} + except ( + exceptions.RequestException, + json.decoder.JSONDecodeError, + AttributeError, + ): + traceback.print_exc() + return {"code": 2333, "msg": "请重试，若多次失败可能是系统错误维护或需更新接口"} + except Exception as e: + traceback.print_exc() + return {"code": 999, "msg": f"选课时未记录的错误：{str(e)}"} + + # ============= utils ================= + + def get_gpa(self): + """获取GPA""" + url = urljoin( + self.base_url, + "xsxy/xsxyqk_cxXsxyqkIndex.html?gnmkdm=N105515&layout=default", + ) + req_gpa = self.sess.get( + url, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + ) + doc = pq(req_gpa.text) + if doc("h5").text() == "用户登录": + return {"code": 1006, "msg": "未登录或已过期，请重新登录"} + allc_str = [allc.text() for allc in doc("font[size='2px']").items()] + try: + gpa = float(allc_str[2]) + return gpa + except Exception: + return "init" + + def get_course_category(self, type, item): + """根据课程号获取类别""" + if type not in self.detail_category_type: + return item.get("KCLBMC") + if not item.get("KCH"): + return None + url = urljoin(self.base_url, f"jxjhgl/common_cxKcJbxx.html?id={item['KCH']}") + req_category = self.sess.get( + url, + headers=self.headers, + cookies=self.cookies, + timeout=self.timeout, + ) + doc = pq(req_category.text) + ths = doc("th") + try: + data_list = [(th.text).strip() for th in ths] + return data_list[6] + except: + return None + + @classmethod + def encrypt_password(cls, pwd, n, e): + """对密码base64编码""" + message = str(pwd).encode() + rsa_n = binascii.b2a_hex(binascii.a2b_base64(n)) + rsa_e = binascii.b2a_hex(binascii.a2b_base64(e)) + key = rsa.PublicKey(int(rsa_n, 16), int(rsa_e, 16)) + encropy_pwd = rsa.encrypt(message, key) + result = binascii.b2a_base64(encropy_pwd) + return result + + @classmethod + def parse_int(cls, digits): + if not digits: + return None + if not digits.isdigit(): + return digits + return int(digits) + + @classmethod + def align_floats(cls, floats): + if not floats: + return None + if floats == "无": + return "0.0" + return format(float(floats), ".1f") + + @classmethod + def display_course_time(cls, sessions): + if not sessions: + return None + args = re.findall(r"(\d+)", sessions) + start_time = cls.raspisanie[int(args[0]) + 1][0] + end_time = cls.raspisanie[int(args[0]) + 1][1] + return f"{start_time}~{end_time}" + + @classmethod + def list_sessions(cls, sessions): + if not sessions: + return None + args = re.findall(r"(\d+)", sessions) + return [n for n in range(int(args[0]), int(args[1]) + 1)] + + @classmethod + def list_weeks(cls, weeks): + """返回课程所含周列表""" + if not weeks: + return None + args = re.findall(r"[^,]+", weeks) + week_list = [] + for item in args: + if "-" in item: + weeks_pair = re.findall(r"(\d+)", item) + if len(weeks_pair) != 2: + continue + if "单" in item: + for i in range(int(weeks_pair[0]), int(weeks_pair[1]) + 1): + if i % 2 == 1: + week_list.append(i) + elif "双" in item: + for i in range(int(weeks_pair[0]), int(weeks_pair[1]) + 1): + if i % 2 == 0: + week_list.append(i) + else: + for i in range(int(weeks_pair[0]), int(weeks_pair[1]) + 1): + week_list.append(i) + else: + week_num = re.findall(r"(\d+)", item) + if len(week_num) == 1: + week_list.append(int(week_num[0])) + return week_list + + @classmethod + def get_academia_statistics(cls, display_statistics): + display_statistics = "".join(display_statistics.split()) + gpa_list = re.findall(r"([0-9]{1,}[.][0-9]*)", display_statistics) + if len(gpa_list) == 0 or not cls.is_number(gpa_list[0]): + gpa = None + else: + gpa = float(gpa_list[0]) + plan_list = re.findall( + r"计划总课程(\d+)门通过(\d+)门?.*未通过(\d+)门?.*未修(\d+)?.*在读(\d+)门?.*计划外?.*通过(\d+)门?.*未通过(\d+)门", + display_statistics, + ) + if len(plan_list) == 0 or len(plan_list[0]) < 7: + return {"gpa": gpa} + plan_list = plan_list[0] + return { + "gpa": gpa, # 平均学分绩点GPA + "planed_courses": { + "total": int(plan_list[0]), # 计划内总课程数 + "passed": int(plan_list[1]), # 计划内已过课程数 + "failed": int(plan_list[2]), # 计划内未过课程数 + "missed": int(plan_list[3]), # 计划内未修课程数 + "in": int(plan_list[4]), # 计划内在读课程数 + }, + "unplaned_courses": { + "passed": int(plan_list[5]), # 计划外已过课程数 + "failed": int(plan_list[6]), # 计划外未过课程数 + }, + } + + @classmethod + def get_academia_type_statistics(cls, content: str): + finder = re.findall( + r"\"(.*) .*要求学分.*:([0-9]{1,}[.][0-9]*|0| ).*获得学分.*:([0-9]{1,}[.][0-9]*|0| ).*未获得学分.*:([0-9]{1,}[.][0-9]*|0| )[\s\S]*?", + content, + ) + finder_list = list({}.fromkeys(finder).keys()) + academia_list = [ + list(i) + for i in finder_list + if i[0] != "" # 类型名称不为空 + and len(i[0]) <= 20 # 避免正则到首部过长类型名称 + and "span" not in i[-1] # 避免正则到尾部过长类型名称 + and i[0] not in cls.ignore_type # 忽略的类型名称 + ] + result = { + i[0]: { + "id": i[-1], + "credits": { + "required": i[1] if cls.is_number(i[1]) and i[1] != "0" else None, + "earned": i[2] if cls.is_number(i[2]) and i[2] != "0" else None, + "missed": i[3] if cls.is_number(i[3]) and i[3] != "0" else None, + }, + } + for i in academia_list + } + return result + + @classmethod + def get_display_term(cls, sid, year, term): + """ + 计算培养方案具体学期转化成中文 + note: 留级和当兵等情况会不准确 + """ + if (sid and year and term) is None: + return None + grade = int(sid[0:2]) + year = int(year[2:4]) + term = int(term) + dict = { + grade: "大一上" if term == 1 else "大一下", + grade + 1: "大二上" if term == 1 else "大二下", + grade + 2: "大三上" if term == 1 else "大三下", + grade + 3: "大四上" if term == 1 else "大四下", + } + return dict.get(year) + + @classmethod + def split_merge_display(cls, schedule): + """ + 拆分同周同天同课程不同时段数据合并的问题 + """ + repetIndex = [] + count = 0 + for items in schedule["courses"]: + for index in range(len(schedule["courses"])): + if (schedule["courses"]).index(items) == count: # 如果对比到自己就忽略 + continue + elif ( + items["course_id"] + == schedule["courses"][index]["course_id"] # 同周同天同课程 + and items["weekday"] == schedule["courses"][index]["weekday"] + and items["weeks"] == schedule["courses"][index]["weeks"] + ): + repetIndex.append(index) # 满足条件记录索引 + count += 1 # 记录当前对比课程的索引 + if len(repetIndex) % 2 != 0: # 暂时考虑一天两个时段上同一门课，不满足条件不进行修改 + return schedule + for r in range(0, len(repetIndex), 2): # 索引数组两两成对，故步进2循环 + fir = repetIndex[r] + sec = repetIndex[r + 1] + if len(re.findall(r"(\d+)", schedule["courses"][fir]["sessions"])) == 4: + schedule["courses"][fir]["sessions"] = ( + re.findall(r"(\d+)", schedule["courses"][fir]["sessions"])[0] + + "-" + + re.findall(r"(\d+)", schedule["courses"][fir]["sessions"])[1] + + "节" + ) + schedule["courses"][fir]["list_sessions"] = cls.list_sessions( + schedule["courses"][fir]["sessions"] + ) + schedule["courses"][fir]["time"] = cls.display_course_time( + schedule["courses"][fir]["sessions"] + ) + + schedule["courses"][sec]["sessions"] = ( + re.findall(r"(\d+)", schedule["courses"][sec]["sessions"])[2] + + "-" + + re.findall(r"(\d+)", schedule["courses"][sec]["sessions"])[3] + + "节" + ) + schedule["courses"][sec]["list_sessions"] = cls.list_sessions( + schedule["courses"][sec]["sessions"] + ) + schedule["courses"][sec]["time"] = cls.display_course_time( + schedule["courses"][sec]["sessions"] + ) + return schedule + + @classmethod + def split_notifications(cls, item): + if not item.get("xxnr"): + return {"type": None, "content": None} + content_list = re.findall(r"(.*):(.*)", item["xxnr"]) + if len(content_list) == 0: + return {"type": None, "content": item["xxnr"]} + return {"type": content_list[0][0], "content": content_list[0][1]} + + @classmethod + def get_place(cls, place): + return place.split("
")[0] if "
" in place else place + + @classmethod + def get_course_time(cls, time): + return "、".join(time.split("
")) if "
" in time else time + + @classmethod + def is_number(cls, s): + if s == "": + return False + try: + float(s) + return True + except ValueError: + pass + try: + for i in s: + unicodedata.numeric(i) + return True + except (TypeError, ValueError): + pass + return False + + +if __name__ == "__main__": + from pprint import pprint + import json + import base64 + import sys + import os + + base_url = "https://www.nianbroken.top" # 教务系统URL + sid = "NianBroken" # 学号 + password = "NianBroken" # 密码 + lgn_cookies = ( + { + # "insert_cookie": "", + # "route": "", + "JSESSIONID": "" + } + if False + else None + ) # cookies登录，调整成True使用cookies登录，反之使用密码登录 + test_year = 2022 # 查询学年 + test_term = 2 # 查询学期（1-上|2-下） + + # 初始化 + lgn = Client(lgn_cookies if lgn_cookies is not None else {}, base_url=base_url) + # 判断是否需要使用cookies登录 + if lgn_cookies is None: + # 登录 + pre_login = lgn.login(sid, password) + # 判断登录结果 + if pre_login["code"] == 1001: + # 需要验证码 + pre_dict = pre_login["data"] + with open(os.path.abspath("temp.json"), mode="w", encoding="utf-8") as f: + f.write(json.dumps(pre_dict)) + with open(os.path.abspath("kaptcha.png"), "wb") as pic: + pic.write(base64.b64decode(pre_dict["kaptcha_pic"])) + kaptcha = input("输入验证码：") + result = lgn.login_with_kaptcha( + pre_dict["sid"], + pre_dict["csrf_token"], + pre_dict["cookies"], + pre_dict["password"], + pre_dict["modulus"], + pre_dict["exponent"], + kaptcha, + ) + if result["code"] != 1000: + pprint(result) + sys.exit() + lgn_cookies = lgn.cookies + elif pre_login["code"] == 1000: + # 不需要验证码，直接登录 + lgn_cookies = lgn.cookies + else: + # 出错 + pprint(pre_login) + sys.exit() + + # 下面是各个函数调用，想调用哪个，取消注释即可 + """ 获取个人信息 """ + result = lgn.get_info() + + """ 获取成绩单PDF """ + # result = lgn.get_academia_pdf() + # if result["code"] == 1000: + # with open(os.path.abspath("grade.pdf"), "wb") as pdf: + # pdf.write(result["data"]) + # result = "已保存到本地" + + """ 获取学业情况 """ + # result = lgn.get_academia() + + """ 获取GPA """ + # result = lgn.get_gpa() + + """ 获取课程表 """ + # result = lgn.get_schedule(test_year, test_term) + + """ 获取成绩 """ + # result = lgn.get_grade(test_year, test_term) + + # 输出结果 + pprint(result)