import httpx
import re
from pydantic import BaseModel, Field
from typing import Optional, List, Dict
from .utils import (
encrypt_password,
get_salt_and_execution_from_html,
process_student_courses_from_data,
)
from .exception import (
LoginError,
InvalidCredentialsError,
AccountNotActivatedError,
GetUserConfigError,
AcademicLoginError,
ExamScheduleError,
NotLoggedInError,
ClassScheduleError,
FetchScoresError,
HTMLExtractionError,
)
[文档]
class LoginClient(BaseModel):
"""用于与广东工业大学统一身份认证平台进行交互的客户端。
该类封装了与统一身份认证平台交互所需的参数和方法，
包括初始化登录数据和执行登录操作。
"""
SERVER: str = r"https://authserver.gdut.edu.cn"
AUTH_URL: str = (
f"{SERVER}/authserver/login" # ?service={the_url_encoded_services_you_want_to_go}
)
CHECK_CAPTCHA_URL: str = (
f"{SERVER}/authserver/checkNeedCaptcha.htl" # ?username={number}&_={timestamp}
)
GET_LANGUAGE_URL: str = f"{SERVER}/authserver/common/getLanguageTypes.htl"
GET_SERVER_INFO_URL: str = f"{SERVER}/authserver/tenant/info"
GET_USER_CONF_URL: str = f"{SERVER}/personalInfo/common/getUserConf"
salt: Optional[str] = Field(None, description="登录时用于加密密码的哈希盐")
execution: Optional[str] = Field(None, description="登录操作的执行内容")
client: httpx.Client = Field(
default_factory=lambda: httpx.Client(follow_redirects=True),
description="用于发送 HTTP 请求的客户端实例",
)
class Config:
arbitrary_types_allowed = True
def __init__(self):
"""初始化 LoginClient 类的实例。"""
super().__init__()
def __init_login_data(self) -> None:
"""初始化登录所需的数据：salt 和 execution。
发送 GET 请求到统一身份认证平台的登录页面，并解析 HTML
以获取 salt 和 execution 值。
Raises:
HTMLExtractionError: 如果无法从登录页面提取到哈希盐和 execution。
"""
self.client = httpx.Client(
headers={
"User-Agent": "gduter-py/0.1.0-alpha (Windows NT 10.0; Win64; x64)"
},
follow_redirects=True,
)
response = self.client.get(self.AUTH_URL)
data = get_salt_and_execution_from_html(response.text)
if len(data) != 2:
raise HTMLExtractionError(f"无法提取哈希盐和execution：{data}")
self.salt = data[0]
self.execution = data[1]
[文档]
def login(self, username: str, password: str) -> None:
"""进行统一身份验证平台的登录操作。
Args:
username: 登录的用户名。
password: 用户的明文密码。
Raises:
InvalidCredentialsError: 用户名或密码错误。
AccountNotActivatedError: 账户未激活。
GetUserConfigError: 获取用户配置失败。
LoginError: 其他登录失败情况。
"""
self.__init_login_data()
encrypted_password = encrypt_password(password, self.salt.encode())
params = {
"captcha": "",
"cllt": "userNameLogin",
"dllt": "generalLogin",
"execution": self.execution,
"lt": "",
"password": encrypted_password,
"username": username,
"_eventId": "submit",
}
response = self.client.post(self.AUTH_URL, params=params)
if response.status_code in [200, 301, 302]:
user_config = self.client.post(self.GET_USER_CONF_URL)
if user_config.status_code != 200:
raise GetUserConfigError(f"获取用户配置失败！{user_config.text}")
return
else:
pattern = (
r'<span id="showErrorTip" class="form-error"><span>(.*?)</span></span>'
)
error = re.search(pattern, response.text)
if error:
error_msg = error.group(1)
if "该账号非常用账号或用户名密码有误" in error_msg:
raise InvalidCredentialsError("提供的用户名可能不存在或密码错误！")
elif "用户名密码错误或者账号未激活" in error_msg:
raise AccountNotActivatedError("用户名密码错误或者账号未激活！")
else:
raise LoginError(f"登录失败！{error_msg}")
else:
raise LoginError(
f"登录失败，状态码: {response.status_code}，内容: {response.text}"
)
[文档]
class AcademicAffairsOfficeClient(BaseModel):
"""用于与广东工业大学教务系统进行交互的客户端。
依赖于已登录的 `LoginClient` 实例，提供获取课程表、
考试安排、成绩等功能。
"""
HOST: str = r"https://jxfw.gdut.edu.cn"
AUTH_URL: str = (
r"https://authserver.gdut.edu.cn/authserver/login?service=https%3A%2F%2Fjxfw.gdut.edu.cn%2Fnew%2FssoLogin"
)
GET_INFO_URL: str = HOST + r"/xjkpxx!xjkpxx.action"
GET_COURSES_SCHEDULE_URL: str = (
HOST + r"/xsgrkbcx!xsAllKbList.action"
) # ?xnxqdm={year + 01/02}
GET_EXAM_SCHEDULE_URL: str = HOST + r"/xsksap!getDataList.action"
GET_EXAM_SCORES_URL: str = HOST + r"/xskccjxx!getDataList.action"
client: httpx.Client = Field(
default_factory=lambda: httpx.Client(follow_redirects=True),
description="用于发送 HTTP 请求的客户端实例",
)
status: bool = Field(False, description="标记是否成功登陆教务处")
class Config:
arbitrary_types_allowed = True
def __init__(self, login_instance: LoginClient):
"""初始化 AcademicAffairsOfficeClient 类的实例。
Args:
login_instance: 已登录的 LoginClient 实例。
"""
super().__init__(login_instance=login_instance)
self.client = login_instance.client
[文档]
def login(self) -> None:
"""登录教务系统。
使用共享的客户端发送 GET 请求到教务系统登录 URL。
Raises:
AcademicLoginError: 登录教务系统失败。
"""
response = self.client.get(self.AUTH_URL)
if (
'<a href="/new/logout">注销</a>' in response.text
and response.status_code == 200
):
self.status = True
return
else:
raise AcademicLoginError(
f"登录教务处失败了！状态码：{response.status_code}"
)
[文档]
def get_course_schedule(self, period: int) -> List[Dict]:
"""获取指定学期代码的课程表。
Args:
period: 学年学期代码，例如 202401。
Returns:
包含课程信息的字典列表。
Raises:
NotLoggedInError: 账号未登录。
ClassScheduleError: 无法获取课程表或课表为空。
"""
header = {
"Host": "jxfw.gdut.edu.cn",
"Referer": "https://jxfw.gdut.edu.cn/xsgrkbcx!getXsgrbkList.action",
}
if not self.status:
raise NotLoggedInError("账号未登录！")
response = self.client.get(
self.GET_COURSES_SCHEDULE_URL + f"?xnxqdm={period}", headers=header
)
if "使用统一认证中心登录" in response.text:
self.status = False
raise NotLoggedInError("请进行重新登陆！")
if response.status_code == 200:
courses = process_student_courses_from_data(response.text)
else:
raise ClassScheduleError(f"无法获取课程表！状态码：{response.status_code}")
return courses
[文档]
def get_academic_scores(
self,
period: int,
plan: str = "",
page: int = 1,
rows: int = 60,
sort: str = "xnxqdm",
order: str = "asc",
) -> dict:
"""获取学生指定学期的学业成绩。
Args:
period: 学年学期代码，例如 202401。
plan: 计划类型代码，默认为空字符串。
page: 页码，默认为 1。
rows: 每页显示的行数，默认为 60。
sort: 排序字段，默认为 "xnxqdm"。
order: 排序顺序，默认为 "asc"。
Returns:
包含成绩信息的字典。
Raises:
NotLoggedInError: 账号未登录。
FetchScoresError: 获取课程成绩失败。
"""
if not self.status:
raise NotLoggedInError("账号未登录！")
payload = {
"xnxqdm": str(period),
"jhlxdm": plan,
"page": str(page),
"rows": str(rows),
"sort": sort,
"order": order,
}
header = {
"Host": "jxfw.gdut.edu.cn",
"Origin": "https://jxfw.gdut.edu.cn",
"Referer": "https://jxfw.gdut.edu.cn/xskccjxx!xskccjList.action?firstquery=1",
}
response = self.client.post(
self.GET_EXAM_SCORES_URL, data=payload, headers=header
)
if "使用统一认证中心登录" in response.text:
self.status = False
raise NotLoggedInError("请进行重新登陆！")
if response.status_code == 200:
return response.json()
else:
raise FetchScoresError(f"获取课程成绩失败！状态码：{response.status_code}")
[文档]
def get_exam_schedule(
self,
period: int,
page: int = 1,
rows: int = 60,
sort: str = "zc,xq,jcdm2",
order: str = "asc",
) -> Dict:
"""获取指定学期代码的考试安排。
Args:
period: 学年学期代码，例如 202401。
page: 页码，默认为 1。
rows: 每页显示的行数，默认为 60。
sort: 排序字段，默认为 "zc,xq,jcdm2"。
order: 排序顺序，默认为 "asc"。
Returns:
包含考试安排信息的字典。
Raises:
NotLoggedInError: 账号未登录。
ExamScheduleError: 获取考试信息失败。
"""
if not self.status:
raise NotLoggedInError("账号未登录！")
payload = {
"xnxqdm": str(period),
"page": str(page),
"rows": str(rows),
"sort": sort,
"order": order,
}
header = {
"Host": "jxfw.gdut.edu.cn",
"Origin": "https://jxfw.gdut.edu.cn",
"Referer": "https://jxfw.gdut.edu.cn/xsksap!ksapList.action",
}
response = self.client.post(
self.GET_EXAM_SCHEDULE_URL, data=payload, headers=header
)
if "使用统一认证中心登录" in response.text:
self.status = False
raise NotLoggedInError("请进行重新登陆！")
if response.status_code == 200:
return response.json()
else:
raise ExamScheduleError(f"获取考试信息失败！{response.text}")