do not send for relogin

.gitignore

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

main.py

@@ -1,199 +1,35 @@
-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 serverchan_sdk import sc_send as _sc_send
 
-# 配置日志
-logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
-)
-logger = logging.getLogger(__name__)
+from utils import URP, LessonsException, ReloginException, logger, sc_send
 
-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)  # 日志输出到文件
+try:
+    from rich import print
+    from rich.markdown import Markdown
+
+    RICH = True
+except ImportError:
+    print("Some function in console may disabled due to no rich.")
+    RICH = False
 
 
-class LessonsException(Exception):
-    """自定义异常类"""
-    pass
-class ReloginException(LessonsException):
-    """用于处理需要重新登录的异常"""
-    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:
-
+class Lessons(URP):
     def __init__(self):
-        self.session = requests.session()
+        super().__init__()
+
+        self.env_check()
+
         self.term: Optional[str] = None
         self.fajhh: Optional[str] = None
-
-        # 加载环境变量
-        dotenv.load_dotenv()
-
-        # 检查必需的环境变量
-        required_keys = [
-            "uname",
-            "password",
-            "recap_username",
-            "recap_password",
-            "FILE",
-        ]
-        for key in required_keys:
-            if not environ.get(key):
-                raise LessonsException(f"请在环境变量中设置{key}")
-
-        self.base = environ.get("base", "http://jwstudent.lnu.edu.cn")
         self.interval_1 = int(environ.get("INTERVAL_1", 2))  # 请求间隔，默认为2秒
         self.interval_2 = int(environ.get("INTERVAL_2", 10))  # 请求间隔，默认为10秒
 
-    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 ReloginException("有人登录了您的账号！")
-
     def get_base_info(self):
         res = self.session.get(f"{self.base}/student/courseSelect/gotoSelect/index")
         res.raise_for_status()
@@ -211,19 +47,19 @@ class Lessons:
         html = res.text
         # 使用正则表达式替代 BeautifulSoup 来查找选中的学期选项
         # 由于 HTML 结构特殊，selected 在单独行上，需要向前查找对应的 option
-        lines = html.split('\n')
+        lines = html.split("\n")
         for i, line in enumerate(lines):
-            if 'selected' in line.strip():
+            if "selected" in line.strip():
                 # 向前查找包含 option value 的行
-                for j in range(i-1, max(0, i-10), -1):
-                    if 'option value=' in lines[j]:
+                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
-        
+
         if not self.term:
             raise LessonsException("未找到学期信息")
 
@@ -276,9 +112,9 @@ class Lessons:
         response = self._retry_request(lambda: self.session.post(url, data=params))
         with open("response.json", "w", encoding="utf-8") as f:
             f.write(response.text)
-        data:dict = response.json()
-        
-        cls:list[dict] = data.get("rwfalist", [])
+        data: dict = response.json()
+
+        cls: list[dict] = data.get("rwfalist", [])
         if not cls:
             logger.error(f"课程 {cl[2]} 的课程信息为空: {data}")
             return None
@@ -295,8 +131,8 @@ class Lessons:
                         desp=f"课程 {cl[2]} 的课程名与查询信息不匹配: {item['kcm']} != {cl[2]}",
                     )
                     return None
-    
-        kyl:dict[str,str] = data["kylMap"]
+
+        kyl: dict[str, str] = data["kylMap"]
         if len(kyl) == 0:
             logger.error(f"课程 {cl[2]} 的余量信息为空: {kyl}")
             return
@@ -388,19 +224,6 @@ class Lessons:
         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:
@@ -472,7 +295,9 @@ class Lessons:
                                     raise e
                                 except Exception as e:
                                     errs.appendleft(time())
-                                    logger.error(f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}")
+                                    logger.error(
+                                        f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}"
+                                    )
                                 finally:
                                     sleep(self.interval_1)  # 避免请求过快导致服务器拒绝
                             elif left == -1:
@@ -517,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.auto_spider()
+    # les = Lessons()
+    # les.auto_spider()
+    gra = Grade()
+    gra.auto_check()

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()