format

main.py

@@ -1,11 +1,5 @@
-
-
-from utils import LessonsException, ReloginException
-from utils import sc_send,URP,logger
-
 import re
 from collections import deque
-
 from os import environ
 from pathlib import Path
 from time import sleep, time
@@ -13,18 +7,28 @@ from typing import Callable, List, Optional
 
 import pandas as pd
 
+from utils import URP, LessonsException, ReloginException, logger, sc_send
+
+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 Lessons(URP):
     def __init__(self):
         super().__init__()
-        
+
         self.env_check()
-        
+
         self.term: Optional[str] = None
-        self.fajhh: 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秒
-    
 
     def get_base_info(self):
         res = self.session.get(f"{self.base}/student/courseSelect/gotoSelect/index")
@@ -43,19 +47,19 @@ class Lessons(URP):
         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):
+                for j in range(i - 1, max(0, i - 10), -1):
-                    if 'option value=' in lines[j]:
+                    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("未找到学期信息")
 
@@ -108,9 +112,9 @@ class Lessons(URP):
         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()
+        data: dict = response.json()
-        
+
-        cls:list[dict] = data.get("rwfalist", [])
+        cls: list[dict] = data.get("rwfalist", [])
         if not cls:
             logger.error(f"课程 {cl[2]} 的课程信息为空: {data}")
             return None
@@ -127,8 +131,8 @@ class Lessons(URP):
                         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
@@ -220,7 +224,6 @@ class Lessons(URP):
         logger.warning(f"选课 {cl[2]} 结果查询超时，可能未成功选课")
         return False
 
-    
     def auto_spider(self):
         """自动选课主程序"""
         try:
@@ -292,7 +295,9 @@ class Lessons(URP):
                                     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:
@@ -336,6 +341,7 @@ class Lessons(URP):
             sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
             raise e
 
+
 class Grade(URP):
     def __init__(self):
         super().__init__()
@@ -347,52 +353,53 @@ class Grade(URP):
             "recap_password",
         ]
         self.env_check(required_keys)
-        self.interval_2 = int(environ.get("interval_2",3600))
+        self.interval_2 = int(environ.get("interval_2", 3600))
-    
+
-    def query(self) -> tuple[dict[str,dict[str,str]],set[str]]:
+    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 = 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)
+        match = re.search(
+            f"/student/integratedQuery/scoreQuery/.+/thisTermScores/data", html
+        )
         if match:
-            url = self.base+match.group(0)
+            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))
+        res = self._retry_request(lambda: self.session.get(url))
         # print(res.text)
-        res_json = res.json()   
+        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)}")
+            sc_send("成绩查询异常", f"课程数发生变化 {self.total}!={len(l)}")
-        ret = {x["courseName"]: x for x in l if x["avgcj"].strip()!=""}
+        ret = {x["courseName"]: x for x in l if x["avgcj"].strip() != ""}
-        return ret,set(ret.keys())
+        return ret, set(ret.keys())
-    
+
     @staticmethod
-    def format(x:dict[str,str]):
+    def format(x: dict[str, str]):
         return f"|{x['courseName']}|{x['courseScore']}|{x['maxcj']}|{x['avgcj']}|"
-        
+
     def auto_check(self):
         self.login()
-        # self.session.cookies.update({"student.urpSoft.cn":"aaapnXQb62LApgwx7lkFz","UqZBpD3n3iXPAw1X9DmYiUaISMkd8YhMUen0":"v1IraGSUs3hnH"})
+
-        
         grades = set()
         self.query()
-        
+
-        assert isinstance(self.total,int)
+        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]]
+                cls: dict[str, dict[str, str]]
-                new:set[str]
+                new: set[str]
                 if new != grades:
                     delta_names = new - grades
                     delta = [cls[x] for x in delta_names]
@@ -400,48 +407,45 @@ class Grade(URP):
                     t.append("新成绩")
                     t.append("|学科|成绩|最高分|平均分|")
                     t.append("|-|-|-|-|")
-                    t.extend(map(self.format,delta))
+                    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.extend(map(self.format, cls.values()))
-                    
+
                     t = "\n".join(t)
-                    sc_send("成绩发布",t)
+                    sc_send("成绩发布", t)
-                    try:
+                    if RICH:
-                        from rich.markdown import Markdown
-                        from rich import print
                         print(Markdown(t))
-                    except ImportError:
+                    else:
-                        print("Cannot import rich, show markdown directly.")
                         print(t)
                     grades = new
-                    if err > 0: err-=1
+                    if err > 0:
+                        err -= 1
             except ReloginException as e:
                 logger.info("Relogin")
-                sc_send("成绩监控","重新登录")
+                sc_send("成绩监控", "重新登录")
                 self.login()
             except Exception as e:
                 logger.error(f"Failed to update due to {e}")
-                err+=1
+                err += 1
                 if err >= 5:
                     logger.error("Try to relogin")
-                    sc_send("成绩监控","多次失败，尝试重新登录")
+                    sc_send("成绩监控", "多次失败，尝试重新登录")
                     self.login()
-            
+
             logger.info(f"Next query will start after {self.interval_2}s")
             sleep(self.interval_2)
         logger.info("Normal terminated due to all grades is out.")
-        sc_send("成绩监控","所有成绩均已公布")
+        sc_send("成绩监控", "所有成绩均已公布")
-                
+
-            
 
 if __name__ == "__main__":
     les = Lessons()
     les.auto_spider()
     # gra = Grade()
-    # gra.auto_check()
+    # gra.auto_check()

utils.py

@@ -1,29 +1,37 @@
-from serverchan_sdk import sc_send as _sc_send
-from datetime import datetime
-import logging
-from pathlib import Path
-from os import environ
-from dotenv import load_dotenv
-import requests
 import base64
 import hashlib
 import json
+import logging
 import re
-from typing import NoReturn,Optional,Iterable
+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 未设置，无法发送通知")
@@ -43,16 +51,17 @@ def log_init():
     file.setLevel(logging.INFO)
     logger.addHandler(file)  # 日志输出到文件
 
+
 class URP:
-    def __init__(self,base=None):
+    def __init__(self, base=None):
         self.session = requests.session()
         if base is None:
-            self.base = environ.get("base","http://jwstudent.lnu.edu.cn")
+            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:
+    def env_check(required_keys: Optional[Iterable] = None) -> None | NoReturn:
         # 检查必需的环境变量
         if required_keys is None:
             required_keys = [
@@ -65,7 +74,7 @@ class URP:
         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 = "请求失败"
@@ -123,7 +132,7 @@ class URP:
         md5_part2 = hashlib.md5(string.encode()).hexdigest().lower()
         final_result = md5_part1 + "*" + md5_part2
         return final_result
-    
+
     def _login(self):
         """登录模块"""
         username = environ.get("uname")
@@ -195,7 +204,6 @@ class URP:
         if not flag:
             raise LessonsException("登录失败，无法获取token")
 
-    
     @staticmethod
     def _judge_logout(response: requests.Response):
         """检查账号是否在其他地方被登录"""
@@ -203,6 +211,5 @@ class URP:
             raise ReloginException("有人登录了您的账号！")
 
 
-
 load_dotenv(override=True)
-log_init()
+log_init()