"""Command-line interface for offline checks and explicit local shadow runs.""" from __future__ import annotations import argparse import importlib import importlib.util import json import os import platform import sys import tempfile from pathlib import Path from typing import Sequence from . import __version__ from .config import PROHIBITED_CAPABILITIES, AppConfig from .domain.models import ProfileVersion from .errors import ApplicationError, InvalidConfiguration from .normalization.models import NormalizationContext from .pbf.models import PbfParseLimits, VerifiedPbfInput from .pbf.parser import OsmiumPbfParser from .profiles import ProfileEvaluator, builtin_profile_registry from .shadow import run_profile_shadow _PACKAGE_NAME = "osm_lead_source_service" _PROHIBITED_MODULES = ( "osm_lead_source_service.production_writer", "osm_lead_source_service.noco_write", "osm_lead_source_service.delete_service", "osm_lead_source_service.stale_deleter", "osm_lead_source_service.scheduler", "osm_lead_source_service.ports.noco_write", ) def build_parser() -> argparse.ArgumentParser: """Build the CLI parser without performing any I/O.""" parser = argparse.ArgumentParser(prog="osm-lead-source") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("version", help="print the package version") subparsers.add_parser("validate-config", help="validate offline configuration") doctor = subparsers.add_parser("doctor", help="run offline foundation checks") doctor.add_argument( "--offline", action="store_true", required=True, help="required safety acknowledgment; no network or database access is allowed", ) doctor.add_argument( "--json", action="store_true", help="emit deterministic machine-readable JSON", ) shadow = subparsers.add_parser( "shadow-run", help="parse one verified local PBF and emit a read-only profile report", ) shadow.add_argument("--pbf", type=Path, required=True, help="verified local .osm.pbf path") shadow.add_argument("--sha256", required=True, help="expected lowercase SHA-256") shadow.add_argument("--size-bytes", type=int, required=True, help="expected exact byte count") shadow.add_argument("--region", required=True, help="stable region identifier") shadow.add_argument("--profile", required=True, help="exact built-in profile identifier") shadow.add_argument("--profile-version", required=True, help="exact built-in profile version") shadow.add_argument("--country-code", required=True, help="ISO alpha-2 country code") shadow.add_argument("--calling-code", help="optional international calling code") shadow.add_argument( "--national-trunk-prefix", default="0", help="national phone trunk prefix (default: 0)", ) shadow.add_argument( "--output", type=Path, required=True, help="JSON output path, or - for standard output", ) shadow.add_argument( "--max-file-size-bytes", type=int, default=8 * 1024 * 1024, help="hard input file limit for this run", ) shadow.add_argument( "--max-objects", type=int, default=100_000, help="hard parsed-object limit for this run", ) shadow.add_argument( "--max-total-tags", type=int, default=500_000, help="hard total-tag limit for this run", ) shadow.add_argument( "--max-blob-size-bytes", type=int, default=32 * 1024 * 1024, help="hard compressed size limit for each PBF blob", ) shadow.add_argument( "--max-uncompressed-blob-bytes", type=int, default=32 * 1024 * 1024, help="hard uncompressed size limit for each PBF blob", ) shadow.add_argument( "--max-total-uncompressed-bytes", type=int, default=64 * 1024 * 1024, help="hard cumulative uncompressed PBF payload limit for this run", ) shadow.add_argument( "--max-blob-count", type=int, default=1024, help="hard PBF blob-count limit for this run", ) shadow.add_argument( "--max-leads", type=int, default=100_000, help="hard active-lead output limit for this run", ) shadow.add_argument( "--max-review-items", type=int, default=100_000, help="hard pending-review output limit for this run", ) shadow.add_argument( "--overwrite", action="store_true", help="replace an existing regular output file atomically", ) acquire = subparsers.add_parser( "geofabrik-acquire", help="preview or acquire one approved Geofabrik pilot extract", ) acquire.add_argument( "--region", default="au-act", help="approved pilot region identifier (currently only au-act)", ) acquire.add_argument( "--cache-dir", type=Path, required=True, help="content-addressed PBF cache directory", ) acquire.add_argument( "--timeout-seconds", type=float, default=120.0, help="bounded HTTPS request timeout", ) acquire.add_argument( "--execute", action="store_true", help="perform approved Geofabrik HTTPS requests; otherwise preview only", ) pilot = subparsers.add_parser( "geofabrik-shadow-pilot", help="preview or run the read-only ACT acquisition and profile pipeline", ) pilot.add_argument( "--cache-dir", type=Path, required=True, help="content-addressed PBF cache directory", ) pilot.add_argument( "--output", type=Path, required=True, help="shadow report JSON path", ) pilot.add_argument( "--evidence-output", type=Path, required=True, help="acquisition and run-evidence JSON path", ) pilot.add_argument( "--timeout-seconds", type=float, default=120.0, help="bounded HTTPS timeout", ) pilot.add_argument( "--max-leads", type=int, default=100_000, help="hard active-lead output limit", ) pilot.add_argument( "--max-review-items", type=int, default=100_000, help="hard pending-review output limit", ) pilot.add_argument( "--overwrite", action="store_true", help="replace existing regular output files atomically", ) pilot.add_argument( "--execute", action="store_true", help="perform approved Geofabrik HTTPS and local parsing; never accesses NocoDB", ) market_run = subparsers.add_parser( "market-run-once", help="preview or execute exactly one configured read-only multi-market cycle", ) market_run.add_argument( "--execute", action="store_true", help="perform Geofabrik network, download, extraction, parsing, and report work", ) insert = subparsers.add_parser( "insert-run", help="execute approved insert-only NocoDB writes from one shadow report", ) insert.add_argument("--report", type=Path, required=True, help="Phase 2A.6 JSON report") insert.add_argument("--report-hash", required=True, help="expected canonical report SHA-256") insert.add_argument("--approval-id", required=True, help="active sidecar cutover approval") insert.add_argument("--noco-table-id", required=True, help="exact NocoDB destination table ID") insert.add_argument("--run-id", required=True, help="stable operational run identifier") insert.add_argument("--run-timestamp", required=True, help="timezone-aware ISO-8601 timestamp") insert.add_argument( "--max-report-bytes", type=int, default=64 * 1024 * 1024, help="hard JSON report size limit", ) insert.add_argument( "--max-candidates", type=int, default=100_000, help="hard candidate count limit", ) insert.add_argument( "--max-inserts", type=int, default=100, help="hard insert cap; must not exceed the database approval", ) insert.add_argument( "--execute", action="store_true", help="required acknowledgment that this command may create NocoDB rows", ) service = subparsers.add_parser( "serve", help="serve read-only health, readiness, and status endpoints", ) service.add_argument("--host", default="0.0.0.0", help="HTTP bind host") service.add_argument("--port", type=int, default=8080, help="HTTP bind port") return parser def _production_writer_modules_present() -> tuple[str, ...]: present: list[str] = [] for module_name in _PROHIBITED_MODULES: if importlib.util.find_spec(module_name) is not None: present.append(module_name) return tuple(present) def _doctor_result(config: AppConfig | None, config_error: str | None) -> dict[str, object]: version_info = sys.version_info python_supported = version_info >= (3, 12) importable = importlib.util.find_spec(_PACKAGE_NAME) is not None writer_modules = _production_writer_modules_present() read_only_enforced = config is not None and config.read_only is True no_writer_configured = ( config is not None and all(getattr(config, capability) is False for capability in PROHIBITED_CAPABILITIES) and not writer_modules ) basic_configuration = config is not None and config_error is None checks: dict[str, object] = { "basic_configuration": {"ok": basic_configuration}, "no_production_writer_configured": { "ok": no_writer_configured, "modules": list(writer_modules), }, "package_importability": {"ok": importable, "package": _PACKAGE_NAME}, "python_version_supported": { "minimum": "3.12", "ok": python_supported, "running": platform.python_version(), }, "read_only_enforced": {"ok": read_only_enforced}, } result: dict[str, object] = { "checks": checks, "command": "doctor", "config_error": config_error, "offline": True, "ok": all(isinstance(check, dict) and check.get("ok") is True for check in checks.values()), "phase": "2A.6", } return result def _run_doctor(json_output: bool) -> int: config: AppConfig | None config_error: str | None = None try: config = AppConfig.from_env() except InvalidConfiguration as exc: config = None config_error = str(exc) result = _doctor_result(config, config_error) if json_output: print(json.dumps(result, sort_keys=True, separators=(",", ":"))) else: print("doctor: " + ("ok" if result["ok"] else "failed")) if config_error: print(f"configuration error: {config_error}", file=sys.stderr) return 0 if result["ok"] else 1 def _run_validate_config() -> int: try: AppConfig.from_env() except InvalidConfiguration as exc: print(f"invalid configuration: {exc}", file=sys.stderr) return 2 print("configuration valid") return 0 def _write_report(path: Path, content: str, *, overwrite: bool) -> None: if path == Path("-"): print(content) return destination = path.expanduser() parent = destination.parent if not parent.exists() or not parent.is_dir(): raise InvalidConfiguration("output parent must be an existing directory") if destination.is_symlink(): raise InvalidConfiguration("output path cannot be a symbolic link") if destination.exists(): if not destination.is_file(): raise InvalidConfiguration("output path must be a regular file") if not overwrite: raise InvalidConfiguration("output already exists; pass --overwrite to replace it") payload = (content + "\n").encode("utf-8") fd, temporary_name = tempfile.mkstemp(prefix=f".{destination.name}.", suffix=".tmp", dir=parent) temporary = Path(temporary_name) try: os.chmod(temporary, 0o600) with os.fdopen(fd, "wb") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) if overwrite: os.replace(temporary, destination) else: try: os.link(temporary, destination) except FileExistsError as exc: raise InvalidConfiguration( "output was created concurrently; no file was replaced" ) from exc temporary.unlink() directory_fd = os.open(parent, os.O_RDONLY) try: os.fsync(directory_fd) finally: os.close(directory_fd) except OSError as exc: raise InvalidConfiguration(f"failed to write shadow report: {exc}") from exc finally: if temporary.exists(): temporary.unlink() def _run_shadow(args: argparse.Namespace) -> int: registry = builtin_profile_registry() profile = ProfileVersion(args.profile, args.profile_version) definition = registry.get(profile) context = NormalizationContext( country_code=args.country_code, calling_code=args.calling_code, national_trunk_prefix=args.national_trunk_prefix, ) limits = PbfParseLimits( max_file_size_bytes=args.max_file_size_bytes, max_objects=args.max_objects, max_total_tags=args.max_total_tags, max_blob_size_bytes=args.max_blob_size_bytes, max_uncompressed_blob_bytes=args.max_uncompressed_blob_bytes, max_total_uncompressed_bytes=args.max_total_uncompressed_bytes, max_blob_count=args.max_blob_count, ) approved_input = VerifiedPbfInput( local_path=args.pbf.expanduser(), content_sha256=args.sha256, size_bytes=args.size_bytes, provenance=f"explicit local shadow-run input for {args.region}", ) report = run_profile_shadow( approved_input=approved_input, parser=OsmiumPbfParser(limits), evaluator=ProfileEvaluator(registry), definition=definition, region_id=args.region, normalization_context=context, max_leads=args.max_leads, max_review_items=args.max_review_items, ) _write_report(args.output, report.to_json(), overwrite=args.overwrite) if args.output != Path("-"): print( json.dumps( { "lead_count": len(report.leads), "output": str(args.output), "report_hash": report.report_hash, "review_count": len(report.review_items), }, sort_keys=True, separators=(",", ":"), ) ) return 0 def _run_insert(args: argparse.Namespace) -> int: if args.execute is not True: raise InvalidConfiguration("insert-run requires --execute") # Keep database and network dependencies out of imports and offline doctor. from .writer.runtime import execute_insert_report, summary_json summary = execute_insert_report( report_path=args.report, expected_report_hash=args.report_hash, approval_id=args.approval_id, noco_table_id=args.noco_table_id, max_report_bytes=args.max_report_bytes, max_candidates=args.max_candidates, max_inserts=args.max_inserts, run_id=args.run_id, run_timestamp=args.run_timestamp, ) print(summary_json(summary)) return 0 def _run_geofabrik_acquire(args: argparse.Namespace) -> int: # Keep production-network dependencies out of imports and offline doctor. from .pbf.geofabrik import acquire_region, approved_region region = approved_region(args.region) if args.execute is not True: print( json.dumps( { "cache_dir": str(args.cache_dir.expanduser()), "execute": False, "network_access": False, "region": region.as_mapping(), }, sort_keys=True, separators=(",", ":"), ) ) return 0 evidence = acquire_region( region, args.cache_dir, timeout_seconds=args.timeout_seconds, ) print(json.dumps(evidence.as_mapping(), sort_keys=True, separators=(",", ":"))) return 0 def _run_geofabrik_shadow_pilot(args: argparse.Namespace) -> int: from .pbf.geofabrik import AU_ACT_PILOT output = args.output.expanduser() evidence_output = args.evidence_output.expanduser() if output == evidence_output: raise InvalidConfiguration("pilot report and evidence outputs must differ") if args.execute is not True: print( json.dumps( { "cache_dir": str(args.cache_dir.expanduser()), "evidence_output": str(evidence_output), "execute": False, "network_access": False, "noco_access": False, "output": str(output), "profile": {"id": "luxeillum", "version": "0.1.0-draft"}, "region": AU_ACT_PILOT.as_mapping(), "scheduler_enabled": False, "writes_enabled": False, }, sort_keys=True, separators=(",", ":"), ) ) return 0 from .pbf.pilot import run_act_shadow_pilot result = run_act_shadow_pilot( args.cache_dir, timeout_seconds=args.timeout_seconds, max_leads=args.max_leads, max_review_items=args.max_review_items, ) _write_report(output, result.report.to_json(), overwrite=args.overwrite) evidence = json.dumps( result.evidence_mapping(), sort_keys=True, separators=(",", ":"), ensure_ascii=False, ) _write_report(evidence_output, evidence, overwrite=args.overwrite) print(evidence) return 0 def _run_market_once(args: argparse.Namespace) -> int: """Preview or execute one foreground run of the shared read-only worker.""" from .worker import WorkerConfig, run_worker_once try: config = WorkerConfig.from_env() if args.execute is not True: print( json.dumps( { "cache_dir": str(config.cache_dir), "command": "market-run-once", "countries": [market.country_code for market in config.markets], "dashboard_snapshot_path": str(config.dashboard_snapshot_path), "execute": False, "network_access": False, "noco_access": False, "report_dir": str(config.report_dir), "scheduler_required": False, "scheduler_started": False, "writes_enabled": False, }, sort_keys=True, separators=(",", ":"), ) ) return 0 run_worker_once(config) return 0 except Exception as exc: print( json.dumps( { "command": "market-run-once", "error_code": type(exc).__name__, "event": "read_only_cycle_failed", "scheduler_started": False, "writes_enabled": False, }, sort_keys=True, separators=(",", ":"), ), file=sys.stderr, ) if isinstance(exc, ApplicationError): raise raise InvalidConfiguration("market-run-once failed") from exc def _run_service(args: argparse.Namespace) -> int: from .http_service import serve serve(host=args.host, port=args.port) return 0 def main(argv: Sequence[str] | None = None) -> int: """Run the CLI and return a process exit code.""" parser = build_parser() args = parser.parse_args(argv) try: if args.command == "version": print(__version__) return 0 if args.command == "validate-config": return _run_validate_config() if args.command == "doctor": return _run_doctor(args.json) if args.command == "shadow-run": return _run_shadow(args) if args.command == "geofabrik-acquire": return _run_geofabrik_acquire(args) if args.command == "geofabrik-shadow-pilot": return _run_geofabrik_shadow_pilot(args) if args.command == "market-run-once": return _run_market_once(args) if args.command == "insert-run": return _run_insert(args) if args.command == "serve": return _run_service(args) except ApplicationError as exc: print(f"error: {exc}", file=sys.stderr) return 2 raise InvalidConfiguration(f"unknown command: {args.command}") __all__ = ["build_parser", "main"]