diff --git a/docs/note.md b/docs/note.md
index cea0698..e17f817 100644
--- a/docs/note.md
+++ b/docs/note.md
@@ -1,5 +1,5 @@
# Note
- 一些逆向过程的笔记
+ 一些逆向过程的笔记
### 充值电费
ZZU 的电费充值系统对充值的请求体进行了加密,这本没有问题,但他居然选择的是国密算法,SM2 和 SM4。可能是国产化指标吧,唉😮💨。
diff --git a/docs/usage/web/network.md b/docs/usage/web/network.md
index f4e61d3..ab22731 100644
--- a/docs/usage/web/network.md
+++ b/docs/usage/web/network.md
@@ -124,7 +124,7 @@ with SelfServiceSystem("http://10.2.7.16:8080") as system:
devices = system.get_online_devices()
for device in devices:
- print(device.ip, device.mac, device.loginTime)
+ print(device.ip, device.mac, device.login_time)
```
### 踢设备下线
@@ -135,7 +135,7 @@ with SelfServiceSystem("http://10.2.7.16:8080") as system:
devices = system.get_online_devices()
if devices:
- system.kick_device(devices[0].sessionId)
+ system.kick_device(devices[0].session_id)
```
## 数据模型
@@ -146,6 +146,9 @@ with SelfServiceSystem("http://10.2.7.16:8080") as system:
- `AuthResult`:Portal 认证结果
- `OnlineDevice`:自助服务系统中的在线设备信息
+!!! note "字段命名"
+ `OnlineDevice` 使用 snake_case 字段名,例如 `login_time`、`session_id`。旧的 camelCase 属性仍可读取,但会触发 `DeprecationWarning`。
+
## 异步版本
异步接口位于 `zzupy.aio.web.network`:
diff --git a/zzupy/aio/app/auth.py b/zzupy/aio/app/auth.py
index adec049..2e9b2e5 100644
--- a/zzupy/aio/app/auth.py
+++ b/zzupy/aio/app/auth.py
@@ -122,7 +122,8 @@ def _validate_jwt(self, pre_set_token: bool = False) -> bool:
user_token_plain: dict = jwt.decode(
user_token, options={"verify_signature": False}
)
- expire_date = datetime.fromtimestamp(user_token_plain.get("exp"))
+ exp = float(user_token_plain["exp"])
+ expire_date = datetime.fromtimestamp(exp)
now = datetime.now()
time_to_expire = (expire_date - now).total_seconds()
@@ -246,6 +247,7 @@ async def _ensure_public_key(self) -> RSAPublicKey:
"""确保统一认证客户端已获取 RSA 公钥。"""
if self._cas._public_key is None:
self._cas._public_key = await self._cas._get_public_key()
+ assert self._cas._public_key is not None
return self._cas._public_key
def _attest_url(self, path: str) -> str:
@@ -583,6 +585,8 @@ async def login(self, force_login: bool = False) -> None:
if self._public_key is None:
self._public_key = await self._get_public_key()
+ assert self._public_key is not None
+
if self.mfa.state:
mfa_state_invalid = self.mfa.required and not self.mfa.verified
else:
diff --git a/zzupy/aio/web/network.py b/zzupy/aio/web/network.py
index 0b79f78..9cc5815 100644
--- a/zzupy/aio/web/network.py
+++ b/zzupy/aio/web/network.py
@@ -20,6 +20,7 @@
ParsingError,
ZZUError,
)
+from zzupy.logging import logger
from zzupy.model.network import AuthResult, OnlineDevice, PortalInfo
from zzupy.utils import (
extract_first_html_attr,
@@ -66,19 +67,22 @@ def _extract_user_ip(portal_url: str) -> str:
def _extract_auth_url(portal_url: str) -> str:
"""提取网页认证 URL"""
parsed = urllib.parse.urlparse(portal_url)
+ if not parsed.scheme or not parsed.netloc:
+ raise ParsingError("无法从Portal URL获取认证服务器地址")
return f"{parsed.scheme}://{parsed.netloc}"
async def _get_portal_server_url(client: httpx2.AsyncClient, auth_url: str) -> str:
"""获取 Portal 服务器 URL"""
DEFAULT_HTTP_PORT = 801
DEFAULT_HTTPS_PORT = 802
+ hostname = urllib.parse.urlparse(auth_url).hostname
+ if hostname is None:
+ raise ParsingError("无法从认证 URL 获取 Portal 主机名")
try:
response = await client.get(f"{auth_url}/a41.js")
js_params = _parse_js_config(response.text)
- hostname = urllib.parse.urlparse(auth_url).hostname
-
if js_params.get("enableHttps") == 0:
port = js_params.get("epHTTPPort", DEFAULT_HTTP_PORT)
return f"http://{hostname}:{port}"
@@ -86,9 +90,8 @@ async def _get_portal_server_url(client: httpx2.AsyncClient, auth_url: str) -> s
port = js_params.get("enHTTPSPort", DEFAULT_HTTPS_PORT)
return f"https://{hostname}:{port}"
- except Exception:
- # 降级到默认配置
- hostname = urllib.parse.urlparse(auth_url).hostname
+ except (httpx2.RequestError, ValueError) as exc:
+ logger.debug("获取 Portal 服务器配置失败,降级到默认配置: {}", exc)
return f"http://{hostname}:{DEFAULT_HTTP_PORT}"
def _parse_js_config(js_content: str) -> dict[str, int]:
diff --git a/zzupy/app/auth.py b/zzupy/app/auth.py
index a4f892e..354d51e 100644
--- a/zzupy/app/auth.py
+++ b/zzupy/app/auth.py
@@ -120,7 +120,8 @@ def _validate_jwt(self, pre_set_token: bool = False) -> bool:
user_token_plain: dict = jwt.decode(
user_token, options={"verify_signature": False}
)
- expire_date = datetime.fromtimestamp(user_token_plain.get("exp"))
+ exp = float(user_token_plain["exp"])
+ expire_date = datetime.fromtimestamp(exp)
now = datetime.now()
time_to_expire = (expire_date - now).total_seconds()
@@ -245,6 +246,7 @@ def _ensure_public_key(self) -> RSAPublicKey:
"""确保统一认证客户端已获取 RSA 公钥。"""
if self._cas._public_key is None:
self._cas._public_key = self._cas._get_public_key()
+ assert self._cas._public_key is not None
return self._cas._public_key
def _attest_url(self, path: str) -> str:
@@ -477,7 +479,7 @@ def request_sms_code(self) -> None:
def send_sms(self) -> None:
"""[`request_sms_code()`][zzupy.app.auth.CASClient.MFAClient.request_sms_code] 的别名。"""
- return self.request_sms_code()
+ self.request_sms_code()
def verify_sms_code(self, code: str) -> str:
"""校验 MFA 短信验证码。
@@ -584,6 +586,8 @@ def login(self, force_login: bool = False) -> None:
if self._public_key is None:
self._public_key = self._get_public_key()
+ assert self._public_key is not None
+
if self.mfa.state:
mfa_state_invalid = self.mfa.required and not self.mfa.verified
else:
diff --git a/zzupy/model/eas.py b/zzupy/model/eas.py
index f1dbdcd..6f58eb8 100644
--- a/zzupy/model/eas.py
+++ b/zzupy/model/eas.py
@@ -1,13 +1,13 @@
from __future__ import annotations
import uuid
-from typing import Any, List
+from typing import Any, List, ClassVar
from icalendar import Calendar
from icalendar.cal import Event
from pydantic import BaseModel, model_validator, ConfigDict, RootModel, Field
from pydantic.alias_generators import to_camel
-from whenever import ZonedDateTime, Date, PlainDateTime, Instant
+from whenever import ZonedDateTime, Date, Instant, Time
class Campus(BaseModel):
@@ -60,7 +60,7 @@ class PeriodInfo(BaseModel):
require_theory: int | None
"""要求完成的理论学时数"""
practice: None
- practice_unit: str
+ practice_unit: str | None
require_practice: None
focus_practice: None
focus_practice_unit: None
@@ -325,9 +325,9 @@ def assemble_whenever_datetime(cls, data: dict) -> dict:
return data
try:
- schedule_date = Date.parse_iso(date_str)
- except Exception:
- return data
+ schedule_date = Date.parse_iso(str(date_str))
+ except ValueError as exc:
+ raise ValueError(f"无法解析课程日期 date={date_str!r}") from exc
time_keys = [
"startTime",
@@ -339,16 +339,19 @@ def assemble_whenever_datetime(cls, data: dict) -> dict:
for key in time_keys:
time_val = data.get(key)
if time_val:
+ if isinstance(time_val, ZonedDateTime):
+ continue
+
time_str = str(time_val).strip().zfill(4)
try:
- schedule_time = PlainDateTime.parse_strptime(
- time_str, format="%H%M"
- ).time()
+ schedule_time = Time.parse(time_str, format="hhmm")
data[key] = schedule_date.at(schedule_time).assume_tz(
"Asia/Shanghai"
)
- except Exception:
- pass
+ except ValueError as exc:
+ raise ValueError(
+ f"无法解析课程时间字段 {key}={time_val!r}"
+ ) from exc
return data
@@ -431,8 +434,8 @@ class TeachingWeek(BaseModel):
model_config = ConfigDict(frozen=False)
- DAYS: int = 7
- UNITS: int = 10
+ DAYS: ClassVar[int] = 7
+ UNITS: ClassVar[int] = 10
lessons: dict[tuple[int, int], Lesson] = Field(default_factory=dict)
"""内部存储:仅存储非空课程,key 为 (weekday, unit),value 为 Lesson"""
@@ -553,17 +556,16 @@ def to_calendar(
# 起始时间
start_time = (
- schedule.real_start_time.py_datetime()
- or schedule.start_time.py_datetime()
+ schedule.real_start_time.to_stdlib() or schedule.start_time.to_stdlib()
)
end_time = (
- schedule.real_end_time.py_datetime() or schedule.end_time.py_datetime()
+ schedule.real_end_time.to_stdlib() or schedule.end_time.to_stdlib()
)
event.add("dtstart", start_time)
event.add("dtend", end_time)
# 事件生成时间
- event.add("dtstamp", Instant.now().py_datetime())
+ event.add("dtstamp", Instant.now().to_stdlib())
# 事件 UID
event.add("uid", f"{uuid.uuid4()}@schedule")
@@ -647,18 +649,17 @@ def to_calendar(
# 起始时间
start_time = (
- schedule.real_start_time.py_datetime()
- or schedule.start_time.py_datetime()
+ schedule.real_start_time.to_stdlib()
+ or schedule.start_time.to_stdlib()
)
end_time = (
- schedule.real_end_time.py_datetime()
- or schedule.end_time.py_datetime()
+ schedule.real_end_time.to_stdlib() or schedule.end_time.to_stdlib()
)
event.add("dtstart", start_time)
event.add("dtend", end_time)
# 事件生成时间
- event.add("dtstamp", Instant.now().py_datetime())
+ event.add("dtstamp", Instant.now().to_stdlib())
# 事件 UID
event.add("uid", f"{uuid.uuid4()}@schedule")
diff --git a/zzupy/model/network.py b/zzupy/model/network.py
index 025f952..f26d273 100644
--- a/zzupy/model/network.py
+++ b/zzupy/model/network.py
@@ -1,35 +1,87 @@
import json
+import warnings
from typing import Optional
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+
+def _warn_deprecated_alias(old_name: str, new_name: str) -> None:
+ warnings.warn(
+ f"{old_name} 已弃用,请使用 {new_name}",
+ DeprecationWarning,
+ stacklevel=3,
+ )
class OnlineDevice(BaseModel):
"""在线设备信息"""
+ model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
brasid: str
"""BRAS ID"""
- downFlow: str
+ down_flow: str
"""下行流量"""
- hostName: str = ""
+ host_name: str = ""
"""主机名"""
ip: str
"""IP地址"""
- loginTime: str
+ login_time: str
"""登录时间,格式为YYYY-MM-DD HH:MM:SS"""
mac: str
"""MAC地址"""
- sessionId: str
+ session_id: str
"""会话ID"""
- terminalType: str
+ terminal_type: str
"""终端类型"""
- upFlow: str
+ up_flow: str
"""上行流量"""
- useTime: str
+ use_time: str
"""使用时间(秒)"""
- userId: int
+ user_id: int
"""用户ID"""
+ @property
+ def downFlow(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.downFlow", "down_flow")
+ return self.down_flow
+
+ @property
+ def hostName(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.hostName", "host_name")
+ return self.host_name
+
+ @property
+ def loginTime(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.loginTime", "login_time")
+ return self.login_time
+
+ @property
+ def sessionId(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.sessionId", "session_id")
+ return self.session_id
+
+ @property
+ def terminalType(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.terminalType", "terminal_type")
+ return self.terminal_type
+
+ @property
+ def upFlow(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.upFlow", "up_flow")
+ return self.up_flow
+
+ @property
+ def useTime(self) -> str:
+ _warn_deprecated_alias("OnlineDevice.useTime", "use_time")
+ return self.use_time
+
+ @property
+ def userId(self) -> int:
+ _warn_deprecated_alias("OnlineDevice.userId", "user_id")
+ return self.user_id
+
def dump_json(self, indent: Optional[int] = None) -> str:
"""格式化为JSON字符串"""
return json.dumps(self.model_dump(), ensure_ascii=False, indent=indent)
@@ -38,6 +90,8 @@ def dump_json(self, indent: Optional[int] = None) -> str:
class AuthResult(BaseModel):
"""Portal 认证结果"""
+ model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
result: int
"""认证结果"""
message: str = Field(..., alias="msg")
@@ -52,6 +106,8 @@ def success(self) -> bool:
class PortalInfo(BaseModel):
"""探测出的 Portal 认证信息"""
+ model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
auth_url: str
"""认证网页 URL"""
portal_server_url: str
diff --git a/zzupy/utils.py b/zzupy/utils.py
index 1bae0c9..7b6bec8 100644
--- a/zzupy/utils.py
+++ b/zzupy/utils.py
@@ -259,9 +259,9 @@ def sync_wrapper(self, *args, **kwargs):
raise NotLoggedInError("需要登录")
return func(self, *args, **kwargs)
- import asyncio
+ import inspect
- if asyncio.iscoroutinefunction(func):
+ if inspect.iscoroutinefunction(func):
return async_wrapper
else:
return sync_wrapper
diff --git a/zzupy/web/network.py b/zzupy/web/network.py
index 21bdd4c..0fbbc5b 100644
--- a/zzupy/web/network.py
+++ b/zzupy/web/network.py
@@ -20,6 +20,7 @@
ParsingError,
ZZUError,
)
+from zzupy.logging import logger
from zzupy.model.network import AuthResult, OnlineDevice, PortalInfo
from zzupy.utils import (
extract_first_html_attr,
@@ -66,19 +67,22 @@ def _extract_user_ip(portal_url: str) -> str:
def _extract_auth_url(portal_url: str) -> str:
"""提取网页认证 URL"""
parsed = urllib.parse.urlparse(portal_url)
+ if not parsed.scheme or not parsed.netloc:
+ raise ParsingError("无法从Portal URL获取认证服务器地址")
return f"{parsed.scheme}://{parsed.netloc}"
def _get_portal_server_url(client: httpx2.Client, auth_url: str) -> str:
"""获取 Portal 服务器 URL"""
DEFAULT_HTTP_PORT = 801
DEFAULT_HTTPS_PORT = 802
+ hostname = urllib.parse.urlparse(auth_url).hostname
+ if hostname is None:
+ raise ParsingError("无法从认证 URL 获取 Portal 主机名")
try:
response = client.get(f"{auth_url}/a41.js")
js_params = _parse_js_config(response.text)
- hostname = urllib.parse.urlparse(auth_url).hostname
-
if js_params.get("enableHttps") == 0:
port = js_params.get("epHTTPPort", DEFAULT_HTTP_PORT)
return f"http://{hostname}:{port}"
@@ -86,9 +90,8 @@ def _get_portal_server_url(client: httpx2.Client, auth_url: str) -> str:
port = js_params.get("enHTTPSPort", DEFAULT_HTTPS_PORT)
return f"https://{hostname}:{port}"
- except Exception:
- # 降级到默认配置
- hostname = urllib.parse.urlparse(auth_url).hostname
+ except (httpx2.RequestError, ValueError) as exc:
+ logger.debug("获取 Portal 服务器配置失败,降级到默认配置: {}", exc)
return f"http://{hostname}:{DEFAULT_HTTP_PORT}"
def _parse_js_config(js_content: str) -> dict[str, int]: