Nexus Agent
3864d120fb
Wire alert history items-per-page to the API and parse fire_at ISO strings to datetime so single-shot push schedules save instead of 500.
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""Pydantic coercers for Linux / remote path fields (wraps posix_paths)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from server.utils.posix_paths import PosixPathError, normalize_remote_abs_path, to_posix
|
|
|
|
|
|
def coerce_optional_iso_datetime(value: object) -> datetime | None:
|
|
"""Parse ISO-8601 / MySQL-style datetime for ORM DateTime columns."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value.replace(tzinfo=None) if value.tzinfo else value
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
if text.endswith("Z"):
|
|
text = f"{text[:-1]}+00:00"
|
|
dt = datetime.fromisoformat(text)
|
|
except ValueError as exc:
|
|
raise ValueError("执行时间格式无效") from exc
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return dt
|
|
|
|
|
|
def coerce_optional_remote_abs_path(value: str | None) -> str | None:
|
|
"""Optional remote absolute path (e.g. server.target_path)."""
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return normalize_remote_abs_path(text)
|
|
except PosixPathError as exc:
|
|
raise ValueError(str(exc)) from exc
|
|
|
|
|
|
def coerce_remote_abs_path(value: str) -> str:
|
|
"""Required remote absolute path."""
|
|
try:
|
|
return normalize_remote_abs_path(str(value).strip())
|
|
except PosixPathError as exc:
|
|
raise ValueError(str(exc)) from exc
|
|
|
|
|
|
def coerce_remote_abs_path_list(values: list[str]) -> list[str]:
|
|
return [coerce_remote_abs_path(v) for v in values]
|
|
|
|
|
|
def coerce_nexus_host_abs_path(value: str) -> str:
|
|
"""Nexus 主机上的绝对路径（推送源、上传暂存区等）。"""
|
|
text = str(value).strip()
|
|
if not text:
|
|
raise ValueError("路径不能为空")
|
|
if "\\" in text or "\0" in text:
|
|
raise ValueError("路径必须使用 Linux 路径（不允许反斜杠）")
|
|
path = to_posix(text)
|
|
if not path.startswith("/"):
|
|
raise ValueError("路径必须为绝对路径")
|
|
parts = [p for p in path.split("/") if p and p != "."]
|
|
if ".." in parts:
|
|
raise ValueError("路径不允许包含 '..'")
|
|
normalized = "/" + "/".join(parts) if parts else "/"
|
|
if len(normalized) > 1 and normalized.endswith("/"):
|
|
normalized = normalized.rstrip("/")
|
|
return normalized or "/"
|
|
|
|
|
|
def coerce_remote_relative_path(value: str) -> str:
|
|
"""远程目录下的相对路径（不含前导 /）。"""
|
|
rel = to_posix(str(value).strip()).lstrip("/")
|
|
if not rel or ".." in rel.split("/"):
|
|
raise ValueError("相对路径无效")
|
|
if "\\" in str(value):
|
|
raise ValueError("路径必须使用 Linux 路径（不允许反斜杠）")
|
|
return rel
|