chore: merge dev into main

docs/note.md

@@ -1,5 +1,5 @@
 # Note
-<font color=gray size=3> 一些逆向过程的笔记</font>
+<span  style="color: gray; font-size: small; "> 一些逆向过程的笔记</span>
 
 ### 充值电费
 ZZU 的电费充值系统对充值的请求体进行了加密，这本没有问题，但他居然选择的是国密算法，SM2 和 SM4。可能是国产化指标吧，唉😮‍💨。  

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`：

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:

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,29 +67,31 @@ 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}"
             else:
                 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]:

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:

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")

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

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