From 32b34736593be87ac339cb03deaada307f909cbd Mon Sep 17 00:00:00 2001 From: flt6 <1404262047@qq.com> Date: Tue, 8 Jul 2025 23:39:18 +0800 Subject: [PATCH] add grade --- main.py | 104 +++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 100 insertions(+), 4 deletions(-) diff --git a/main.py b/main.py index 6b94844..7e9d6d3 100644 --- a/main.py +++ b/main.py @@ -50,7 +50,7 @@ def sc_send(title: str, desp: str): class Lessons: - def __init__(self): + def __init__(self,skip_check:bool =False): self.session = requests.session() self.term: Optional[str] = None self.fajhh: Optional[str] = None @@ -66,9 +66,10 @@ class Lessons: "recap_password", "FILE", ] - for key in required_keys: - if not environ.get(key): - raise LessonsException(f"请在环境变量中设置{key}") + if not skip_check: + 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秒 @@ -516,7 +517,102 @@ class Lessons: sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}") raise e +class Grade: + def __init__(self): + self.total = None + self.lesson = Lessons(skip_check=False) + + def query(self) -> tuple[dict[str,dict[str,str]],set[str]]: + url = f"{self.lesson.base}/student/integratedQuery/scoreQuery/thisTermScores/index" + res = self.lesson._retry_request(lambda:self.lesson.session.get(url)) + res.raise_for_status() + html = res.text + match = re.search(f"/student/integratedQuery/scoreQuery/.+/thisTermScores/data",html) + if match: + url = self.lesson.base+match.group(0) + else: + raise RuntimeError("Cannot find url") + # url = f"{self.lesson.base}/student/integratedQuery/scoreQuery/U6I5OXib09/thisTermScores/data" + res = self.lesson._retry_request(lambda :self.lesson.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.lesson.login() + # self.lesson.session.cookies.update({"student.urpSoft.cn":"aaapnXQb62LApgwx7lkFz","UqZBpD3n3iXPAw1X9DmYiUaISMkd8YhMUen0":"v1IraGSUs3hnH"}) + + grades = set() + self.query() + + assert isinstance(self.total,int) + assert self.total > 0 + + err = 0 + + while len(grades) < self.total: + try: + 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) + try: + from rich.markdown import Markdown + from rich import print + print(Markdown(t)) + except ImportError: + print("Cannot import rich, show markdown directly.") + print(t) + grades = new + if err > 0: err-=1 + except ReloginException as e: + logger.info("Relogin") + sc_send("成绩监控","重新登录") + self.lesson.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.lesson.login() + + + sleep(self.lesson.interval_2) + + + if __name__ == "__main__": les = Lessons() les.auto_spider() + # gra = Grade() + # gra.auto_check() \ No newline at end of file