Added import class.

ptlinux/import_state.py

@@ -0,0 +1,85 @@
+"""Persistent trust state for game trees seeded from another installation."""
+
+from __future__ import annotations
+
+from datetime import datetime
+import json
+import os
+from pathlib import Path
+from uuid import uuid4
+
+from .paths import AppPaths
+
+
+IMPORT_STATE_SCHEMA = 2
+
+
+def game_import_requires_verification(paths: AppPaths) -> bool:
+	"""Return whether a managed imported tree still needs a full depot repair."""
+
+	manifest = paths.game_import_manifest
+	if not manifest.exists():
+		return False
+	try:
+		value = json.loads(manifest.read_text(encoding="utf-8"))
+	except (OSError, UnicodeError, json.JSONDecodeError):
+		return True
+	if not isinstance(value, dict):
+		return True
+	build = value.get("verification_build")
+	completed_at = value.get("verification_completed_at")
+	return not (
+		value.get("verification_required") is False
+		and isinstance(build, int)
+		and not isinstance(build, bool)
+		and build > 0
+		and isinstance(completed_at, str)
+		and bool(completed_at.strip())
+	)
+
+
+def mark_game_import_verified(
+	paths: AppPaths,
+	*,
+	build: int,
+	files_quarantined: int,
+	quarantine_directory: str | None,
+) -> None:
+	"""Record a completed full-manifest verification of an imported tree."""
+
+	if isinstance(build, bool) or build <= 0:
+		raise ValueError("verified import build must be positive")
+	if files_quarantined < 0:
+		raise ValueError("quarantined file count cannot be negative")
+	try:
+		value = json.loads(paths.game_import_manifest.read_text(encoding="utf-8"))
+	except (OSError, UnicodeError, json.JSONDecodeError):
+		value = {}
+	if not isinstance(value, dict):
+		value = {}
+	value.update(
+		{
+			"schema": IMPORT_STATE_SCHEMA,
+			"verification_required": False,
+			"verification_build": build,
+			"verification_completed_at": datetime.now().astimezone().isoformat(),
+			"files_quarantined": files_quarantined,
+			"quarantine_directory": quarantine_directory,
+		}
+	)
+	_write_json(paths.game_import_manifest, value)
+
+
+def _write_json(path: Path, value: object) -> None:
+	path.parent.mkdir(parents=True, exist_ok=True)
+	temporary = path.with_name(f".{path.name}.{os.getpid()}.{uuid4().hex}.tmp")
+	try:
+		with temporary.open("x", encoding="utf-8") as handle:
+			json.dump(value, handle, indent=2, sort_keys=True)
+			handle.write("\n")
+			handle.flush()
+			os.fsync(handle.fileno())
+		temporary.replace(path)
+	except BaseException:
+		temporary.unlink(missing_ok=True)
+		raise