#!/usr/bin/env python3 """Wisdom Exposed local agent: metadata indexing and explicitly approved disk operations. Standard library only. Start with --doctor and --dry-run before using --apply. """ from __future__ import annotations import argparse import hashlib import json import mimetypes import os from pathlib import Path import shutil import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime, timezone from typing import Any CHUNK = 1024 * 1024 SKIP_DIRS = {'.git', 'node_modules', '.next', '.wisdom-exposed-trash', '.sidecar-trash', '__pycache__'} def request_json(server: str, path: str, token: str, payload: Any = None, method: str | None = None, timeout: int = 30) -> Any: data = None if payload is None else json.dumps(payload).encode('utf-8') request = urllib.request.Request( server.rstrip('/') + path, data=data, method=method or ('POST' if data is not None else 'GET'), headers={'Content-Type': 'application/json', 'Accept': 'application/json', 'X-Wisdom-Exposed-Token': token}, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode('utf-8')) except urllib.error.HTTPError as error: detail = error.read().decode('utf-8', errors='replace')[:1000] raise RuntimeError(f'Wisdom Exposed returned HTTP {error.code}: {detail}') from error except urllib.error.URLError as error: raise RuntimeError(f'Cannot reach Wisdom Exposed at {server}: {error.reason}') from error def file_hash(path: Path, fast: bool = False) -> str: digest = hashlib.sha256() remaining = CHUNK if fast else None with path.open('rb') as handle: while True: size = CHUNK if remaining is None else min(CHUNK, remaining) if size <= 0: break block = handle.read(size) if not block: break digest.update(block) if remaining is not None: remaining -= len(block) if remaining <= 0: break return digest.hexdigest() def scan_roots(roots: list[Path], fast: bool = False) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for root in roots: for current, dirs, names in os.walk(root, followlinks=False): dirs[:] = [name for name in dirs if name not in SKIP_DIRS and not name.startswith('.Trash')] for name in names: path = Path(current) / name try: if path.is_symlink() or not path.is_file(): continue stat = path.stat() records.append({ 'path': str(path), 'name': path.name, 'folder': str(path.parent), 'size': stat.st_size, 'sha256': file_hash(path, fast), 'modified': datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(), 'type': mimetypes.guess_type(path.name)[0] or 'application/octet-stream', }) except (OSError, PermissionError) as error: print(f'skip: {path} ({error})', file=sys.stderr) return records def collision_safe(destination: Path) -> Path: if not destination.exists(): return destination for index in range(1, 10000): candidate = destination.with_name(f'{destination.stem} ({index}){destination.suffix}') if not candidate.exists(): return candidate raise RuntimeError(f'Could not find a collision-safe name for {destination}') def inside(path: Path, root: Path) -> bool: try: path.resolve().relative_to(root.resolve()) return True except ValueError: return False def apply_operation(operation: dict[str, Any], apply_root: Path, trash: Path, allow_purge: bool = False) -> str: kind = operation.get('kind') or operation.get('type') source = Path(operation.get('source', '')).expanduser() legacy_trash = Path.home() / '.sidecar-trash' source_in_quarantine = inside(source, trash) or inside(source, legacy_trash) if not source.is_absolute() or (not inside(source, apply_root) and not (kind == 'move' and source_in_quarantine)): raise RuntimeError(f'Source is outside --apply-root and approved quarantine: {source}') if kind == 'move': destination = Path(operation.get('destination', '')).expanduser() if not destination.is_absolute() or not inside(destination, apply_root): raise RuntimeError(f'Destination is outside --apply-root: {destination}') destination.parent.mkdir(parents=True, exist_ok=True) destination = collision_safe(destination) shutil.move(str(source), str(destination)) return f'moved to {destination}' if kind == 'trash': trash.mkdir(parents=True, exist_ok=True) destination = collision_safe(trash / source.name) shutil.move(str(source), str(destination)) return f'quarantined at {destination}' if kind == 'purge': if not allow_purge: raise RuntimeError('Purge requires the additional --allow-purge flag') if not inside(source, trash) and not inside(source, legacy_trash): raise RuntimeError('Purge is restricted to Wisdom Exposed quarantine folders') source.unlink() return 'purged from quarantine' raise RuntimeError(f'Unsupported operation: {kind}') def doctor(args: argparse.Namespace, roots: list[Path]) -> bool: ready = True print(f'Python: {sys.version.split()[0]} — ready' if sys.version_info >= (3, 9) else 'Python 3.9+ is required') ready &= sys.version_info >= (3, 9) try: health = request_json(args.server, '/api/health', args.token or '', timeout=6) print(f'Server: reachable ({health.get("mode", "ok")})') except Exception as error: print(f'Server: {error}') ready = False if not args.token: print('Token: missing — set WISDOM_EXPOSED_TOKEN or pass --token') ready = False for root in roots: readable = root.is_dir() and os.access(root, os.R_OK) print(f'Root {root}: {"readable" if readable else "missing or unreadable"}') ready &= readable if args.apply: apply_root = Path(args.apply_root).expanduser() writable = apply_root.is_dir() and os.access(apply_root, os.W_OK) print(f'Apply root {apply_root}: {"writable" if writable else "not writable"}') ready &= writable free = shutil.disk_usage(str(roots[0])).free if roots and roots[0].exists() else 0 print(f'Free disk space: {free / (1024 ** 3):.1f} GB') print(f'ffmpeg: {shutil.which("ffmpeg") or "not found (optional video frames disabled)"}') print('READY' if ready else 'NOT READY — fix the items above') return ready def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('--root', action='append', default=[], help='Folder to index; repeat for multiple roots') parser.add_argument('--server', default=os.getenv('WISDOM_EXPOSED_SERVER') or os.getenv('SIDECAR_SERVER') or 'http://localhost:3000') parser.add_argument('--token', default=os.getenv('WISDOM_EXPOSED_TOKEN') or os.getenv('SIDECAR_TOKEN') or '') parser.add_argument('--label', default=os.getenv('HOSTNAME', 'Local computer')) parser.add_argument('--fast', action='store_true', help='Hash only the first 1 MB (not a full-file duplicate proof)') parser.add_argument('--dry-run', action='store_true', help='Index and preview operations without touching disk') parser.add_argument('--apply', action='store_true', help='Apply operations already approved in Wisdom Exposed') parser.add_argument('--apply-root', default=str(Path.home())) parser.add_argument('--allow-purge', action='store_true', help='Allow explicit purge operations inside Wisdom Exposed trash') parser.add_argument('--doctor', action='store_true') parser.add_argument('--interval', type=int, default=0, help='Repeat every N seconds; 0 runs once') args = parser.parse_args() roots = [Path(item).expanduser() for item in args.root] if not roots: parser.error('at least one --root is required') if args.doctor: return 0 if doctor(args, roots) else 2 if not args.token: parser.error('missing token; set WISDOM_EXPOSED_TOKEN or use --token') if args.apply and args.dry_run: parser.error('choose either --apply or --dry-run, not both') def cycle() -> None: records = scan_roots(roots, args.fast) print(f'Indexed {len(records)} files; sending metadata only…') result = request_json(args.server, '/api/agent/scan', args.token, { 'label': args.label, 'roots': [str(root) for root in roots], 'hashMode': 'first-1mb' if args.fast else 'full-sha256', 'scannedAt': datetime.now(timezone.utc).isoformat(), 'files': records, }, timeout=120) print(f'Server accepted scan: {result.get("indexed", len(records))} records') if args.apply or args.dry_run: queue = request_json(args.server, '/api/agent/operations', args.token, {'action': 'preview' if args.dry_run else 'claim', 'label': args.label}) operations = queue.get('operations', []) print(f'{len(operations)} approved operations in queue') for operation in operations: try: if args.dry_run: print(f'DRY RUN: {operation.get("kind")}: {operation.get("source")} → {operation.get("destination", "~/.wisdom-exposed-trash")}') continue detail = apply_operation(operation, Path(args.apply_root).expanduser(), Path.home() / '.wisdom-exposed-trash', args.allow_purge) status, output = 'complete', detail except Exception as error: status, output = 'failed', str(error) print(f'FAILED: {output}', file=sys.stderr) if not args.dry_run: request_json(args.server, '/api/agent/operations', args.token, {'action': 'report', 'id': operation.get('id'), 'status': status, 'output': output}) while True: cycle() if args.interval <= 0: break time.sleep(max(30, args.interval)) return 0 if __name__ == '__main__': try: raise SystemExit(main()) except KeyboardInterrupt: print('\nStopped by user') raise SystemExit(130)