mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 21:34:32 +08:00
334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""
|
||
文件系统操作代理 worker。
|
||
|
||
**本文件不能被 import，只能作为独立脚本执行**（fsproxy 用
|
||
`subprocess.Popen([sys.executable, <本文件绝对路径>])` 启动）。直接执行文件
|
||
路径不会触发 `app/__init__.py` 的导入链，因此这个进程只依赖标准库、启动是
|
||
毫秒级的；一旦走 import 就会把整个应用的依赖拉进来，代理被强杀后的重启成本
|
||
会高到无法接受。
|
||
|
||
存在的理由：FUSE/网络挂载进入 block 型故障时，`stat`/`listdir`/`rename` 这类
|
||
系统调用既不返回错误也不返回结果，而 Python 没有中断线程的手段——阻塞其上的
|
||
线程永远无法回收。放进独立进程后，父进程可以在超时后 SIGKILL 掉它，把
|
||
「不可处理的 block」转换成「可处理的 crash」。
|
||
|
||
协议：stdin/stdout 逐行 JSON。
|
||
请求 {"op": "stat", "path": "/mnt/cd2/x.mkv"}
|
||
成功 {"ok": true, "result": {...}}
|
||
失败 {"ok": false, "errno": 2, "error": "No such file or directory"}
|
||
"""
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Callable, Required, TypedDict
|
||
|
||
|
||
class _Request(TypedDict, total=False):
|
||
"""
|
||
父进程发送的文件系统操作请求，除 op 外的键由具体操作决定是否必需。
|
||
|
||
:ivar op: 必需的操作名，例如 stat、copy 或 rename
|
||
:ivar path: 路径查询、删除或递归扫描的目标路径，这些操作必须提供
|
||
:ivar src: copy 或 rename 操作必须提供的源路径
|
||
:ivar dst: copy 或 rename 操作必须提供的目标路径
|
||
:ivar chunk_size: 可选的复制块大小，单位为字节；缺失或为 0 时使用 1 MiB
|
||
:ivar progress_interval: 可选的复制进度发送间隔，单位为秒；缺失或为 0 时使用 1 秒
|
||
:ivar max_check: 可选的递归扫描文件数阈值；缺失或为 0 时使用 10000，超过后停止扫描
|
||
:ivar extensions: 后缀查询必须提供的扩展名列表，按忽略大小写的规则匹配
|
||
"""
|
||
|
||
op: Required[str]
|
||
path: str
|
||
src: str
|
||
dst: str
|
||
chunk_size: int
|
||
progress_interval: float
|
||
max_check: int
|
||
extensions: list[str]
|
||
|
||
|
||
class _StatResult(TypedDict):
|
||
"""
|
||
路径属性与删除前用于核对文件身份的信息，所有键均存在。
|
||
|
||
:ivar size: 文件大小，单位为字节
|
||
:ivar mtime: 文件修改时间的 Unix 秒级时间戳
|
||
:ivar mtime_ns: 文件修改时间的 Unix 纳秒级时间戳
|
||
:ivar ctime_ns: 文件状态变化时间的 Unix 纳秒级时间戳
|
||
:ivar dev: 文件所属设备号，与 ino 一起用于核对身份
|
||
:ivar ino: 文件系统索引节点号
|
||
:ivar is_dir: 查询时目标是否为目录
|
||
:ivar is_file: 查询时目标是否为普通文件
|
||
"""
|
||
|
||
size: int
|
||
mtime: float
|
||
mtime_ns: int
|
||
ctime_ns: int
|
||
dev: int
|
||
ino: int
|
||
is_dir: bool
|
||
is_file: bool
|
||
|
||
|
||
class _CopyProgress(TypedDict):
|
||
"""
|
||
复制期间的进度心跳或结束时的结果，所有键均存在。
|
||
|
||
:ivar copied: 已复制的字节数
|
||
:ivar total: 复制开始时读取的源文件总字节数
|
||
"""
|
||
|
||
copied: int
|
||
total: int
|
||
|
||
|
||
class _EntryCounts(TypedDict):
|
||
"""
|
||
递归扫描实际观察到的文件和目录数量，所有键均存在。
|
||
|
||
:ivar file_count: 已发现的文件数，达到扫描停止条件时可能不是完整目录总数
|
||
:ivar dir_count: 已发现的子目录数，不包含扫描根目录
|
||
"""
|
||
|
||
file_count: int
|
||
dir_count: int
|
||
|
||
|
||
_Result = bool | str | list[str] | _StatResult | _CopyProgress | _EntryCounts
|
||
|
||
|
||
class _Response(TypedDict, total=False):
|
||
"""
|
||
逐行发送的复制进度、成功结果或失败信息。
|
||
|
||
:ivar ok: 必需的成功标志；进度心跳也为 True
|
||
:ivar progress: 复制期间才存在的字节进度，不表示操作已经结束
|
||
:ivar result: 成功结束时返回的操作结果，形状由请求操作决定
|
||
:ivar errno: 失败时携带的系统错误号；未知操作、无错误号或其他异常使用 0
|
||
:ivar error: 失败时携带的可读异常说明
|
||
"""
|
||
|
||
ok: Required[bool]
|
||
progress: _CopyProgress
|
||
result: _Result
|
||
errno: int
|
||
error: str
|
||
|
||
|
||
_Emit = Callable[[_Response], None]
|
||
|
||
|
||
def _stat(payload: _Request, _emit: _Emit) -> _StatResult:
|
||
"""
|
||
读取路径属性及设备、inode 身份，供删除前确认仍是同一个文件。
|
||
"""
|
||
path = payload["path"]
|
||
info = os.stat(path)
|
||
return {
|
||
"size": info.st_size,
|
||
"mtime": info.st_mtime,
|
||
"mtime_ns": info.st_mtime_ns,
|
||
"ctime_ns": info.st_ctime_ns,
|
||
"dev": info.st_dev,
|
||
"ino": info.st_ino,
|
||
"is_dir": os.path.isdir(path),
|
||
"is_file": os.path.isfile(path),
|
||
}
|
||
|
||
|
||
def _exists(payload: _Request, _emit: _Emit) -> bool:
|
||
"""
|
||
判断路径是否存在。
|
||
|
||
用 os.stat 而不是 os.path.exists：后者会把任意 OSError 都归为「不存在」，
|
||
挂载抖动会被误判成文件消失。这里让异常原样抛出，由父进程按 errno 区分。
|
||
"""
|
||
os.stat(payload["path"])
|
||
return True
|
||
|
||
|
||
def _realpath(payload: _Request, _emit: _Emit) -> str:
|
||
"""
|
||
解析已有父目录的符号链接，目标末段可尚不存在；异常交由代理保守处理。
|
||
"""
|
||
return str(Path(payload["path"]).resolve())
|
||
|
||
|
||
def _listdir(payload: _Request, _emit: _Emit) -> list[str]:
|
||
"""
|
||
列出目录下的条目名。
|
||
"""
|
||
return sorted(os.listdir(payload["path"]))
|
||
|
||
|
||
def _copy(payload: _Request, emit: _Emit) -> _CopyProgress:
|
||
"""
|
||
分块复制文件内容并周期上报进度。
|
||
|
||
进度上报同时充当心跳：复制大文件可能持续几小时，父进程无法用固定超时判断
|
||
挂死，只能看「两次上报之间隔了多久」。因此这里按固定时间间隔上报，即使
|
||
某一秒没读到数据也照常发——一旦挂载卡住，read/write 不返回，上报自然断流，
|
||
父进程据此判定并强杀本进程。
|
||
|
||
只复制内容和时间戳，不复制权限：目标目录的默认权限与继承 ACL 是媒体库的
|
||
访问策略，用源文件权限覆盖会清除已继承的 ACL。
|
||
"""
|
||
src, dst = payload["src"], payload["dst"]
|
||
chunk_size = payload.get("chunk_size") or 1024 * 1024
|
||
interval = payload.get("progress_interval") or 1.0
|
||
|
||
info = os.stat(src)
|
||
total = info.st_size
|
||
copied = 0
|
||
# 先发一次 0%：既让心跳立刻开始，也保证父进程在传输开始前就有一次检查
|
||
# 取消的机会——否则小文件会在首次定时上报之前就复制完，取消形同虚设
|
||
emit({"ok": True, "progress": {"copied": 0, "total": total}})
|
||
last_emit = time.monotonic()
|
||
with open(src, "rb") as fsrc, open(dst, "wb") as fdst:
|
||
while True:
|
||
buf = fsrc.read(chunk_size)
|
||
if not buf:
|
||
break
|
||
fdst.write(buf)
|
||
copied += len(buf)
|
||
now = time.monotonic()
|
||
if now - last_emit >= interval:
|
||
last_emit = now
|
||
emit({"ok": True, "progress": {"copied": copied, "total": total}})
|
||
os.utime(dst, ns=(info.st_atime_ns, info.st_mtime_ns))
|
||
return {"copied": copied, "total": total}
|
||
|
||
|
||
def _count_entries(payload: _Request, _emit: _Emit) -> _EntryCounts:
|
||
"""
|
||
统计目录下的文件与子目录数量，超过上限即提前结束。
|
||
|
||
放在子进程里做而不是逐层 listdir 走 IPC：递归遍历一棵大目录树会产生成千
|
||
上万次往返，代价不可接受；一次调用在子进程内跑完 os.walk，父进程只需对
|
||
这一次调用设超时即可整体放弃。
|
||
"""
|
||
directory = payload["path"]
|
||
max_check = payload.get("max_check") or 10000
|
||
file_count = 0
|
||
dir_count = 0
|
||
for _, dirs, files in os.walk(directory):
|
||
file_count += len(files)
|
||
dir_count += len(dirs)
|
||
if file_count > max_check:
|
||
break
|
||
return {"file_count": file_count, "dir_count": dir_count}
|
||
|
||
|
||
def _has_file_suffix(payload: _Request, _emit: _Emit) -> bool:
|
||
"""
|
||
遍历目录树；读取失败必须抛出，避免将未完成的扫描视为安全。
|
||
"""
|
||
suffixes = {ext.casefold() for ext in payload["extensions"]}
|
||
|
||
def raise_walk_error(error: OSError) -> None:
|
||
"""
|
||
向调用方传递目录扫描错误。
|
||
"""
|
||
raise error
|
||
|
||
for _, _, files in os.walk(payload["path"], onerror=raise_walk_error):
|
||
if any(Path(name).suffix.casefold() in suffixes for name in files):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _rename(payload: _Request, _emit: _Emit) -> bool:
|
||
"""
|
||
同一存储内重命名/移动。
|
||
|
||
这是第一版唯一放行的写操作：同文件系统内的 rename 由内核保证原子性，
|
||
进程被强杀后要么完全成功要么完全没发生，不存在需要清理的中间状态。
|
||
跨存储的复制+删除不走这里，它需要单独的可恢复语义。
|
||
"""
|
||
src, dst = payload["src"], payload["dst"]
|
||
# rename 操作处理的是软链接目录项本身，不能把目标文件所在设备误当成源设备。
|
||
if os.lstat(src).st_dev != os.stat(os.path.dirname(dst) or ".").st_dev:
|
||
raise OSError(18, "Cross-device rename is not handled by the proxy")
|
||
os.rename(src, dst)
|
||
return True
|
||
|
||
|
||
def _unlink(payload: _Request, _emit: _Emit) -> bool:
|
||
"""
|
||
删除单个文件。unlink 是原子操作，强杀后要么删掉了要么没删，没有中间状态。
|
||
"""
|
||
os.unlink(payload["path"])
|
||
return True
|
||
|
||
|
||
def _rmtree(payload: _Request, _emit: _Emit) -> bool:
|
||
"""
|
||
递归删除目录。
|
||
|
||
这一项不是原子的，强杀可能只删掉一部分。放行的理由是：删除被中断的后果
|
||
（残留若干文件）远轻于写入被中断（留下叫最终文件名的半成品），而且调用方
|
||
本来就以 ignore_errors 容忍部分失败、可以重复执行直到成功。
|
||
"""
|
||
shutil.rmtree(payload["path"], ignore_errors=True)
|
||
return True
|
||
|
||
|
||
_HANDLERS: dict[str, Callable[[_Request, _Emit], _Result]] = {
|
||
"stat": _stat,
|
||
"realpath": _realpath,
|
||
"exists": _exists,
|
||
"listdir": _listdir,
|
||
"copy": _copy,
|
||
"count_entries": _count_entries,
|
||
"has_file_suffix": _has_file_suffix,
|
||
"rename": _rename,
|
||
"unlink": _unlink,
|
||
"rmtree": _rmtree,
|
||
"ping": lambda _payload, _emit: True,
|
||
}
|
||
|
||
|
||
def _write(message: _Response) -> None:
|
||
"""
|
||
输出一行响应。
|
||
"""
|
||
sys.stdout.write(json.dumps(message) + "\n")
|
||
sys.stdout.flush()
|
||
|
||
|
||
def main() -> None:
|
||
"""
|
||
请求循环：每读一行处理一个请求，直到 stdin 关闭。
|
||
|
||
一个请求可能对应多行响应：长耗时操作先流式发若干 progress 行，最后发一行
|
||
终态（result 或 error）。父进程据此区分「还在推进」和「已经挂死」。
|
||
"""
|
||
response: _Response
|
||
for line in sys.stdin:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
payload: _Request = json.loads(line)
|
||
handler = _HANDLERS.get(payload.get("op", ""))
|
||
if handler is None:
|
||
response = {"ok": False, "errno": 0,
|
||
"error": f"unknown op: {payload.get('op')}"}
|
||
else:
|
||
response = {"ok": True, "result": handler(payload, _write)}
|
||
except OSError as err:
|
||
response = {"ok": False, "errno": err.errno or 0,
|
||
"error": err.strerror or str(err)}
|
||
# worker 不能因任何异常退出
|
||
except Exception as err: # noqa: BLE001
|
||
response = {"ok": False, "errno": 0, "error": str(err)}
|
||
_write(response)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|