AdamWen95
76 lines
3.6 KiB
Python
76 lines
3.6 KiB
Python
"""Read-only output discovery, history preservation and small runtime snapshots."""
|
|
import glob
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
from .artifacts import files_under
|
|
|
|
|
|
def outputs(patterns):
|
|
paths = []
|
|
for pattern in patterns:
|
|
expanded = glob.iglob(pattern) if glob.has_magic(pattern) else [pattern]
|
|
for name in expanded:
|
|
path = Path(name)
|
|
if path.is_symlink():
|
|
continue
|
|
if path.is_file():paths.append(path)
|
|
elif path.is_dir():paths.extend(files_under([path], limit=4097))
|
|
if len(paths) > 4096:
|
|
raise ValueError("原输出文件超过 4096 项，请收窄登记输出范围")
|
|
return sorted(set(paths))
|
|
|
|
|
|
def preserve(paths, directory, maximum=64 * 1024**3):
|
|
sizes = [path.stat().st_size for path in paths]
|
|
total = sum(sizes)
|
|
if total > maximum or shutil.disk_usage(directory).free < total + 10 * 1024**3:
|
|
raise ValueError("保护原有同名结果所需空间不足或超过 64 GiB，本次未启动")
|
|
destination = directory / "previous-outputs"
|
|
destination.mkdir(mode=0o700)
|
|
manifest = []
|
|
for path in paths:
|
|
identity = hashlib.sha256(str(path).encode()).hexdigest()
|
|
target = destination / identity
|
|
before = path.stat()
|
|
with path.open("rb") as source, target.open("xb") as output:
|
|
shutil.copyfileobj(source, output, length=1024 * 1024)
|
|
after = path.stat()
|
|
if (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns):
|
|
raise ValueError("原输出仍被其他任务写入，请等待后再试")
|
|
manifest.append({"path": str(path), "backup": identity, "size": before.st_size})
|
|
(destination / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return len(manifest)
|
|
|
|
|
|
def runtime_snapshot():
|
|
result = {"containers": [], "tmux_sessions": [], "errors": []}
|
|
try:
|
|
raw = subprocess.run(["/usr/bin/docker", "ps", "-a", "--format", "{{json .}}"], capture_output=True, timeout=10, text=True)
|
|
if raw.returncode == 0:
|
|
for line in raw.stdout.splitlines():
|
|
value = json.loads(line)
|
|
result["containers"].append({key: value.get(key) for key in ("ID", "Names", "Image", "State", "Status")})
|
|
else:result["errors"].append("容器状态读取失败")
|
|
except (OSError, ValueError, subprocess.TimeoutExpired):result["errors"].append("容器状态不可用")
|
|
try:
|
|
raw = subprocess.run(["/usr/bin/tmux", "list-sessions", "-F", "#{session_id}\t#{session_name}"], capture_output=True, timeout=5, text=True)
|
|
if raw.returncode == 0:
|
|
result["tmux_sessions"] = [dict(zip(("id", "name"), line.split("\t", 1))) for line in raw.stdout.splitlines() if "\t" in line]
|
|
except (OSError, subprocess.TimeoutExpired):result["errors"].append("tmux 状态不可用")
|
|
return result
|
|
|
|
|
|
def differences(before, after):
|
|
old = {v["ID"]: v for v in before["containers"]}
|
|
new = {v["ID"]: v for v in after["containers"]}
|
|
sessions = {v["id"] for v in before["tmux_sessions"]}
|
|
return {"created_containers": [v for k, v in new.items() if k not in old],
|
|
"removed_containers": [v for k, v in old.items() if k not in new],
|
|
"changed_containers": [v for k, v in new.items() if k in old and v["State"] != old[k]["State"]],
|
|
"new_tmux_sessions": [v for v in after["tmux_sessions"] if v["id"] not in sessions],
|
|
"note": "共享宿主前后快照差异不等于资源归属证明；不会自动删除这些容器或会话。"}
|