Compare commits
4 Commits
d78e9902f2
...
main
Author | SHA1 | Date | |
---|---|---|---|
0bf300d95b | |||
a33ff2c117 | |||
d56d2a3267 | |||
eebe92bab0 |
63
main.py
63
main.py
@ -1,11 +1,5 @@
|
|||||||
|
|
||||||
|
|
||||||
from utils import LessonsException, ReloginException
|
|
||||||
from utils import sc_send,URP,logger
|
|
||||||
|
|
||||||
import re
|
import re
|
||||||
from collections import deque
|
from collections import deque
|
||||||
|
|
||||||
from os import environ
|
from os import environ
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from time import sleep, time
|
from time import sleep, time
|
||||||
@ -13,6 +7,17 @@ from typing import Callable, List, Optional
|
|||||||
|
|
||||||
import pandas as pd
|
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):
|
class Lessons(URP):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
@ -25,7 +30,6 @@ class Lessons(URP):
|
|||||||
self.interval_1 = int(environ.get("INTERVAL_1", 2)) # 请求间隔,默认为2秒
|
self.interval_1 = int(environ.get("INTERVAL_1", 2)) # 请求间隔,默认为2秒
|
||||||
self.interval_2 = int(environ.get("INTERVAL_2", 10)) # 请求间隔,默认为10秒
|
self.interval_2 = int(environ.get("INTERVAL_2", 10)) # 请求间隔,默认为10秒
|
||||||
|
|
||||||
|
|
||||||
def get_base_info(self):
|
def get_base_info(self):
|
||||||
res = self.session.get(f"{self.base}/student/courseSelect/gotoSelect/index")
|
res = self.session.get(f"{self.base}/student/courseSelect/gotoSelect/index")
|
||||||
res.raise_for_status()
|
res.raise_for_status()
|
||||||
@ -43,12 +47,12 @@ class Lessons(URP):
|
|||||||
html = res.text
|
html = res.text
|
||||||
# 使用正则表达式替代 BeautifulSoup 来查找选中的学期选项
|
# 使用正则表达式替代 BeautifulSoup 来查找选中的学期选项
|
||||||
# 由于 HTML 结构特殊,selected 在单独行上,需要向前查找对应的 option
|
# 由于 HTML 结构特殊,selected 在单独行上,需要向前查找对应的 option
|
||||||
lines = html.split('\n')
|
lines = html.split("\n")
|
||||||
for i, line in enumerate(lines):
|
for i, line in enumerate(lines):
|
||||||
if 'selected' in line.strip():
|
if "selected" in line.strip():
|
||||||
# 向前查找包含 option value 的行
|
# 向前查找包含 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])
|
value_match = re.search(r'value="([^"]*)"', lines[j])
|
||||||
if value_match:
|
if value_match:
|
||||||
self.term = str(value_match.group(1))
|
self.term = str(value_match.group(1))
|
||||||
@ -220,7 +224,6 @@ class Lessons(URP):
|
|||||||
logger.warning(f"选课 {cl[2]} 结果查询超时,可能未成功选课")
|
logger.warning(f"选课 {cl[2]} 结果查询超时,可能未成功选课")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def auto_spider(self):
|
def auto_spider(self):
|
||||||
"""自动选课主程序"""
|
"""自动选课主程序"""
|
||||||
try:
|
try:
|
||||||
@ -292,7 +295,9 @@ class Lessons(URP):
|
|||||||
raise e
|
raise e
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
errs.appendleft(time())
|
errs.appendleft(time())
|
||||||
logger.error(f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}")
|
logger.error(
|
||||||
|
f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}"
|
||||||
|
)
|
||||||
finally:
|
finally:
|
||||||
sleep(self.interval_1) # 避免请求过快导致服务器拒绝
|
sleep(self.interval_1) # 避免请求过快导致服务器拒绝
|
||||||
elif left == -1:
|
elif left == -1:
|
||||||
@ -336,6 +341,7 @@ class Lessons(URP):
|
|||||||
sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
|
sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
|
|
||||||
class Grade(URP):
|
class Grade(URP):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
@ -347,6 +353,7 @@ class Grade(URP):
|
|||||||
"recap_password",
|
"recap_password",
|
||||||
]
|
]
|
||||||
self.env_check(required_keys)
|
self.env_check(required_keys)
|
||||||
|
self.interval_1 = int(environ.get("interval_1", 5))
|
||||||
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]]:
|
||||||
@ -354,7 +361,9 @@ class Grade(URP):
|
|||||||
res = self._retry_request(lambda: self.session.get(url))
|
res = self._retry_request(lambda: self.session.get(url))
|
||||||
res.raise_for_status()
|
res.raise_for_status()
|
||||||
html = res.text
|
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:
|
if match:
|
||||||
url = self.base + match.group(0)
|
url = self.base + match.group(0)
|
||||||
else:
|
else:
|
||||||
@ -377,7 +386,6 @@ class Grade(URP):
|
|||||||
|
|
||||||
def auto_check(self):
|
def auto_check(self):
|
||||||
self.login()
|
self.login()
|
||||||
# self.session.cookies.update({"student.urpSoft.cn":"aaapnXQb62LApgwx7lkFz","UqZBpD3n3iXPAw1X9DmYiUaISMkd8YhMUen0":"v1IraGSUs3hnH"})
|
|
||||||
|
|
||||||
grades = set()
|
grades = set()
|
||||||
self.query()
|
self.query()
|
||||||
@ -412,18 +420,18 @@ class Grade(URP):
|
|||||||
|
|
||||||
t = "\n".join(t)
|
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))
|
print(Markdown(t))
|
||||||
except ImportError:
|
else:
|
||||||
print("Cannot import rich, show markdown directly.")
|
|
||||||
print(t)
|
print(t)
|
||||||
grades = new
|
grades = new
|
||||||
if err > 0: err-=1
|
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:
|
except ReloginException as e:
|
||||||
logger.info("Relogin")
|
logger.info("Relogin")
|
||||||
sc_send("成绩监控","重新登录")
|
# sc_send("成绩监控", "重新登录")
|
||||||
self.login()
|
self.login()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Failed to update due to {e}")
|
logger.error(f"Failed to update due to {e}")
|
||||||
@ -432,16 +440,15 @@ class Grade(URP):
|
|||||||
logger.error("Try to relogin")
|
logger.error("Try to relogin")
|
||||||
sc_send("成绩监控", "多次失败,尝试重新登录")
|
sc_send("成绩监控", "多次失败,尝试重新登录")
|
||||||
self.login()
|
self.login()
|
||||||
|
finally:
|
||||||
|
sleep(self.interval_1)
|
||||||
|
|
||||||
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.")
|
logger.info("Normal terminated due to all grades is out.")
|
||||||
sc_send("成绩监控", "所有成绩均已公布")
|
sc_send("成绩监控", "所有成绩均已公布")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
les = Lessons()
|
# les = Lessons()
|
||||||
les.auto_spider()
|
# les.auto_spider()
|
||||||
# gra = Grade()
|
gra = Grade()
|
||||||
# gra.auto_check()
|
gra.auto_check()
|
||||||
|
30
utils.py
30
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 base64
|
||||||
import hashlib
|
import hashlib
|
||||||
import json
|
import json
|
||||||
|
import logging
|
||||||
import re
|
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):
|
class LessonsException(Exception):
|
||||||
"""自定义异常类"""
|
"""自定义异常类"""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class ReloginException(LessonsException):
|
class ReloginException(LessonsException):
|
||||||
"""用于处理需要重新登录的异常"""
|
"""用于处理需要重新登录的异常"""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
# 配置日志
|
# 配置日志
|
||||||
logging.basicConfig(
|
logging.basicConfig(
|
||||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||||
)
|
)
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def sc_send(title: str, desp: str):
|
def sc_send(title: str, desp: str):
|
||||||
if not environ.get("SC_KEY"):
|
if not environ.get("SC_KEY"):
|
||||||
logger.error("SC_KEY 未设置,无法发送通知")
|
logger.error("SC_KEY 未设置,无法发送通知")
|
||||||
@ -43,6 +51,7 @@ def log_init():
|
|||||||
file.setLevel(logging.INFO)
|
file.setLevel(logging.INFO)
|
||||||
logger.addHandler(file) # 日志输出到文件
|
logger.addHandler(file) # 日志输出到文件
|
||||||
|
|
||||||
|
|
||||||
class URP:
|
class URP:
|
||||||
def __init__(self, base=None):
|
def __init__(self, base=None):
|
||||||
self.session = requests.session()
|
self.session = requests.session()
|
||||||
@ -74,6 +83,7 @@ class URP:
|
|||||||
for attempt in range(1, max_retries + 1):
|
for attempt in range(1, max_retries + 1):
|
||||||
try:
|
try:
|
||||||
response = func()
|
response = func()
|
||||||
|
print(response.url)
|
||||||
URP._judge_logout(response)
|
URP._judge_logout(response)
|
||||||
return response
|
return response
|
||||||
except (
|
except (
|
||||||
@ -195,14 +205,12 @@ class URP:
|
|||||||
if not flag:
|
if not flag:
|
||||||
raise LessonsException("登录失败,无法获取token")
|
raise LessonsException("登录失败,无法获取token")
|
||||||
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _judge_logout(response: requests.Response):
|
def _judge_logout(response: requests.Response):
|
||||||
"""检查账号是否在其他地方被登录"""
|
"""检查账号是否在其他地方被登录"""
|
||||||
if "errorCode=concurrentSessionExpired" in response.url:
|
if "/login" in response.url:
|
||||||
raise ReloginException("有人登录了您的账号!")
|
raise ReloginException("有人登录了您的账号!")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
load_dotenv(override=True)
|
load_dotenv(override=True)
|
||||||
log_init()
|
log_init()
|
Reference in New Issue
Block a user