do not send for relogin

.env_demo

@@ -0,0 +1,8 @@
+uname=20240000000
+password=123456
+recap_username=this_is_recap_unam
+recap_password=123456
+SC_KEY=this_is_sc_key
+FILE = "lessons.xlsx"
+INTERVAL_1=2
+INTERVAL_2=10

.gitignore

@@ -0,0 +1,4 @@
+.env
+logs
+lessons*
+__pycache__

README.md

@@ -0,0 +1,144 @@
+# 自动选课脚本
+
+一个用于某URP系统的自动监控课程余量并进行选课的工具。当目标课程有余量时，程序会自动尝试选课，并推送选课结果。
+
+## 功能特点
+
+- 🔄 自动监控课程余量
+- 📱 微信消息推送通知
+- 🔁 失败自动重试
+- 📊 支持多种格式的课程文件
+- 📝 详细的日志记录
+
+## 使用前准备
+
+### 1. 准备课程文件
+
+创建一个Excel文件（建议命名 `lessons.xlsx`），**至少**包含以下三列：
+
+| 课程号 | 课序号 | 课程名 |
+|--------|--------|--------|
+| 080302001 | 01 | 高等数学A |
+| 080302002 | 02 | 线性代数 |
+
+**说明：**
+- **课程名**：课程的完整名称，不允许增加空格等，建议直接复制选课手册行。
+- 允许出现除上述内容以外的项，所以如果是辽大的可以直接复制选课手册对应行到文件中。
+- 允许存在**同一课程多个课序号**，将按照自上而下的顺序选，任何一个选课成功就认为该课程选课成功。
+- 具体格式参考`demo.xlsx`和`demo.csv`
+- 支持的文件格式：`.xlsx`、`.xls`、`.csv`、`.json`
+
+### 2. 配置环境变量
+
+创建 `.env` 文件，包含以下配置：
+
+```env
+# 教务系统登录信息
+uname=你的学号
+password=你的密码
+
+# 验证码识别服务账号（必需）
+recap_username=验证码识别服务用户名
+recap_password=验证码识别服务密码
+
+# 课程文件路径
+# 改成自己起的名字，注意拓展名
+FILE=lessons.xlsx
+
+# 微信推送服务（可选）
+SC_KEY=你的ServerChan密钥
+
+# 教务系统地址（可选，默认为辽宁大学）
+base=http://jwstudent.lnu.edu.cn
+
+# 查询间隔
+# 每次选课请求间隔
+INTERVAL_1=2
+# 完成一轮选课后的间隔
+INTERVAL_2=10
+```
+
+你可以查看`.env_demo`查看样例。
+
+### 3. 获取验证码识别服务
+
+目前使用 http://www.fdyscloud.com.cn/ 提供的验证码识别，0.003 元/张。
+这是一个付费平台，与作者无关。
+
+操作方法：
+1. 打开并注册账号（密码建议随机，因为要明文保存）
+2. 充值合适余额（1元估计就够用到大学毕业了）
+3. 在.env文件中`recap_username`和`recap_password`输入你的账号密码。
+
+如果您不希望付费，可以修改代码中的`Lessons.recapture`自行实现本地验证码OCR。
+
+
+### 4. 设置推送（可选）
+
+如需接收选课结果的微信推送：
+1. 访问 [Server酱](https://sct.ftqq.com/)或者[Server$^3$酱](https://sct.ftqq.com/)
+2. 注册并获取SendKey
+3. 将密钥填入 `.env` 文件的 `SC_KEY` 字段
+   
+**注意，Server$^3$酱会产生费用**
+
+### 5. 安装依赖
+```bash
+pip install -r requirements.txt
+```
+
+## 使用方法
+
+1. 确保已完成上述准备工作
+2. 将课程文件和 `.env` 文件放在脚本同目录下
+3. 运行脚本：
+   ```
+   python main.py
+   ```
+
+## 运行流程
+
+1. **登录验证**: 自动登录教务系统
+2. **读取课程**: 从课程文件中读取要选择的课程信息
+3. **监控余量**: 循环检查各课程的剩余名额
+4. **自动选课**: 发现有余量时立即尝试选课
+5. **结果通知**: 通过推送选课结果
+6. **持续监控**: 未选上的课程继续监控
+
+## 注意事项
+
+⚠️ **重要提醒**：
+- 请在选课开放时间内使用，在未开放时会发生未知错误。
+- 确保网络连接稳定
+- 验证码识别服会产生费用
+- 建议在选课高峰期前测试配置是否正确
+- 程序会自动处理网络异常和重新登录
+
+## 日志文件
+
+程序运行时会在 `logs` 目录下生成详细的日志文件，文件名格式为 `lessons_YYYYMMDD_HHMMSS.log`，可用于排查问题。
+
+## 文件结构
+
+```
+项目目录/
+├── main.py          # 主程序
+├── .env             # 环境变量配置文件
+├── lessons.xlsx     # 课程信息文件
+├── logs/            # 日志文件目录
+└── README.md        # 说明文档
+```
+
+## 常见问题
+
+**Q: 程序提示"请在环境变量中设置xxx"**
+A: 检查 `.env` 文件是否存在且包含所有必需的配置项
+
+**Q: 一直提示登录失败**
+A: 检查用户名密码是否正确，以及验证码识别服务是否可用
+
+**Q: 找不到课程信息**
+A: 检查课程文件中的课程号和课序号是否正确
+
+**Q: 没有收到推送**
+A: 检查 ServerChan 配置是否正确，或查看日志中的错误信息

demo.csv

@@ -0,0 +1,3 @@
+课程号, 课序号,课程名
+2310031,01,体育（二）
+2310031,02,体育（二）

main.py

@@ -1,196 +1,34 @@
-import base64
-import hashlib
-import json
-import logging
 import re
 from collections import deque
-from datetime import datetime
 from os import environ
 from pathlib import Path
 from time import sleep, time
 from typing import Callable, List, Optional
 
-import dotenv
 import pandas as pd
-import requests
-from bs4 import BeautifulSoup
-from serverchan_sdk import sc_send as _sc_send
 
-# 配置日志
+from utils import URP, LessonsException, ReloginException, logger, sc_send
-logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
-)
-logger = logging.getLogger(__name__)
 
-Path("logs").mkdir(exist_ok=True)  # 确保日志目录存在
+try:
-now = datetime.now().strftime("%Y%m%d_%H%M%S")
+    from rich import print
-file = logging.FileHandler(f"logs/lessons_{now}.log", mode="w", encoding="utf-8")
+    from rich.markdown import Markdown
-file.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
+
-file.setLevel(logging.INFO)
+    RICH = True
-logger.addHandler(file)  # 日志输出到文件
+except ImportError:
+    print("Some function in console may disabled due to no rich.")
+    RICH = False
 
 
-class LessonsException(Exception):
+class Lessons(URP):
-    """自定义异常类"""
-
-    pass
-
-
-def sc_send(title: str, desp: str):
-    if not environ.get("SC_KEY"):
-        logger.error("SC_KEY 未设置，无法发送通知")
-        return
-    sc_key = environ.get("SC_KEY")
-    try:
-        _sc_send(sc_key, title, desp, options={"tags": "自动选课"})
-    except Exception as e:
-        logger.error(f"发送失败: {e}")
-
-
-class Lessons:
-
     def __init__(self):
-        self.session = requests.session()
+        super().__init__()
+
+        self.env_check()
+
         self.term: Optional[str] = None
         self.fajhh: Optional[str] = None
-
+        self.interval_1 = int(environ.get("INTERVAL_1", 2))  # 请求间隔，默认为2秒
-        # 加载环境变量
+        self.interval_2 = int(environ.get("INTERVAL_2", 10))  # 请求间隔，默认为10秒
-        dotenv.load_dotenv()
-
-        # 检查必需的环境变量
-        required_keys = [
-            "uname",
-            "password",
-            "recap_username",
-            "recap_password",
-            "FILE",
-            "SC_KEY",
-        ]
-        for key in required_keys:
-            if not environ.get(key):
-                raise LessonsException(f"请在环境变量中设置{key}")
-
-        self.base = environ.get("base", "http://jwstudent.lnu.edu.cn")
-
-    def _retry_request(
-        self, func, max_retries: int = 10, error_msg: str = "请求失败"
-    ) -> requests.Response:
-        """通用的请求重试方法"""
-        for attempt in range(1, max_retries + 1):
-            try:
-                response = func()
-                self.judge_logout(response)
-                return response
-            except (
-                requests.ConnectionError,
-                requests.HTTPError,
-                requests.Timeout,
-            ) as e:
-                logger.warning(f"{error_msg}！{type(e).__name__}: {e}")
-                if attempt < max_retries:
-                    logger.info(f"第{attempt}次重试")
-                else:
-                    raise LessonsException(f"{error_msg}！请检查网络连接！")
-
-        # 这行不会被执行，但为了类型检查
-        raise LessonsException(f"{error_msg}！重试次数耗尽")
-
-    @staticmethod
-    def recapture(b64: str) -> str:
-        """验证码识别"""
-        recap_username = environ.get("recap_username")
-        recap_password = environ.get("recap_password")
-
-        if not recap_username or not recap_password:
-            raise LessonsException("验证码识别服务配置不完整")
-
-        data = {
-            "username": recap_username,
-            "password": recap_password,
-            "ID": "04897896",
-            "b64": b64,
-            "version": "3.1.1",
-        }
-
-        try:
-            response = requests.post(
-                "http://www.fdyscloud.com.cn/tuling/predict", json=data, timeout=10
-            )
-            response.raise_for_status()
-            result = response.json()
-            return result["data"]["result"]
-        except (requests.RequestException, KeyError, json.JSONDecodeError) as e:
-            raise LessonsException(f"验证码识别失败: {e}")
-
-    @staticmethod
-    def pwd_md5(string: str) -> str:
-        md5_part1 = hashlib.md5((string + "{Urp602019}").encode()).hexdigest().lower()
-        md5_part2 = hashlib.md5(string.encode()).hexdigest().lower()
-        final_result = md5_part1 + "*" + md5_part2
-        return final_result
-
-    def _login(self):
-        """登录模块"""
-        username = environ.get("uname")
-        password = environ.get("password")
-
-        if not username or not password:
-            raise LessonsException("用户名或密码未设置")
-
-        try:
-            # 获取登录页面的token
-            req = self.session.get("http://jwstudent.lnu.edu.cn/login")
-            req.raise_for_status()
-            html = req.text
-            match = re.search(r'name="tokenValue" value="(.+?)">', html)
-            if not match:
-                raise LessonsException("未找到 tokenValue")
-            token_value = match.group(1)
-
-            # 获取验证码
-            req = self.session.get(f"{self.base}/img/captcha.jpg")
-            req.raise_for_status()
-            im = req.content
-            b64 = base64.b64encode(im).decode("utf-8")
-            captcha_code = self.recapture(b64=b64)
-
-            logger.info(f"验证码识别结果: {captcha_code}")
-
-            hashed_password = self.pwd_md5(password)
-
-            # 模拟请求的 payload
-            payload = {
-                "j_username": username,
-                "j_password": hashed_password,
-                "j_captcha": captcha_code,
-                "tokenValue": token_value,
-            }
-
-            # 发送 POST 请求
-            url = f"{self.base}/j_spring_security_check"
-            headers = {
-                "User-Agent": "Mozilla/5.0",
-                "Content-Type": "application/x-www-form-urlencoded",
-            }
-            response = self.session.post(url, data=payload, headers=headers)
-
-            if "发生错误" in response.text:
-                err = re.search(r"<strong>发生错误！</strong>(.+)", response.text)
-                if err:
-                    error_message = err.group(1).strip()
-                    raise LessonsException(f"登录失败: {error_message}")
-                raise LessonsException("登录失败")
-
-            logger.info("登录成功")
-            return True
-
-        except requests.RequestException as e:
-            raise LessonsException(f"登录过程中网络错误: {e}")
-
-    def judge_logout(self, response: requests.Response):
-        """检查账号是否在其他地方被登录"""
-        if response.url == f"{self.base}/login?errorCode=concurrentSessionExpired":
-            raise LessonsException("有人登录了您的账号！")
 
     def get_base_info(self):
         res = self.session.get(f"{self.base}/student/courseSelect/gotoSelect/index")
@@ -198,7 +36,7 @@ class Lessons:
         html = res.text
         match = re.search(r"fajhh=(\d+)", html)
         if not match:
-            print(html)
+            # print(html)
             raise LessonsException("未找到培养方案编号")
         self.fajhh = match.group(1)
 
@@ -207,13 +45,25 @@ class Lessons:
         )
         res.raise_for_status()
         html = res.text
-        bs = BeautifulSoup(html, "html.parser")
+        # 使用正则表达式替代 BeautifulSoup 来查找选中的学期选项
-        match = bs.select_one("select#jhxn option[selected]")
+        # 由于 HTML 结构特殊，selected 在单独行上，需要向前查找对应的 option
-        if not match:
+        lines = html.split("\n")
-            raise LessonsException("未找到学期信息")
+        for i, line in enumerate(lines):
-        self.term = str(match.get("value"))
+            if "selected" in line.strip():
+                # 向前查找包含 option value 的行
+                for j in range(i - 1, max(0, i - 10), -1):
+                    if "option value=" in lines[j]:
+                        value_match = re.search(r'value="([^"]*)"', lines[j])
+                        if value_match:
+                            self.term = str(value_match.group(1))
+                            break
+                if self.term:
+                    break
 
-        print(self.fajhh, self.term)
+        if not self.term:
+            raise LessonsException("未找到学期信息")
+
+        # print(self.fajhh, self.term)
 
     def read_lessons(self) -> List[tuple[str, str, str]]:
         classes = []
@@ -260,20 +110,42 @@ class Lessons:
             "jc": 0,
         }
         response = self._retry_request(lambda: self.session.post(url, data=params))
-        data = response.json()["kylMap"]
+        with open("response.json", "w", encoding="utf-8") as f:
-        if len(data) == 0:
+            f.write(response.text)
-            logger.error(f"课程 {cl[2]} 的余量信息为空: {data}")
+        data: dict = response.json()
-            return
 
+        cls: list[dict] = data.get("rwfalist", [])
+        if not cls:
+            logger.error(f"课程 {cl[2]} 的课程信息为空: {data}")
+            return None
+
+        for item in cls:
+            if item["classNum"] in cl[1]:
+                # print(item["classNum"],type(item["classNum"]))
+                if item["kcm"] != cl[2]:
+                    logger.critical(
+                        f"课程 {cl[2]} 的课程名与查询信息不匹配: {item['kcm']} != {cl[2]}"
+                    )
+                    sc_send(
+                        "选课异常",
+                        desp=f"课程 {cl[2]} 的课程名与查询信息不匹配: {item['kcm']} != {cl[2]}",
+                    )
+                    return None
+
+        kyl: dict[str, str] = data["kylMap"]
+        if len(kyl) == 0:
+            logger.error(f"课程 {cl[2]} 的余量信息为空: {kyl}")
+            return
         ret = []
         for kxh in cl[1]:
             key = f"{self.term}_{cl[0]}_{kxh}"
-            left = data.get(key, None)
+            left = kyl.get(key, None)
             if left is None:
                 logger.error(
-                    f"课程 {cl[2]} 的余量信息不存在: {key} not in {data.keys()}"
+                    f"课程 {cl[2]} 的余量信息不存在: {key} not in {kyl.keys()}"
                 )
                 ret.append(-1)
+                continue
             ret.append(int(left))
         return ret
 
@@ -319,7 +191,7 @@ class Lessons:
         html = response.text
         redisKey = re.search(r'var redisKey = "(.+)";', html)
         if not redisKey:
-            print(html)
+            # print(html)
             logger.error(f"选课 {cl[2]} 时未找到 redisKey")
             return False
         redisKey = redisKey.group(1)
@@ -345,26 +217,13 @@ class Lessons:
                 else:
                     logger.info(f"选课成功: {text}")
             else:
-                print(f"第{cnt}次查询中...")
+                logger.info(f"第{cnt}次查询中...")
                 cnt += 1
                 sleep(1)
 
         logger.warning(f"选课 {cl[2]} 结果查询超时，可能未成功选课")
         return False
 
-    def login(self):
-        logger.info("尝试登录")
-        flag = False
-        for i in range(10):
-            try:
-                if self._login():
-                    flag = True
-                    break
-            except Exception as e:
-                logger.error(f"登录失败: {e}")
-        if not flag:
-            raise LessonsException("登录失败，无法获取token")
-
     def auto_spider(self):
         """自动选课主程序"""
         try:
@@ -379,6 +238,10 @@ class Lessons:
                 if classes.get(id) is None:
                     classes[id] = (id, [kxh], name)
                 else:
+                    if classes[id][2] != name:
+                        raise LessonsException(
+                            f"课程 {name}_{kxh} 的名称不一致: {classes[id][2]} != {name}"
+                        )
                     classes[id][1].append(kxh)
 
             logger.info(f"读取课程信息，共有 {len(classes)} 门课程")
@@ -396,53 +259,76 @@ class Lessons:
                     sc_send("选课异常", desp="最近5次获取课程余量异常，尝试重新登录")
                     self.login()
                     master_err += 1
+                    errs.clear()
                 if master_err >= 3:
                     logger.error("反复发生重要异常，退出程序")
                     sc_send("选课异常", desp="反复发生重要异常，退出程序")
                     return
-                for lcl in classes.copy().values():
+                try:
-                    logger.info(f"检查课程 {lcl[2]} 余量")
+                    for lcl in classes.copy().values():
-                    try:
+                        logger.info(f"检查课程 {lcl[2]} 余量")
-                        lefts = self.get_left(lcl)
+                        try:
-                        if lefts is None:
+                            lefts = self.get_left(lcl)
-                            errs.appendleft(time())
+                            if lefts is None:
-                            logger.error(f"获取课程 {lcl[2]} 余量时返回异常")
-                            continue
-                    except Exception as e:
-                        errs.appendleft(time())
-                        logger.error(f"获取课程 {lcl[2]} 余量时发生错误: {e}")
-                        continue
-                    for i, left in enumerate(lefts):
-                        cl = (lcl[0], lcl[1][i], lcl[2])
-                        if left > 0:
-                            logger.info(
-                                f"课程 {cl[2]}_{cl[1]} 有余量: {left}，开始选课"
-                            )
-                            try:
-                                ret = self.select(cl)
-                                if ret:
-                                    suc.append(cl)
-                                    classes.pop(cl[0])
-                                    break
-                            except Exception as e:
                                 errs.appendleft(time())
-                                logger.error(f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}")
+                                logger.error(f"获取课程 {lcl[2]} 余量时返回异常")
-                            finally:
+                                continue
-                                sleep(2)  # 避免请求过快导致服务器拒绝
+                        except ReloginException as e:
-                        elif left == -1:
+                            raise e
-                            logger.error(f"课程 {cl[2]}_{cl[1]} 余量信息异常")
+                        except Exception as e:
-                        else:
+                            errs.appendleft(time())
-                            logger.info(f"课程 {cl[2]}_{cl[1]} 无余量")
+                            logger.error(f"获取课程 {lcl[2]} 余量时发生错误: {e}")
-                    sleep(2)
+                            continue
-                logger.info(
+                        for i, left in enumerate(lefts):
-                    f"当前还有{len(classes)}门课程未选上，分别为{','.join(mp[i] for i in classes.keys())}。等待10秒后继续检查"
+                            cl = (lcl[0], lcl[1][i], lcl[2])
-                )
+                            if left > 0:
-                if suc:
+                                logger.info(
-                    sc_send(
+                                    f"课程 {cl[2]}_{cl[1]} 有余量: {left}，开始选课"
-                        "选课成功",
+                                )
-                        desp=f"已成功选上课程: {', '.join(f'{i[2]}_{i[1]}' for i in suc)}",
+                                try:
+                                    ret = self.select(cl)
+                                    if ret:
+                                        suc.append(cl)
+                                        classes.pop(cl[0])
+                                        break
+                                except ReloginException as e:
+                                    raise e
+                                except Exception as e:
+                                    errs.appendleft(time())
+                                    logger.error(
+                                        f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}"
+                                    )
+                                finally:
+                                    sleep(self.interval_1)  # 避免请求过快导致服务器拒绝
+                            elif left == -1:
+                                logger.error(f"课程 {cl[2]}_{cl[1]} 余量信息异常")
+                            else:
+                                logger.info(f"课程 {cl[2]}_{cl[1]} 无余量")
+                        sleep(self.interval_1)
+                    logger.info(
+                        f"当前还有{len(classes)}门课程未选上，分别为{','.join(mp[i] for i in classes.keys())}。等待{self.interval_2}秒后继续检查"
                     )
-                sleep(10)  # 等待10秒后继续检查
+                    if suc:
+                        sc_send(
+                            "选课成功",
+                            desp=f"已成功选上课程: {', '.join(f'{i[2]}_{i[1]}' for i in suc)}",
+                        )
+                except ReloginException as e:
+                    logger.error(f"需要重新登录: {e}")
+                    sc_send("选课异常", desp=f"需要重新登录: {e}")
+                    self.login()
+                    continue
+                except LessonsException as e:
+                    logger.error(f"选课过程中发生错误: {e}")
+                    sc_send("选课异常", desp=f"选课过程中发生错误: {e}")
+                    errs.appendleft(time())
+                    continue
+                except Exception as e:
+                    logger.error(f"意外错误: {e}")
+                    sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
+                    errs.appendleft(time())
+                    continue
+                sleep(self.interval_2)  # 等待10秒后继续检查
 
             logger.info("自动选课完成")
 
@@ -456,6 +342,113 @@ class Lessons:
             raise e
 
 
+class Grade(URP):
+    def __init__(self):
+        super().__init__()
+        self.total = None
+        required_keys = [
+            "uname",
+            "password",
+            "recap_username",
+            "recap_password",
+        ]
+        self.env_check(required_keys)
+        self.interval_1 = int(environ.get("interval_1", 5))
+        self.interval_2 = int(environ.get("interval_2", 3600))
+
+    def query(self) -> tuple[dict[str, dict[str, str]], set[str]]:
+        url = f"{self.base}/student/integratedQuery/scoreQuery/thisTermScores/index"
+        res = self._retry_request(lambda: self.session.get(url))
+        res.raise_for_status()
+        html = res.text
+        match = re.search(
+            f"/student/integratedQuery/scoreQuery/.+/thisTermScores/data", html
+        )
+        if match:
+            url = self.base + match.group(0)
+        else:
+            raise RuntimeError("Cannot find url")
+        # url = f"{self.base}/student/integratedQuery/scoreQuery/U6I5OXib09/thisTermScores/data"
+        res = self._retry_request(lambda: self.session.get(url))
+        # print(res.text)
+        res_json = res.json()
+        l = res_json[0]["list"]
+        if self.total is None:
+            self.total = len(l)
+        elif self.total != len(l):
+            sc_send("成绩查询异常", f"课程数发生变化 {self.total}!={len(l)}")
+        ret = {x["courseName"]: x for x in l if x["avgcj"].strip() != ""}
+        return ret, set(ret.keys())
+
+    @staticmethod
+    def format(x: dict[str, str]):
+        return f"|{x['courseName']}|{x['courseScore']}|{x['maxcj']}|{x['avgcj']}|"
+
+    def auto_check(self):
+        self.login()
+
+        grades = set()
+        self.query()
+
+        assert isinstance(self.total, int)
+        assert self.total > 0
+
+        err = 0
+
+        while len(grades) < self.total:
+            try:
+                logger.info("Querying")
+                cls, new = self.query()
+                cls: dict[str, dict[str, str]]
+                new: set[str]
+                if new != grades:
+                    delta_names = new - grades
+                    delta = [cls[x] for x in delta_names]
+                    t = []
+                    t.append("新成绩")
+                    t.append("|学科|成绩|最高分|平均分|")
+                    t.append("|-|-|-|-|")
+                    t.extend(map(self.format, delta))
+
+                    logger.info("\n".join(t))
+
+                    t.append("---")
+                    t.append("所有成绩")
+                    t.append("|学科|成绩|最高分|平均分|")
+                    t.append("|-|-|-|-|")
+                    t.extend(map(self.format, cls.values()))
+
+                    t = "\n".join(t)
+                    sc_send("成绩发布", t)
+                    if RICH:
+                        print(Markdown(t))
+                    else:
+                        print(t)
+                    grades = new
+                    if err > 0:
+                        err -= 1
+                sleep(self.interval_2)
+                logger.info(f"Next query will start after {self.interval_2}s")
+            except ReloginException as e:
+                logger.info("Relogin")
+                # sc_send("成绩监控", "重新登录")
+                self.login()
+            except Exception as e:
+                logger.error(f"Failed to update due to {e}")
+                err += 1
+                if err >= 5:
+                    logger.error("Try to relogin")
+                    sc_send("成绩监控", "多次失败，尝试重新登录")
+                    self.login()
+            finally:
+                sleep(self.interval_1)
+
+        logger.info("Normal terminated due to all grades is out.")
+        sc_send("成绩监控", "所有成绩均已公布")
+
+
 if __name__ == "__main__":
-    les = Lessons()
+    # les = Lessons()
-    les.auto_spider()
+    # les.auto_spider()
+    gra = Grade()
+    gra.auto_check()

requirements.txt

@@ -0,0 +1,5 @@
+python-dotenv
+pandas
+requests
+serverchan-sdk
+openpyxl

utils.py

@@ -0,0 +1,216 @@
+import base64
+import hashlib
+import json
+import logging
+import re
+from datetime import datetime
+from os import environ
+from pathlib import Path
+from typing import Iterable, NoReturn, Optional
+
+import requests
+from dotenv import load_dotenv
+from serverchan_sdk import sc_send as _sc_send
+
+
+class LessonsException(Exception):
+    """自定义异常类"""
+
+    pass
+
+
+class ReloginException(LessonsException):
+    """用于处理需要重新登录的异常"""
+
+    pass
+
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+def sc_send(title: str, desp: str):
+    if not environ.get("SC_KEY"):
+        logger.error("SC_KEY 未设置，无法发送通知")
+        return
+    sc_key = environ.get("SC_KEY")
+    try:
+        _sc_send(sc_key, title, desp, options={"tags": "自动选课"})
+    except Exception as e:
+        logger.error(f"发送失败: {e}")
+
+
+def log_init():
+    Path("logs").mkdir(exist_ok=True)  # 确保日志目录存在
+    now = datetime.now().strftime("%Y%m%d_%H%M%S")
+    file = logging.FileHandler(f"logs/lessons_{now}.log", mode="w", encoding="utf-8")
+    file.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
+    file.setLevel(logging.INFO)
+    logger.addHandler(file)  # 日志输出到文件
+
+
+class URP:
+    def __init__(self, base=None):
+        self.session = requests.session()
+        if base is None:
+            self.base = environ.get("base", "http://jwstudent.lnu.edu.cn")
+        else:
+            self.base = base
+
+    @staticmethod
+    def env_check(required_keys: Optional[Iterable] = None) -> None | NoReturn:
+        # 检查必需的环境变量
+        if required_keys is None:
+            required_keys = [
+                "uname",
+                "password",
+                "recap_username",
+                "recap_password",
+                "FILE",
+            ]
+        for key in required_keys:
+            if not environ.get(key):
+                raise LessonsException(f"请在环境变量中设置{key}")
+
+    @staticmethod
+    def _retry_request(
+        func, max_retries: int = 10, error_msg: str = "请求失败"
+    ) -> requests.Response:
+        """通用的请求重试方法"""
+        for attempt in range(1, max_retries + 1):
+            try:
+                response = func()
+                print(response.url)
+                URP._judge_logout(response)
+                return response
+            except (
+                requests.ConnectionError,
+                requests.HTTPError,
+                requests.Timeout,
+            ) as e:
+                logger.warning(f"{error_msg}！{type(e).__name__}: {e}")
+                if attempt < max_retries:
+                    logger.info(f"第{attempt}次重试")
+                else:
+                    raise LessonsException(f"{error_msg}！请检查网络连接！")
+
+        # 这行不会被执行，但为了类型检查
+        raise LessonsException(f"{error_msg}！重试次数耗尽")
+
+    @staticmethod
+    def recapture(b64: str) -> str:
+        """验证码识别"""
+        recap_username = environ.get("recap_username")
+        recap_password = environ.get("recap_password")
+
+        if not recap_username or not recap_password:
+            raise LessonsException("验证码识别服务配置不完整")
+
+        data = {
+            "username": recap_username,
+            "password": recap_password,
+            "ID": "04897896",
+            "b64": b64,
+            "version": "3.1.1",
+        }
+
+        try:
+            response = requests.post(
+                "http://www.fdyscloud.com.cn/tuling/predict", json=data, timeout=10
+            )
+            response.raise_for_status()
+            result = response.json()
+            return result["data"]["result"]
+        except (requests.RequestException, KeyError, json.JSONDecodeError) as e:
+            raise LessonsException(f"验证码识别失败: {e}")
+
+    @staticmethod
+    def pwd_md5(string: str) -> str:
+        md5_part1 = hashlib.md5((string + "{Urp602019}").encode()).hexdigest().lower()
+        md5_part2 = hashlib.md5(string.encode()).hexdigest().lower()
+        final_result = md5_part1 + "*" + md5_part2
+        return final_result
+
+    def _login(self):
+        """登录模块"""
+        username = environ.get("uname")
+        password = environ.get("password")
+
+        if not username or not password:
+            raise LessonsException("用户名或密码未设置")
+
+        try:
+            # 获取登录页面的token
+            req = self.session.get("http://jwstudent.lnu.edu.cn/login")
+            req.raise_for_status()
+            html = req.text
+            match = re.search(r'name="tokenValue" value="(.+?)">', html)
+            if not match:
+                raise LessonsException("未找到 tokenValue")
+            token_value = match.group(1)
+
+            # 获取验证码
+            req = self.session.get(f"{self.base}/img/captcha.jpg")
+            req.raise_for_status()
+            im = req.content
+            b64 = base64.b64encode(im).decode("utf-8")
+            captcha_code = self.recapture(b64=b64)
+
+            logger.info(f"验证码识别结果: {captcha_code}")
+
+            hashed_password = self.pwd_md5(password)
+
+            # 模拟请求的 payload
+            payload = {
+                "j_username": username,
+                "j_password": hashed_password,
+                "j_captcha": captcha_code,
+                "tokenValue": token_value,
+            }
+
+            # 发送 POST 请求
+            url = f"{self.base}/j_spring_security_check"
+            headers = {
+                "User-Agent": "Mozilla/5.0",
+                "Content-Type": "application/x-www-form-urlencoded",
+            }
+            response = self.session.post(url, data=payload, headers=headers)
+
+            if "发生错误" in response.text:
+                err = re.search(r"<strong>发生错误！</strong>(.+)", response.text)
+                if err:
+                    error_message = err.group(1).strip()
+                    raise LessonsException(f"登录失败: {error_message}")
+                raise LessonsException("登录失败")
+
+            logger.info("登录成功")
+            return True
+
+        except requests.RequestException as e:
+            raise LessonsException(f"登录过程中网络错误: {e}")
+
+    def login(self):
+        logger.info("尝试登录")
+        flag = False
+        for i in range(10):
+            try:
+                if self._login():
+                    flag = True
+                    break
+            except Exception as e:
+                logger.error(f"登录失败: {e}")
+        if not flag:
+            raise LessonsException("登录失败，无法获取token")
+
+    @staticmethod
+    def _judge_logout(response: requests.Response):
+        """检查账号是否在其他地方被登录"""
+        if "/login" in response.url:
+            raise ReloginException("有人登录了您的账号！")
+
+
+load_dotenv(override=True)
+log_init()