jxxghp
2026-07-28 19:01:55 +08:00
parent 1a528c7803
commit 68686bc23a
5 changed files with 131 additions and 5 deletions
+14 -1
View File
@@ -1,7 +1,10 @@
from typing import Dict, List, Optional, Type, TypeVar, Generic, Iterator
from pydantic import ValidationError
from app.core.module import ModuleManager
from app.db.systemconfig_oper import SystemConfigOper
from app.log import logger
from app.schemas import DownloaderConf, MediaServerConf, NotificationConf, NotificationSwitchConf, ServiceInfo
from app.schemas.types import NotificationType, SystemConfigKey, ModuleType
@@ -25,8 +28,18 @@ class ServiceConfigHelper:
config_data = SystemConfigOper().get(config_key)
if not config_data:
return []
configs = []
for conf in config_data:
if not isinstance(conf, dict):
logger.warn(f"{config_key.value} 配置格式不正确,已跳过:{conf}")
continue
try:
# 直接使用 conf_type 来实例化配置对象
return [conf_type(**conf) for conf in config_data]
configs.append(conf_type(**conf))
except ValidationError as e:
# 单条配置存在非法值时跳过,避免影响其它服务的初始化
logger.error(f"{config_key.value} 配置 {conf.get('name')} 校验失败,已跳过:{e}")
return configs
@staticmethod
def get_downloader_configs() -> List[DownloaderConf]:
+16 -2
View File
@@ -1,9 +1,13 @@
import re
from typing import Optional
from pathlib import Path
from pydantic import BaseModel, Field
from app.schemas.types import StorageSchema
# Windows 盘符绝对路径,如 Z:/Downloads 或 Z:\Downloads
WINDOWS_DRIVE_PATTERN = re.compile(r"^[A-Za-z]:[\\/]")
class FileURI(BaseModel):
# 文件路径
@@ -13,10 +17,19 @@ class FileURI(BaseModel):
@property
def uri(self) -> str:
"""
文件 URI,本地存储直接返回路径,其他存储带上存储前缀
"""
return self.path if self.storage == "local" else f"{self.storage}:{self.path}"
@classmethod
def from_uri(cls, uri: str) -> "FileURI":
"""
解析文件 URI 为存储类型和路径
:param uri: 文件 URI,如 /media/movie、u115:/media/movie 或 Windows 盘符路径 Z:/media
:return: FileURI 对象
"""
storage, path = 'local', uri
for s in StorageSchema:
protocol = f"{s.value}:"
@@ -24,11 +37,13 @@ class FileURI(BaseModel):
path = uri[len(protocol):]
storage = s.value
break
if not path.startswith("/"):
# Windows 盘符路径本身就是绝对路径,补上根斜杠会得到 /Z:/xxx 这样的非法路径
if not path.startswith("/") and not WINDOWS_DRIVE_PATTERN.match(path):
path = "/" + path
path = Path(path).as_posix()
return cls(storage=storage, path=path)
class FileItem(FileURI):
# 类型 dir/file
type: Optional[str] = None
@@ -68,4 +83,3 @@ class StorageUsage(BaseModel):
class StorageTransType(BaseModel):
# 传输类型
transtype: Optional[dict] = Field(default_factory=dict)
+21 -1
View File
@@ -1,7 +1,7 @@
from dataclasses import dataclass
from typing import Optional, Any
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
@dataclass
@@ -40,6 +40,26 @@ class MediaServerConf(BaseModel):
# 自动同步间隔(小时),未设置时使用旧全局配置
sync_interval: Optional[int] = None
@field_validator("sync_interval", mode="before")
@classmethod
def validate_sync_interval(cls, value: Any) -> Optional[int]:
"""
兼容前端清空输入框后残留的空字符串等非法值,避免历史配置导致模块初始化失败
:param value: 原始配置值
:return: 合法的间隔小时数,无法解析时返回 None
"""
if value is None:
return None
if isinstance(value, str):
value = value.strip()
if not value:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
class DownloaderConf(BaseModel):
"""
+50
View File
@@ -0,0 +1,50 @@
from app.schemas.file import FileURI
def test_from_uri_keeps_windows_drive_path() -> None:
"""Windows 盘符路径已是绝对路径,不能再补根斜杠,否则映射网络驱动器整理会报 WinError 123。"""
file_uri = FileURI.from_uri("Z:/Downloads/电视剧/国产剧")
assert file_uri.storage == "local"
assert file_uri.path == "Z:/Downloads/电视剧/国产剧"
assert file_uri.uri == "Z:/Downloads/电视剧/国产剧"
def test_from_uri_keeps_windows_drive_path_with_backslash() -> None:
"""反斜杠写法的盘符路径同样不能补根斜杠。"""
file_uri = FileURI.from_uri("Z:\\Downloads\\电视剧")
assert file_uri.storage == "local"
assert not file_uri.path.startswith("/")
def test_from_uri_keeps_posix_absolute_path() -> None:
"""POSIX 绝对路径保持原样。"""
file_uri = FileURI.from_uri("/downloads/movies")
assert file_uri.storage == "local"
assert file_uri.path == "/downloads/movies"
def test_from_uri_adds_root_for_relative_path() -> None:
"""无前导斜杠的相对路径仍补全为绝对路径。"""
file_uri = FileURI.from_uri("downloads/movies")
assert file_uri.path == "/downloads/movies"
def test_from_uri_parses_storage_prefix() -> None:
"""带存储前缀的 URI 应拆分出存储类型并保留 POSIX 路径。"""
file_uri = FileURI.from_uri("u115:/media/anime")
assert file_uri.storage == "u115"
assert file_uri.path == "/media/anime"
assert file_uri.uri == "u115:/media/anime"
def test_from_uri_storage_prefix_with_relative_path() -> None:
"""远端存储的无前导斜杠路径补全为绝对路径。"""
file_uri = FileURI.from_uri("rclone:media/anime")
assert file_uri.storage == "rclone"
assert file_uri.path == "/media/anime"
@@ -0,0 +1,29 @@
from app.helper.service import ServiceConfigHelper
from app.schemas.system import MediaServerConf
from app.schemas.types import SystemConfigKey
def test_mediaserver_conf_tolerates_blank_sync_interval():
"""自动同步间隔为空字符串等非法值时应回退为 None 而不是抛出校验错误。"""
assert MediaServerConf(name="blank", sync_interval="").sync_interval is None
assert MediaServerConf(name="spaces", sync_interval=" ").sync_interval is None
assert MediaServerConf(name="invalid", sync_interval="abc").sync_interval is None
assert MediaServerConf(name="text", sync_interval="12").sync_interval == 12
assert MediaServerConf(name="number", sync_interval=6).sync_interval == 6
assert MediaServerConf(name="none", sync_interval=None).sync_interval is None
def test_get_configs_skips_invalid_entries(monkeypatch):
"""单条配置校验失败时应跳过该条,不影响其它服务配置的加载。"""
monkeypatch.setattr(
"app.helper.service.SystemConfigOper.get",
lambda self, key: [
{"name": "good", "type": "emby", "enabled": True},
"bad-format",
{"name": "bad-type", "type": "plex", "enabled": "maybe"},
],
)
configs = ServiceConfigHelper.get_configs(SystemConfigKey.MediaServers, MediaServerConf)
assert [conf.name for conf in configs] == ["good"]