From d78e9902f2b44b0fd1f2234800ceab6d803542b1 Mon Sep 17 00:00:00 2001
From: flt6 <1404262047@qq.com>
Date: Tue, 8 Jul 2025 23:59:54 +0800
Subject: [PATCH] split code into URP class
---
.gitignore | 3 +-
main.py | 249 +++++++++--------------------------------------------
utils.py | 208 ++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 249 insertions(+), 211 deletions(-)
create mode 100644 utils.py
diff --git a/.gitignore b/.gitignore
index 5184213..ef828c8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
.env
logs
-lessons*
\ No newline at end of file
+lessons*
+__pycache__
\ No newline at end of file
diff --git a/main.py b/main.py
index 7e9d6d3..3251a5b 100644
--- a/main.py
+++ b/main.py
@@ -1,199 +1,30 @@
-import base64
-import hashlib
-import json
-import logging
+
+
+from utils import LessonsException, ReloginException
+from utils import sc_send,URP,logger
+
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__)
-
-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 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:
-
- def __init__(self,skip_check:bool =False):
- self.session = requests.session()
+class Lessons(URP):
+ def __init__(self):
+ 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",
- ]
- 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.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 _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"发生错误！(.+)", 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")
@@ -389,19 +220,7 @@ 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:
@@ -517,23 +336,31 @@ class Lessons:
sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
raise e
-class Grade:
+class Grade(URP):
def __init__(self):
+ super().__init__()
self.total = None
- self.lesson = Lessons(skip_check=False)
+ required_keys = [
+ "uname",
+ "password",
+ "recap_username",
+ "recap_password",
+ ]
+ self.env_check(required_keys)
+ self.interval_2 = int(environ.get("interval_2",3600))
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))
+ 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.lesson.base+match.group(0)
+ url = self.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))
+ # 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"]
@@ -549,8 +376,8 @@ class Grade:
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"})
+ self.login()
+ # self.session.cookies.update({"student.urpSoft.cn":"aaapnXQb62LApgwx7lkFz","UqZBpD3n3iXPAw1X9DmYiUaISMkd8YhMUen0":"v1IraGSUs3hnH"})
grades = set()
self.query()
@@ -562,6 +389,7 @@ class Grade:
while len(grades) < self.total:
try:
+ logger.info("Querying")
cls, new = self.query()
cls:dict[str,dict[str,str]]
new:set[str]
@@ -596,18 +424,19 @@ class Grade:
except ReloginException as e:
logger.info("Relogin")
sc_send("成绩监控","重新登录")
- self.lesson.login()
+ 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.lesson.login()
-
+ self.login()
- sleep(self.lesson.interval_2)
-
+ 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("成绩监控","所有成绩均已公布")
diff --git a/utils.py b/utils.py
new file mode 100644
index 0000000..837faa6
--- /dev/null
+++ b/utils.py
@@ -0,0 +1,208 @@
+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 re
+from typing import NoReturn,Optional,Iterable
+
+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()
+ 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"发生错误！(.+)", 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 "errorCode=concurrentSessionExpired" in response.url:
+ raise ReloginException("有人登录了您的账号！")
+
+
+
+load_dotenv(override=True)
+log_init()
\ No newline at end of file