chore: apply pre-commit fixes

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -1,6 +1,5 @@
 import asyncio
 import copy
-import sys
 import time
 import traceback
 import typing as T
@@ -8,6 +7,7 @@ import uuid
 from contextlib import suppress
 from dataclasses import dataclass, field, replace
 from pathlib import Path
+from typing import override
 
 from mcp.types import (
     BlobResourceContents,
@@ -63,11 +63,6 @@ from ..run_context import ContextWrapper, TContext
 from ..tool_executor import BaseFunctionToolExecutor
 from .base import AgentResponse, AgentState, BaseAgentRunner
 
-if sys.version_info >= (3, 12):
-    from typing import override
-else:
-    from typing_extensions import override
-
 
 @dataclass(slots=True)
 class _HandleFunctionToolsResult:

astrbot/core/utils/totp.py

@@ -60,7 +60,7 @@ def _get_verified_totp_timecode(secret: str, code: str) -> int | None:
     code = code.strip()
     try:
         totp = pyotp.TOTP(secret.strip())
-        now = datetime.datetime.now(datetime.timezone.utc)
+        now = datetime.datetime.now(datetime.UTC)
         for offset in (-1, 0, 1):
             candidate_time = now + datetime.timedelta(seconds=offset * totp.interval)
             if hmac.compare_digest(str(totp.at(candidate_time)), code):
@@ -175,7 +175,7 @@ async def is_totp_trusted_device_valid(config, db, cookie_token: str) -> bool:
                 col(DashboardTrustedDevice.token_hash) == token_hash,
                 col(DashboardTrustedDevice.totp_secret_hash) == totp_secret_hash,
                 col(DashboardTrustedDevice.expires_at)
-                > datetime.datetime.now(datetime.timezone.utc),
+                > datetime.datetime.now(datetime.UTC),
             )
         )
         return result.scalar_one_or_none() is not None
@@ -189,7 +189,7 @@ async def issue_totp_trusted_device(config, db) -> str | None:
     if not token_hash or not totp_secret_hash:
         return None
 
-    expires_at = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+    expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(
         seconds=TOTP_TRUSTED_DEVICE_MAX_AGE
     )
     async with db.get_db() as session:
@@ -216,7 +216,7 @@ async def _cleanup_expired_totp_trusted_devices(db) -> None:
             await session.execute(
                 delete(DashboardTrustedDevice).where(
                     col(DashboardTrustedDevice.expires_at)
-                    <= datetime.datetime.now(datetime.timezone.utc)
+                    <= datetime.datetime.now(datetime.UTC)
                 )
             )
 

astrbot/dashboard/services/log_service.py

@@ -26,7 +26,7 @@ class LogService:
         }
         return f"id: {ts}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
 
-    async def replay_cached_logs(self, last_event_id: str) -> AsyncGenerator[str, None]:
+    async def replay_cached_logs(self, last_event_id: str) -> AsyncGenerator[str]:
         try:
             last_ts = float(last_event_id)
             cached_logs = list(self.log_broker.log_cache)
@@ -40,9 +40,7 @@ class LogService:
         except Exception as exc:
             logger.error(f"Log SSE 补发历史错误: {exc}")
 
-    async def stream_log_events(
-        self, last_event_id: str | None
-    ) -> AsyncGenerator[str, None]:
+    async def stream_log_events(self, last_event_id: str | None) -> AsyncGenerator[str]:
         queue = None
         try:
             if last_event_id: