#!/usr/bin/env python3 """Return the distilled TRAPPIST-1 payload, refreshing the local cache when stale. This script is intended to sit behind a thin web/API layer that the Unity client calls. It does not expose arbitrary commands. Its job is narrow: 1. Check whether the distilled cache exists and is still fresh. 2. If the cache is missing or expired, refresh it from NASA through the local fetch script. 3. Return the distilled cache JSON to stdout. The cache-age policy is configurable. The initial default is 168 hours. """ from __future__ import annotations import argparse import fcntl import json import subprocess import sys import time from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any DEFAULT_CACHE_MAX_AGE_HOURS = 168.0 DEFAULT_DATA_DIR = Path(__file__).resolve().parent DEFAULT_FETCH_SCRIPT = DEFAULT_DATA_DIR / "query_nasa_db.py" CANONICAL_FILENAME = "canonical-system.json" RAW_FILENAME = "raw-pscomppars.json" REFRESH_LOCK_FILENAME = "refresh.lock" UTC = timezone.utc @dataclass(frozen=True) class RequestConfig: """Configuration for returning the distilled TRAPPIST-1 payload.""" data_dir: Path fetch_script: Path hostname: str cache_max_age_hours: float timeout_seconds: float pretty: bool force_refresh: bool def parse_args() -> RequestConfig: """Parse command-line arguments.""" parser = argparse.ArgumentParser( description=( "Return the distilled TRAPPIST-1 payload, refreshing the local " "NASA-derived cache when it is missing or stale." ) ) parser.add_argument( "--data-dir", default=str(DEFAULT_DATA_DIR), help="Directory holding the distilled and raw cache files.", ) parser.add_argument( "--fetch-script", default=str(DEFAULT_FETCH_SCRIPT), help="Local script used to refresh the NASA-derived cache.", ) parser.add_argument( "--hostname", default="TRAPPIST-1", help="Host star name to pass to the refresh script.", ) parser.add_argument( "--cache-max-age-hours", type=float, default=DEFAULT_CACHE_MAX_AGE_HOURS, help="Maximum age of the distilled cache before it is refreshed.", ) parser.add_argument( "--timeout-seconds", type=float, default=30.0, help="Timeout passed through to the NASA refresh script.", ) parser.add_argument( "--pretty", action="store_true", help="Pretty-print the JSON response.", ) parser.add_argument( "--force-refresh", action="store_true", help="Refresh even if the local cache is still fresh.", ) args = parser.parse_args() return RequestConfig( data_dir=Path(args.data_dir), fetch_script=Path(args.fetch_script), hostname=args.hostname, cache_max_age_hours=args.cache_max_age_hours, timeout_seconds=args.timeout_seconds, pretty=args.pretty, force_refresh=args.force_refresh, ) def now_utc_iso() -> str: """Return the current UTC time in ISO 8601 format.""" return datetime.now(UTC).replace(microsecond=0).isoformat() def load_json(path: Path) -> Any: """Load a JSON file from disk.""" return json.loads(path.read_text(encoding="utf-8")) def emit_json(payload: dict[str, Any], pretty: bool) -> None: """Emit a JSON payload to stdout.""" if pretty: sys.stdout.write(json.dumps(payload, indent=2, sort_keys=True) + "\n") else: sys.stdout.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n") def cache_paths(data_dir: Path) -> tuple[Path, Path]: """Return the distilled and raw cache file paths.""" return data_dir / CANONICAL_FILENAME, data_dir / RAW_FILENAME @contextmanager def refresh_lock(data_dir: Path): """Hold an exclusive process lock while checking or refreshing the cache. The public PHP endpoint can receive overlapping requests. Without a lock, several requests arriving just after cache expiry could all conclude that the cache is stale and then all query NASA. The lock makes the stale-cache path single-file: one process refreshes, and the others re-check the cache after the lock is released. """ data_dir.mkdir(parents=True, exist_ok=True) lock_path = data_dir / REFRESH_LOCK_FILENAME with lock_path.open("w", encoding="utf-8") as lock_file: fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) try: yield finally: fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) def file_age_seconds(path: Path) -> float: """Return the age of a file in seconds.""" return time.time() - path.stat().st_mtime def cache_is_fresh(canonical_path: Path, raw_path: Path, max_age_hours: float) -> bool: """Return whether both cache files exist and are younger than the max age.""" if not canonical_path.exists() or not raw_path.exists(): return False max_age_seconds = max_age_hours * 3600.0 return ( file_age_seconds(canonical_path) <= max_age_seconds and file_age_seconds(raw_path) <= max_age_seconds ) def refresh_cache(config: RequestConfig) -> dict[str, Any]: """Refresh the local cache through the dedicated fetch script.""" command = [ sys.executable, str(config.fetch_script), "--hostname", config.hostname, "--output-dir", str(config.data_dir), "--timeout-seconds", str(config.timeout_seconds), ] completed = subprocess.run( command, check=False, capture_output=True, text=True, ) if completed.returncode != 0: detail = completed.stderr.strip() or completed.stdout.strip() or ( f"Refresh script exited with code {completed.returncode}." ) raise RuntimeError(detail) return { "refreshed": True, "refresh_stdout": completed.stdout.strip(), } def build_response( config: RequestConfig, payload: dict[str, Any], refreshed: bool, canonical_path: Path, ) -> dict[str, Any]: """Build the JSON response envelope.""" return { "ok": True, "generated_at_utc": now_utc_iso(), "cache": { "data_dir": str(config.data_dir), "canonical_file": str(canonical_path), "cache_max_age_hours": config.cache_max_age_hours, "refreshed": refreshed, "canonical_age_seconds": file_age_seconds(canonical_path), }, "data": payload, } def build_error_response(message: str) -> dict[str, Any]: """Build a JSON error envelope.""" return { "ok": False, "generated_at_utc": now_utc_iso(), "error": message, } def main() -> int: """Serve the distilled TRAPPIST-1 payload through stdout.""" config = parse_args() canonical_path, raw_path = cache_paths(config.data_dir) try: refreshed = False if config.force_refresh or not cache_is_fresh( canonical_path, raw_path, config.cache_max_age_hours, ): with refresh_lock(config.data_dir): if config.force_refresh or not cache_is_fresh( canonical_path, raw_path, config.cache_max_age_hours, ): refresh_cache(config) refreshed = True payload = load_json(canonical_path) emit_json( build_response(config, payload, refreshed, canonical_path), config.pretty, ) return 0 except Exception as exc: # noqa: BLE001 emit_json(build_error_response(str(exc)), config.pretty) return 1 if __name__ == "__main__": raise SystemExit(main())