72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
|
|
|
|
FORBIDDEN_REFERENCE_ROOT = Path("/Users/yoilun/AI-train/zhengxin-vlm-0413")
|
|
|
|
|
|
def resolve_path(path: str | Path, *, base_dir: Path | None = None) -> Path:
|
|
candidate = Path(path).expanduser()
|
|
if not candidate.is_absolute() and base_dir is not None:
|
|
candidate = base_dir / candidate
|
|
return candidate.resolve(strict=False)
|
|
|
|
|
|
def _is_relative_to(path: Path, parent: Path) -> bool:
|
|
try:
|
|
path.relative_to(parent)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def validate_output_dir(
|
|
input_dir: str | Path,
|
|
output_dir: str | Path,
|
|
*,
|
|
forbidden_root: Path = FORBIDDEN_REFERENCE_ROOT,
|
|
) -> Path:
|
|
resolved_input = resolve_path(input_dir)
|
|
resolved_output = resolve_path(output_dir)
|
|
resolved_forbidden = resolve_path(forbidden_root)
|
|
|
|
if resolved_output == resolved_input:
|
|
raise ValueError("output dir must not equal input dir")
|
|
if _is_relative_to(resolved_output, resolved_forbidden):
|
|
raise ValueError(
|
|
f"output dir must not be inside forbidden reference dir: {resolved_forbidden}"
|
|
)
|
|
return resolved_output
|
|
|
|
|
|
def stable_video_id(path: str | Path) -> str:
|
|
resolved = str(resolve_path(path))
|
|
digest = hashlib.sha1(resolved.encode("utf-8")).hexdigest()[:16]
|
|
return f"video-{digest}"
|
|
|
|
|
|
def hik_cloud_download_path(
|
|
output_dir: str | Path,
|
|
device_serial: str,
|
|
channel_no: int | str,
|
|
time_begin: int,
|
|
time_end: int,
|
|
) -> Path:
|
|
safe_device = _safe_path_component(device_serial)
|
|
safe_channel = _safe_path_component(str(channel_no))
|
|
filename = f"{safe_device}_ch{safe_channel}_{int(time_begin)}_{int(time_end)}.mp4"
|
|
return (
|
|
resolve_path(output_dir)
|
|
/ "downloads"
|
|
/ "hik_cloud"
|
|
/ safe_device
|
|
/ f"ch{safe_channel}"
|
|
/ filename
|
|
)
|
|
|
|
|
|
def _safe_path_component(value: str) -> str:
|
|
return "".join(char if char.isalnum() or char in "._-" else "_" for char in value)
|