refactor: resolve all 26 baseline mypy errors; precise test assertions

Per-error treatment (no bulk script): honest annotations for
StoredDownload (Path/bytes/Callable), KeyValue.value (Any for JSONField),
dict[str, Any] for form/config dicts; signature split in create_token;
str() normalization in config int(); platform/type-ignore only where the
checker cannot follow (msvcrt, FastAPI Request injection). Baseline file
and ratchet compare now redundant at zero errors but kept as the guard
against regressions.

Tests: assertions tightened to status_code == 403 with HTTPException
raises; _create_share explicit parameters replace **extra passthrough;
seek-fix pointer-reset assertion persisted.

Author: Marrrrrrrrry

apps/admin/config_service.py

@@ -94,7 +94,7 @@ async def update_config(self, data: dict):
                 raise HTTPException(status_code=400, detail=f"{key} 配置值格式错误")
 
         try:
-            session_expire = int(next_config.get("admin_session_expire"))
+            session_expire = int(str(next_config.get("admin_session_expire")))
         except (TypeError, ValueError):
             raise HTTPException(
                 status_code=400,

apps/admin/dependencies.py

@@ -54,7 +54,8 @@ def get_admin_session(authorization: str = Header(default=None)) -> dict:
 
 
 async def admin_required(
-    authorization: str = Header(default=None), request: Request = None
+    authorization: str = Header(default=None),
+    request: Request = None,  # type: ignore[assignment]  # FastAPI 运行时注入 Request；可选语义由下方 if request 判断
 ):
     """
     验证管理员权限

apps/admin/local_files.py

@@ -1,5 +1,6 @@
 """Admin-side local (NAS) file browsing and deletion."""
 from pathlib import Path
+from typing import Any
 
 from fastapi import HTTPException
 
@@ -20,7 +21,7 @@ async def list_files(self, path: str = ""):
             raise HTTPException(status_code=404, detail="目录不存在")
 
         root = get_local_root()
-        items = []
+        items: list[dict[str, Any]] = []
         try:
             children = list(directory.iterdir())
         except OSError as exc:

apps/admin/views.py

@@ -3,6 +3,7 @@
 # @File    : views.py
 # @Software: PyCharm
 import datetime
+from typing import Any
 from collections import Counter
 from typing import Optional
 
@@ -273,14 +274,15 @@ async def batch_update_files(
     if not data.ids:
         raise HTTPException(status_code=400, detail="请选择要更新的文件")
 
-    update_data = {}
+    update_data: dict[str, Any] = {}
     fields_set = data.model_fields_set
     should_clear_expired_at = bool(data.clear_expired_at)
 
     if should_clear_expired_at:
         update_data["expired_at"] = None
         update_data["expired_count"] = -1
     elif "expired_at" in fields_set and data.expired_at != "":
+        # schema 声明 Union[datetime, str]——透传两种形态，tortoise 均可序列化
         update_data["expired_at"] = data.expired_at
 
     if (

apps/base/auth.py

@@ -62,10 +62,10 @@ def create_token(data: dict, expires_in: int | None = None) -> str:
         ).encode()
     ).decode().rstrip("=")
 
-    signature = hmac.new(
+    signature_digest = hmac.new(
         _get_jwt_secret(), f"{header}.{payload}".encode(), "sha256"
     ).digest()
-    signature = base64.urlsafe_b64encode(signature).decode().rstrip("=")
+    signature = base64.urlsafe_b64encode(signature_digest).decode().rstrip("=")
 
     return f"{header}.{payload}.{signature}"
 

apps/base/models.py

@@ -2,13 +2,13 @@
 # @Author  : Lan
 # @File    : models.py
 # @Software: PyCharm
-from typing import Optional
+
+from typing import Any
 
 from tortoise.models import Model
 from tortoise.contrib.pydantic import pydantic_model_creator
 
 from tortoise import fields, models
-from datetime import datetime
 from core.utils import get_now
 
 
@@ -57,14 +57,13 @@ class UploadChunk(models.Model):
 
 
 class KeyValue(Model):
-    id: Optional[int] = fields.IntField(pk=True)
-    key: Optional[str] = fields.CharField(
-        max_length=255, description="键", index=True, unique=True
-    )
-    value: Optional[str] = fields.JSONField(description="值", null=True)
-    created_at: Optional[datetime] = fields.DatetimeField(
-        auto_now_add=True, description="创建时间"
-    )
+    # 与 FileCodes 保持同一风格：类属性不写值注解（等号右侧的 Field 描述符
+    # 才是类级真身；实例级由 tortoise 运行时给出对应 Python 值）
+    id = fields.IntField(pk=True)
+    key = fields.CharField(max_length=255, description="键", index=True, unique=True)
+    # JSONField 实际承载 dict/list（如 settings 配置）
+    value: Any = fields.JSONField(description="值", null=True)
+    created_at = fields.DatetimeField(auto_now_add=True, description="创建时间")
 
 
 class PresignUploadSession(models.Model):

apps/base/services.py

@@ -197,7 +197,9 @@ async def create_file_share(
             raise
         finally:
             await release_storage(token)
-        return {"code": code, "name": file.filename}
+        # UploadFile.filename 类型为 str | None；路径生成已用 "" 兜底，
+        # 响应处保持 str 以匹配 dict[str, str] 契约
+        return {"code": code, "name": file.filename or ""}
 
     @staticmethod
     async def complete_chunked_upload(
@@ -402,6 +404,9 @@ def response_from_download(download: StoredDownload):
         return Response(
             download.content, media_type=download.media_type, headers=download.headers
         )
+    if download.stream_factory is None:
+        # 数据契约要求三者至少其一；防御性 500 语义
+        raise ValueError("StoredDownload 缺少可用的下载载荷")
     return StreamingResponse(
         download.stream_factory(),
         media_type=download.media_type,

apps/base/setup_wizard.py

@@ -5,6 +5,8 @@
 import html
 from urllib.parse import parse_qs
 
+from typing import Any
+
 from fastapi import Request
 from fastapi.responses import HTMLResponse
 
@@ -73,14 +75,14 @@ def build_public_meta() -> dict:
 
 
 
-def get_form_value(data: dict, key: str, default: str = "") -> str:
+def get_form_value(data: dict[str, Any], key: str, default: str = "") -> str:
     value = data.get(key, default)
     if isinstance(value, list):
         value = value[-1] if value else default
     return str(value if value is not None else default)
 
 
-def get_form_list(data: dict, key: str) -> list[str]:
+def get_form_list(data: dict[str, Any], key: str) -> list[str]:
     value = data.get(key, [])
     if isinstance(value, list):
         return [str(item) for item in value if str(item)]
@@ -96,7 +98,7 @@ def normalize_bool_field(data: dict, key: str, default: bool) -> bool:
 
 
 def parse_int_field(
-    data: dict,
+    data: dict[str, Any],
     key: str,
     default: int,
     label: str,

core/database.py

@@ -47,7 +47,7 @@ def _lock_file(file_obj: IO[str]) -> None:
         if os.fstat(file_obj.fileno()).st_size == 0:
             file_obj.write("0")
             file_obj.flush()
-        msvcrt.locking(file_obj.fileno(), msvcrt.LK_LOCK, 1)
+        msvcrt.locking(file_obj.fileno(), msvcrt.LK_LOCK, 1)  # type: ignore[attr-defined]
     else:
         import fcntl
 
@@ -58,7 +58,7 @@ def _unlock_file(file_obj: IO[str]) -> None:
     if os.name == "nt":
         import msvcrt
 
-        msvcrt.locking(file_obj.fileno(), msvcrt.LK_UNLCK, 1)
+        msvcrt.locking(file_obj.fileno(), msvcrt.LK_UNLCK, 1)  # type: ignore[attr-defined]
     else:
         import fcntl
 

core/security.py

@@ -119,7 +119,9 @@ def _endpoint_host_is_denied(host: str) -> bool:
     except (socket.gaierror, UnicodeError, OSError):
         return False
     for info in infos:
-        if _endpoint_ip_is_denied(info[4][0]):
+        # getaddrinfo 的 sockaddr 首元素在 IPv4/IPv6 下均为 str（int 仅出现在
+        # AF_UNIX 等异形族）——显式 str() 让类型检查器与运行时一致。
+        if _endpoint_ip_is_denied(str(info[4][0])):
             return True
     return False