#!/usr/bin/env python3 """Track source changes without repeatedly reprocessing unchanged files.""" from __future__ import annotations import argparse import hashlib import json import os import tempfile from datetime import datetime, timezone from pathlib import Path, PurePosixPath from typing import Any MANIFEST_VERSION = 1 HASH_ALGORITHM = "sha256" IGNORED_NAMES = {".DS_Store", ".gitkeep"} VALID_STATUSES = {"pending", "processed", "missing", "removed"} def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Scan sources and track their wiki-processing state." ) subparsers = parser.add_subparsers(dest="command", required=True) scan = subparsers.add_parser("scan", help="Detect new, changed, and missing sources") scan.add_argument("project_root", nargs="?", type=Path, default=Path(".")) scan.add_argument( "--full", action="store_true", help="Recompute every hash instead of trusting unchanged size and mtime", ) scan.add_argument("--json", action="store_true", help="Emit machine-readable output") status = subparsers.add_parser("status", help="Show the current manifest state") status.add_argument("project_root", nargs="?", type=Path, default=Path(".")) status.add_argument("--json", action="store_true", help="Emit machine-readable output") mark = subparsers.add_parser( "mark-processed", help="Mark successfully integrated sources as processed", ) mark.add_argument("project_root", nargs="?", type=Path, default=Path(".")) mark.add_argument("paths", nargs="*", help="Paths relative to sources/") mark.add_argument( "--all-pending", action="store_true", help="Mark every currently pending source", ) mark.add_argument( "--processor-version", help="Override the skill version recorded for this processing run", ) mark.add_argument("--json", action="store_true", help="Emit machine-readable output") remove = subparsers.add_parser( "mark-removed", help="Acknowledge a fully integrated source deletion", ) remove.add_argument("project_root", nargs="?", type=Path, default=Path(".")) remove.add_argument("paths", nargs="+", help="Missing paths relative to sources/") remove.add_argument( "--reason", required=True, help="Why deletion is safe and fully reflected in the wiki", ) remove.add_argument( "--replacement", action="append", default=[], help="Retained source path that replaces the deleted source; repeat as needed", ) remove.add_argument( "--processor-version", help="Override the skill version recorded for this processing run", ) remove.add_argument("--json", action="store_true", help="Emit machine-readable output") return parser.parse_args() def project_paths(project_root: Path) -> tuple[Path, Path]: root = project_root.expanduser().resolve() return root / "sources", root / "wiki" / ".source-manifest.json" def empty_manifest() -> dict[str, Any]: return { "version": MANIFEST_VERSION, "hash_algorithm": HASH_ALGORITHM, "files": {}, } def validate_relative_path(value: str) -> str: path = PurePosixPath(value) if not value or path.is_absolute() or ".." in path.parts: raise ValueError(f"Unsafe source path in manifest: {value!r}") return path.as_posix() def load_manifest(path: Path, *, allow_missing: bool = False) -> dict[str, Any]: if not path.exists(): if allow_missing: return empty_manifest() raise SystemExit(f"Source manifest does not exist: {path}") try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise SystemExit(f"Cannot read source manifest {path}: {error}") from error if not isinstance(data, dict): raise SystemExit(f"Source manifest must contain a JSON object: {path}") if data.get("version") != MANIFEST_VERSION: raise SystemExit( f"Unsupported source manifest version: {data.get('version')!r}" ) if data.get("hash_algorithm") != HASH_ALGORITHM: raise SystemExit( f"Unsupported hash algorithm: {data.get('hash_algorithm')!r}" ) if not isinstance(data.get("files"), dict): raise SystemExit("Source manifest field 'files' must be a JSON object") for relative, entry in data["files"].items(): try: validate_relative_path(relative) except ValueError as error: raise SystemExit(str(error)) from error if not isinstance(entry, dict): raise SystemExit(f"Manifest entry must be an object: {relative}") if entry.get("status") not in VALID_STATUSES: raise SystemExit(f"Invalid source status for {relative}: {entry.get('status')!r}") return data def write_manifest(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) rendered = json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n" with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, delete=False ) as handle: handle.write(rendered) temporary = Path(handle.name) os.replace(temporary, path) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def source_files(source_root: Path) -> tuple[dict[str, Path], list[str]]: files: dict[str, Path] = {} warnings: list[str] = [] if not source_root.is_dir(): return files, [f"Source directory does not exist: {source_root}"] for path in sorted(source_root.rglob("*")): relative = path.relative_to(source_root) if ".git" in relative.parts or path.name in IGNORED_NAMES: continue if path.is_symlink(): warnings.append(f"Ignored symbolic link: {relative.as_posix()}") continue if path.is_file(): files[relative.as_posix()] = path return files, warnings def current_hash( path: Path, old: dict[str, Any] | None, *, full: bool ) -> tuple[str, os.stat_result]: stat = path.stat() if ( not full and old and old.get("sha256") and old.get("size") == stat.st_size and old.get("mtime_ns") == str(stat.st_mtime_ns) ): return str(old["sha256"]), stat return sha256(path), stat def scan_sources(project_root: Path, *, full: bool) -> dict[str, Any]: source_root, manifest_path = project_paths(project_root) manifest = load_manifest(manifest_path, allow_missing=True) old_files: dict[str, dict[str, Any]] = manifest["files"] discovered, warnings = source_files(source_root) now = utc_now() present_paths = set(discovered) missing_paths = set(old_files) - present_paths missing_hashes: dict[str, list[str]] = {} for relative in missing_paths: old = old_files[relative] known_hash = old.get("processed_sha256") or old.get("sha256") if known_hash: missing_hashes.setdefault(str(known_hash), []).append(relative) new_files: dict[str, dict[str, Any]] = {} expected_processor = processor_version(None) summary = { "pending": [], "processed": [], "missing": [], "removed": [], "outdated_processor": [], "warnings": warnings, } for relative, path in discovered.items(): old = old_files.get(relative) digest, stat = current_hash(path, old, full=full) processed_hash = old.get("processed_sha256") if old else None status = ( "processed" if processed_hash == digest and old.get("status") != "removed" else "pending" ) entry: dict[str, Any] = { "mtime_ns": str(stat.st_mtime_ns), "sha256": digest, "size": stat.st_size, "status": status, } if processed_hash: entry["processed_sha256"] = processed_hash if old: for key in ("processed_at", "processor_version", "previous_path"): if key in old: entry[key] = old[key] elif len(missing_hashes.get(digest, [])) == 1: entry["previous_path"] = missing_hashes[digest][0] new_files[relative] = entry summary[status].append(relative) if ( status == "processed" and entry.get("processor_version") != expected_processor ): summary["outdated_processor"].append(relative) for relative in sorted(missing_paths): entry = dict(old_files[relative]) if entry.get("status") == "removed": new_files[relative] = entry summary["removed"].append(relative) continue if entry.get("status") != "missing": entry["missing_since"] = now entry["status"] = "missing" new_files[relative] = entry summary["missing"].append(relative) manifest["files"] = dict(sorted(new_files.items())) write_manifest(manifest_path, manifest) return summary def processor_version(override: str | None) -> str: if override: return override version_path = Path(__file__).resolve().parent.parent / "VERSION" version = version_path.read_text(encoding="utf-8").strip() if not version: raise SystemExit(f"Skill version is empty: {version_path}") return f"maintain-knowledge-wiki/{version}" def normalize_requested_path(raw: str, source_root: Path) -> str: path = Path(raw) if path.is_absolute(): try: return path.resolve().relative_to(source_root.resolve()).as_posix() except ValueError as error: raise SystemExit(f"Source is outside sources/: {raw}") from error parts = PurePosixPath(raw).parts if parts and parts[0] == "sources": raw = PurePosixPath(*parts[1:]).as_posix() try: return validate_relative_path(raw) except ValueError as error: raise SystemExit(str(error)) from error def mark_processed( project_root: Path, requested_paths: list[str], *, all_pending: bool, version_override: str | None, ) -> dict[str, Any]: source_root, manifest_path = project_paths(project_root) manifest = load_manifest(manifest_path) files: dict[str, dict[str, Any]] = manifest["files"] if all_pending and requested_paths: raise SystemExit("Use either explicit paths or --all-pending, not both") if all_pending: selected = sorted( relative for relative, entry in files.items() if entry.get("status") == "pending" ) else: selected = [normalize_requested_path(path, source_root) for path in requested_paths] if not selected: raise SystemExit("No pending or explicitly selected sources to mark") verified: list[tuple[str, str, os.stat_result]] = [] for relative in selected: entry = files.get(relative) if entry is None: raise SystemExit(f"Source is not present in the manifest: {relative}") if entry.get("status") == "missing": raise SystemExit(f"Cannot mark missing source as processed: {relative}") path = source_root / PurePosixPath(relative) if not path.is_file() or path.is_symlink(): raise SystemExit(f"Source is missing or not a regular file: {relative}") digest, stat = current_hash(path, entry, full=True) if digest != entry.get("sha256"): raise SystemExit( f"Source changed after the last scan: {relative}. Run scan again." ) verified.append((relative, digest, stat)) now = utc_now() version = processor_version(version_override) for relative, digest, stat in verified: entry = files[relative] entry.update( { "mtime_ns": str(stat.st_mtime_ns), "processed_at": now, "processed_sha256": digest, "processor_version": version, "sha256": digest, "size": stat.st_size, "status": "processed", } ) entry.pop("missing_since", None) entry.pop("removed_at", None) entry.pop("removed_reason", None) entry.pop("replaced_by", None) write_manifest(manifest_path, manifest) return { "marked_processed": [relative for relative, _, _ in verified], "processor_version": version, } def mark_removed( project_root: Path, requested_paths: list[str], *, reason: str, replacement_paths: list[str], version_override: str | None, ) -> dict[str, Any]: source_root, manifest_path = project_paths(project_root) manifest = load_manifest(manifest_path) files: dict[str, dict[str, Any]] = manifest["files"] selected = [normalize_requested_path(path, source_root) for path in requested_paths] replacements = [ normalize_requested_path(path, source_root) for path in replacement_paths ] normalized_reason = reason.strip() if not normalized_reason: raise SystemExit("Removal reason must not be empty") for relative in selected: path = PurePosixPath(relative) entry = files.get(relative) if entry is None: raise SystemExit(f"Source is not present in the manifest: {relative}") if entry.get("status") != "missing": raise SystemExit(f"Source is not missing: {relative}") if (source_root / path).exists(): raise SystemExit(f"Source exists; run scan again: {relative}") for relative in replacements: if relative in selected: raise SystemExit(f"Removed source cannot replace itself: {relative}") entry = files.get(relative) if entry is None: raise SystemExit(f"Replacement is not present in the manifest: {relative}") if entry.get("status") != "processed": raise SystemExit(f"Replacement is not processed: {relative}") path = source_root / PurePosixPath(relative) if not path.is_file() or path.is_symlink(): raise SystemExit(f"Replacement is missing or not a regular file: {relative}") now = utc_now() version = processor_version(version_override) for relative in selected: entry = files[relative] entry.update( { "status": "removed", "removed_at": now, "removed_reason": normalized_reason, "processor_version": version, } ) if replacements: entry["replaced_by"] = sorted(set(replacements)) else: entry.pop("replaced_by", None) write_manifest(manifest_path, manifest) return { "marked_removed": selected, "reason": normalized_reason, "replaced_by": sorted(set(replacements)), "processor_version": version, } def manifest_status(project_root: Path) -> dict[str, Any]: _, manifest_path = project_paths(project_root) manifest = load_manifest(manifest_path) grouped = {status: [] for status in sorted(VALID_STATUSES)} grouped["outdated_processor"] = [] expected_processor = processor_version(None) for relative, entry in manifest["files"].items(): grouped[entry["status"]].append(relative) if ( entry["status"] == "processed" and entry.get("processor_version") != expected_processor ): grouped["outdated_processor"].append(relative) return grouped def print_result(result: dict[str, Any], *, as_json: bool) -> None: if as_json: print(json.dumps(result, indent=2, sort_keys=True, ensure_ascii=False)) return for key, values in result.items(): if isinstance(values, list): if key == "processed": print(f"processed: {len(values)}") continue for value in values: print(f"{key}: {value}") else: print(f"{key}: {values}") if all(not values for values in result.values() if isinstance(values, list)): print("No source entries.") def main() -> int: args = parse_args() if args.command == "scan": result = scan_sources(args.project_root, full=args.full) elif args.command == "status": result = manifest_status(args.project_root) elif args.command == "mark-processed": result = mark_processed( args.project_root, args.paths, all_pending=args.all_pending, version_override=args.processor_version, ) else: result = mark_removed( args.project_root, args.paths, reason=args.reason, replacement_paths=args.replacement, version_override=args.processor_version, ) print_result(result, as_json=args.json) return 0 if __name__ == "__main__": raise SystemExit(main())