103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
|
"""跨平台路径工具：重点兼容 Windows UNC 网络路径（\\\\NAS\\share\\...）。"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path, PureWindowsPath
|
||
|
|
|
||
|
|
|
||
|
|
def normalize_user_path(raw: str) -> str:
|
||
|
|
"""规范化用户输入的路径，保留 UNC 反斜杠格式。
|
||
|
|
|
||
|
|
示例：
|
||
|
|
- \\\\JIULUGNAS\\personal_folder\\Photos -> 原样保留
|
||
|
|
- //JIULUGNAS/personal_folder/Photos -> 转为 UNC
|
||
|
|
- D:/Pictures/Screenshots -> D:\\Pictures\\Screenshots
|
||
|
|
"""
|
||
|
|
raw = (raw or "").strip().strip('"').strip("'")
|
||
|
|
if not raw:
|
||
|
|
return raw
|
||
|
|
|
||
|
|
if sys.platform == "win32":
|
||
|
|
# //server/share -> \\server\share
|
||
|
|
if raw.startswith("//") and not raw.startswith("///"):
|
||
|
|
raw = "\\\\" + raw.lstrip("/").replace("/", "\\")
|
||
|
|
elif raw.startswith("\\\\"):
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
raw = raw.replace("/", "\\")
|
||
|
|
return str(PureWindowsPath(raw))
|
||
|
|
|
||
|
|
return str(Path(raw).expanduser())
|
||
|
|
|
||
|
|
|
||
|
|
def path_from_storage(stored: str) -> Path:
|
||
|
|
"""从数据库读出的路径转为 Path（修复历史 as_posix 导致的 //NAS/...）。"""
|
||
|
|
if not stored:
|
||
|
|
return Path(stored)
|
||
|
|
if sys.platform == "win32":
|
||
|
|
# 历史数据：//JIULUGNAS/foo/bar -> \\JIULUGNAS\foo\bar
|
||
|
|
if stored.startswith("//") and not stored.startswith("///"):
|
||
|
|
stored = "\\\\" + stored.lstrip("/").replace("/", "\\")
|
||
|
|
return Path(stored)
|
||
|
|
|
||
|
|
|
||
|
|
def path_to_storage(path: Path | str) -> str:
|
||
|
|
"""写入数据库 / 比较用的路径字符串；Windows 下保留反斜杠。"""
|
||
|
|
if isinstance(path, Path):
|
||
|
|
if sys.platform == "win32":
|
||
|
|
return str(path)
|
||
|
|
return path.as_posix()
|
||
|
|
return normalize_user_path(str(path)) if sys.platform == "win32" else str(path)
|
||
|
|
|
||
|
|
|
||
|
|
def is_accessible_dir(path: str | Path) -> bool:
|
||
|
|
"""目录是否可访问（UNC / 本地均适用）。"""
|
||
|
|
try:
|
||
|
|
return os.path.isdir(str(path))
|
||
|
|
except OSError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def is_accessible_file(path: str | Path) -> bool:
|
||
|
|
"""文件是否可访问。"""
|
||
|
|
try:
|
||
|
|
return os.path.isfile(str(path))
|
||
|
|
except OSError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def path_is_under(parent: str | Path, child: str | Path) -> bool:
|
||
|
|
"""判断 child 是否在 parent 目录下（用于敏感目录检测）。"""
|
||
|
|
try:
|
||
|
|
parent_norm = os.path.normcase(os.path.normpath(str(parent)))
|
||
|
|
child_norm = os.path.normcase(os.path.normpath(str(child)))
|
||
|
|
if not parent_norm.endswith(os.sep):
|
||
|
|
parent_norm += os.sep
|
||
|
|
return child_norm.startswith(parent_norm) or child_norm == parent_norm.rstrip(os.sep)
|
||
|
|
except OSError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def count_files_sample(root: str | Path, limit: int = 5) -> tuple[int, list[str]]:
|
||
|
|
"""快速抽样统计目录下图片数量（网络路径可能较慢，limit 控制遍历深度）。"""
|
||
|
|
from app.services.thumbnail import is_supported
|
||
|
|
|
||
|
|
root_p = path_from_storage(str(root)) if isinstance(root, str) else root
|
||
|
|
total = 0
|
||
|
|
samples: list[str] = []
|
||
|
|
try:
|
||
|
|
for dirpath, _, filenames in os.walk(str(root_p)):
|
||
|
|
for name in filenames:
|
||
|
|
p = Path(dirpath) / name
|
||
|
|
if not is_supported(p):
|
||
|
|
continue
|
||
|
|
total += 1
|
||
|
|
if len(samples) < limit:
|
||
|
|
samples.append(path_to_storage(p))
|
||
|
|
if total >= 1000:
|
||
|
|
break
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
return total, samples
|