"""Read-only scheduled worker for all configured English-language markets.""" from __future__ import annotations import hashlib import json import os import threading import time from collections import Counter, defaultdict from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Callable, Final from .errors import ( ContractViolation, InvalidConfiguration, PartialCycleError, bounded_contract_reason, ) from .markets import EnglishMarket, parse_market_selection from .pbf.catalog import GeofabrikCatalog, MarketSourceShard from .pbf.market import MarketShardResult, run_market_shard _TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"}) _FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"}) _SNAPSHOT_PATH_ENV: Final = "OSM_LEAD_SOURCE_DASHBOARD_SNAPSHOT_PATH" _CACHE_DIR_ENV: Final = "OSM_LEAD_SOURCE_CACHE_DIR" _REPORT_DIR_ENV: Final = "OSM_LEAD_SOURCE_REPORT_DIR" _COUNTRIES_ENV: Final = "OSM_LEAD_SOURCE_COUNTRIES" _RUN_ON_START_ENV: Final = "OSM_LEAD_SOURCE_RUN_ON_START" _SCHEDULER_ENABLED_ENV: Final = "OSM_LEAD_SOURCE_SCHEDULER_ENABLED" _SCHEDULER_APPROVED_ENV: Final = "OSM_LEAD_SOURCE_SCHEDULER_APPROVED" _SCHEDULE_DAY_ENV: Final = "OSM_LEAD_SOURCE_SCHEDULE_DAY" _SCHEDULE_HOUR_ENV: Final = "OSM_LEAD_SOURCE_SCHEDULE_HOUR_UTC" _TIMEOUT_ENV: Final = "OSM_LEAD_SOURCE_GEOFABRIK_TIMEOUT_SECONDS" _EXTRACTION_TIMEOUT_ENV: Final = "OSM_LEAD_SOURCE_EXTRACTION_TIMEOUT_SECONDS" _MAX_SOURCE_BYTES_ENV: Final = "OSM_LEAD_SOURCE_MAX_SOURCE_BYTES" _MAX_LEADS_ENV: Final = "OSM_LEAD_SOURCE_MAX_LEADS" _MAX_REVIEW_ENV: Final = "OSM_LEAD_SOURCE_MAX_REVIEW_ITEMS" _MAX_SOURCE_BYTES: Final = 4 * 1024 * 1024 * 1024 _MAX_SHARDS: Final = 4096 def _utc_now() -> datetime: return datetime.now(timezone.utc) def _parse_bool(environ: Mapping[str, str], name: str, default: bool) -> bool: raw = environ.get(name) if raw is None: return default normalized = raw.strip().lower() if normalized in _TRUE_VALUES: return True if normalized in _FALSE_VALUES: return False raise InvalidConfiguration(f"{name} must be a boolean") def _bounded_int( environ: Mapping[str, str], name: str, default: int, *, minimum: int, maximum: int, ) -> int: raw = environ.get(name) if raw is None: return default try: value = int(raw) except ValueError: raise InvalidConfiguration(f"{name} must be an integer") from None if not minimum <= value <= maximum: raise InvalidConfiguration(f"{name} is outside the accepted range") return value def _bounded_float( environ: Mapping[str, str], name: str, default: float, *, minimum: float, maximum: float, ) -> float: raw = environ.get(name) if raw is None: return default try: value = float(raw) except ValueError: raise InvalidConfiguration(f"{name} must be numeric") from None if not minimum <= value <= maximum: raise InvalidConfiguration(f"{name} is outside the accepted range") return value def _absolute_path(environ: Mapping[str, str], name: str, default: str) -> Path: raw = environ.get(name, default).strip() if not raw or len(raw) > 1024: raise InvalidConfiguration(f"{name} must be a bounded nonblank path") path = Path(raw).expanduser() if not path.is_absolute(): raise InvalidConfiguration(f"{name} must be an absolute path") return path @dataclass(frozen=True, slots=True) class WorkerConfig: """Validated runtime configuration for the read-only multi-market worker.""" cache_dir: Path report_dir: Path dashboard_snapshot_path: Path run_on_start: bool schedule_day: int schedule_hour_utc: int timeout_seconds: float max_leads: int max_review_items: int markets: tuple[EnglishMarket, ...] = field( default_factory=lambda: parse_market_selection("all") ) extraction_timeout_seconds: float = 7200.0 max_source_bytes: int = _MAX_SOURCE_BYTES def __post_init__(self) -> None: if not self.markets or len(self.markets) > 58: raise InvalidConfiguration("worker market selection is outside bounds") if len({market.country_code for market in self.markets}) != len(self.markets): raise InvalidConfiguration("worker market selection contains duplicates") @classmethod def from_env(cls, environ: Mapping[str, str] | None = None) -> "WorkerConfig": source = os.environ if environ is None else environ if not _parse_bool(source, "OSM_LEAD_SOURCE_READ_ONLY", True): raise InvalidConfiguration("read-only worker requires OSM_LEAD_SOURCE_READ_ONLY=true") for variable in ( "OSM_LEAD_SOURCE_NOCO_WRITE_ENABLED", "OSM_LEAD_SOURCE_WRITER_ENABLED", "OSM_LEAD_SOURCE_STALE_DELETION_ENABLED", ): if _parse_bool(source, variable, False): raise InvalidConfiguration( f"read-only worker refuses enabled capability: {variable}" ) return cls( cache_dir=_absolute_path( source, _CACHE_DIR_ENV, "/tmp/osm-lead-source/cache", ), report_dir=_absolute_path( source, _REPORT_DIR_ENV, "/tmp/osm-lead-source/reports", ), dashboard_snapshot_path=_absolute_path( source, _SNAPSHOT_PATH_ENV, "/tmp/osm-lead-source/dashboard.json", ), run_on_start=_parse_bool(source, _RUN_ON_START_ENV, False), schedule_day=_bounded_int( source, _SCHEDULE_DAY_ENV, 1, minimum=1, maximum=28, ), schedule_hour_utc=_bounded_int( source, _SCHEDULE_HOUR_ENV, 2, minimum=0, maximum=23, ), timeout_seconds=_bounded_float( source, _TIMEOUT_ENV, 120.0, minimum=5.0, maximum=900.0, ), max_leads=_bounded_int( source, _MAX_LEADS_ENV, 1_000_000, minimum=1, maximum=1_000_000, ), max_review_items=_bounded_int( source, _MAX_REVIEW_ENV, 1_000_000, minimum=1, maximum=1_000_000, ), markets=parse_market_selection(source.get(_COUNTRIES_ENV, "all")), extraction_timeout_seconds=_bounded_float( source, _EXTRACTION_TIMEOUT_ENV, 7200.0, minimum=30.0, maximum=14_400.0, ), max_source_bytes=_bounded_int( source, _MAX_SOURCE_BYTES_ENV, _MAX_SOURCE_BYTES, minimum=32 * 1024 * 1024, maximum=_MAX_SOURCE_BYTES, ), ) def next_monthly_run(now: datetime, *, day: int, hour_utc: int) -> datetime: """Return the next configured monthly UTC execution time.""" current = now.astimezone(timezone.utc) candidate = current.replace( day=day, hour=hour_utc, minute=0, second=0, microsecond=0, ) if candidate > current: return candidate if current.month == 12: return candidate.replace(year=current.year + 1, month=1) return candidate.replace(month=current.month + 1) def _ensure_directory(path: Path) -> None: path.mkdir(mode=0o700, parents=True, exist_ok=True) if path.is_symlink() or not path.is_dir(): raise InvalidConfiguration("worker path must be a regular directory") def _atomic_json(path: Path, value: object) -> None: _ensure_directory(path.parent) if path.exists() and (path.is_symlink() or not path.is_file()): raise InvalidConfiguration("worker output must be a regular file") payload = ( json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n" ).encode() temporary = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") try: with temporary.open("xb") as handle: os.chmod(temporary, 0o600) handle.write(payload) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _hash_values(values: list[str], label: str) -> str: payload = json.dumps( {"label": label, "values": sorted(values)}, sort_keys=True, separators=(",", ":"), ).encode() return hashlib.sha256(payload).hexdigest() def _bounded_error_code(error: BaseException) -> str: """Return only bounded class, stage, and one-way reason codes.""" names: list[str] = [] current: BaseException | None = error seen: set[int] = set() while current is not None and len(names) < 3 and id(current) not in seen: seen.add(id(current)) diagnostic_code = getattr(current, "diagnostic_code", None) if ( isinstance(diagnostic_code, str) and diagnostic_code and all(character.isalnum() or character in "._:-" for character in diagnostic_code) ): names.append(diagnostic_code[:192]) break name = type(current).__name__ if isinstance(current, ContractViolation): names.append(f"ContractViolation:{bounded_contract_reason(current)}") else: names.append(name if name.isidentifier() else "Exception") next_error = current.__cause__ if next_error is None and not current.__suppress_context__: next_error = current.__context__ current = next_error return ".".join(names)[:192] def dashboard_snapshot( result: MarketShardResult, *, started_at: datetime, completed_at: datetime, ) -> dict[str, object]: """Build a bounded dashboard snapshot from one completed shard.""" states = dict(result.report.state_counts) return { "kind": "runtime-evidence", "generated_at": completed_at.astimezone(timezone.utc).isoformat(), "region": result.market.country_code, "profile": result.report.profile.profile_id, "profile_version": result.report.profile.version, "report_hash": result.report.report_hash, "source_sha256": result.report.source_sha256, "source_size_bytes": result.report.source_size_bytes, "counts": { "active": states.get("active", 0), "review": states.get("pending_review", 0), "excluded": states.get("excluded", 0), "out_of_scope": states.get("out_of_scope", 0), }, "categories": dict(result.report.category_counts), "coverage": { "configured_markets": 1, "completed_markets": 1, "failed_markets": 0, "configured_shards": 1, "completed_shards": 1, "failed_shards": 0, }, "noco": { "comparison_completed": False, "exact_identity_matches": 0, "unmatched_identities": 0, "duplicate_row_excess_detected": False, }, "run": { "cache_hit": result.acquisition.cache_hit, "completed_at": completed_at.astimezone(timezone.utc).isoformat(), "started_at": started_at.astimezone(timezone.utc).isoformat(), "status": "completed", }, } def failure_snapshot( *, started_at: datetime, completed_at: datetime, error_code: str, ) -> dict[str, object]: """Return secret-free worker failure state without replacing lead evidence.""" return { "completed_at": completed_at.astimezone(timezone.utc).isoformat(), "error_code": error_code, "started_at": started_at.astimezone(timezone.utc).isoformat(), "status": "failed", } def build_market_plan( config: WorkerConfig, *, catalog: GeofabrikCatalog | None = None, ) -> tuple[MarketSourceShard, ...]: """Resolve all configured markets into deterministic smallest-available shards.""" active_catalog = catalog or GeofabrikCatalog.fetch( timeout_seconds=min(config.timeout_seconds, 120.0) ) planned: list[MarketSourceShard] = [] for market in config.markets: planned.extend(active_catalog.plan_market_leaf_shards(market)) if len(planned) > _MAX_SHARDS: raise InvalidConfiguration("worker market plan exceeds the bounded shard count") return tuple(sorted(planned, key=lambda item: item.shard_id)) def _aggregate_snapshot( *, config: WorkerConfig, started_at: datetime, completed_at: datetime, plan: tuple[MarketSourceShard, ...], successful: list[MarketShardResult], failed_shards: Mapping[str, int], failure_codes: Mapping[str, int], run_status: str, ) -> dict[str, object]: active: dict[tuple[str, str, int], str] = {} reviews: set[tuple[str, str, int]] = set() category_counts: Counter[str] = Counter() excluded = 0 out_of_scope = 0 source_sizes: dict[str, int] = {} report_hashes: list[str] = [] cache_hits: list[bool] = [] completed_by_country: Counter[str] = Counter() configured_by_country: Counter[str] = Counter(shard.market.country_code for shard in plan) for result in successful: country = result.market.country_code completed_by_country[country] += 1 states = dict(result.report.state_counts) excluded += states.get("excluded", 0) out_of_scope += states.get("out_of_scope", 0) report_hashes.append(result.report.report_hash) source_sizes[result.report.source_sha256] = result.report.source_size_bytes cache_hits.append( result.acquisition.cache_hit and (result.extraction is None or result.extraction.cache_hit) ) for lead in result.report.leads: key = (country, lead.identity.osm_type.value, lead.identity.osm_id) if key not in active: active[key] = lead.category category_counts[lead.category] += 1 reviews.discard(key) for review in result.report.review_items: key = (country, review.identity.osm_type.value, review.identity.osm_id) if key not in active: reviews.add(key) completed_markets = sum( completed_by_country[code] == configured_by_country[code] and failed_shards.get(code, 0) == 0 for code in configured_by_country ) failed_markets = sum(failed_shards.get(code, 0) > 0 for code in configured_by_country) source_hashes = sorted(source_sizes) profile_id = "luxeillum" profile_version = "0.1.0-draft" if successful: profile_id = successful[0].report.profile.profile_id profile_version = successful[0].report.profile.version return { "kind": "runtime-evidence", "generated_at": completed_at.astimezone(timezone.utc).isoformat(), "region": "english-markets", "profile": profile_id, "profile_version": profile_version, "report_hash": _hash_values(report_hashes, "market-report-hashes"), "source_sha256": _hash_values(source_hashes, "market-source-hashes"), "source_size_bytes": sum(source_sizes.values()), "counts": { "active": len(active), "review": len(reviews), "excluded": excluded, "out_of_scope": out_of_scope, }, "categories": dict(sorted(category_counts.items())), "coverage": { "configured_markets": len(config.markets), "completed_markets": completed_markets, "failed_markets": failed_markets, "configured_shards": len(plan), "completed_shards": len(successful), "failed_shards": sum(failed_shards.values()), }, "failures": {"by_code": dict(sorted(failure_codes.items()))}, "noco": { "comparison_completed": False, "exact_identity_matches": 0, "unmatched_identities": 0, "duplicate_row_excess_detected": False, }, "run": { "cache_hit": bool(cache_hits) and all(cache_hits), "completed_at": completed_at.astimezone(timezone.utc).isoformat(), "started_at": started_at.astimezone(timezone.utc).isoformat(), "status": run_status, }, } def _store_shard_result(config: WorkerConfig, result: MarketShardResult) -> None: root = ( config.report_dir / result.market.country_code.casefold() / result.shard.feature.feature_id ) report_hash = result.report.report_hash _atomic_json(root / f"{report_hash}.json", result.report.as_mapping()) _atomic_json(root / f"{report_hash}.evidence.json", result.evidence_mapping()) def run_worker_once( config: WorkerConfig, *, clock: Callable[[], datetime] = _utc_now, ) -> dict[str, object]: """Process every configured market shard and publish aggregate evidence.""" started_at = clock() _ensure_directory(config.cache_dir) _ensure_directory(config.report_dir) try: plan = build_market_plan(config) except Exception as exc: completed_at = clock() error_code = type(exc).__name__ _atomic_json( config.report_dir / "latest-failure.json", failure_snapshot( started_at=started_at, completed_at=completed_at, error_code=error_code, ), ) raise successful: list[MarketShardResult] = [] failed_shards: dict[str, int] = defaultdict(int) failure_codes: Counter[str] = Counter() for shard in plan: try: result = run_market_shard( shard, config.cache_dir, timeout_seconds=config.timeout_seconds, extraction_timeout_seconds=config.extraction_timeout_seconds, max_source_bytes=config.max_source_bytes, max_leads=config.max_leads, max_review_items=config.max_review_items, ) _store_shard_result(config, result) successful.append(result) print( json.dumps( { "country": result.market.country_code, "event": "read_only_shard_completed", "lead_count": len(result.report.leads), "report_hash": result.report.report_hash, "shard": result.shard.feature.feature_id, "writes_enabled": False, }, sort_keys=True, separators=(",", ":"), ), flush=True, ) except Exception as exc: failed_shards[shard.market.country_code] += 1 error_code = _bounded_error_code(exc) failure_codes[error_code] += 1 print( json.dumps( { "country": shard.market.country_code, "error_code": error_code, "event": "read_only_shard_failed", "shard": shard.feature.feature_id, "writes_enabled": False, }, sort_keys=True, separators=(",", ":"), ), flush=True, ) partial = _aggregate_snapshot( config=config, started_at=started_at, completed_at=clock(), plan=plan, successful=successful, failed_shards=failed_shards, failure_codes=failure_codes, run_status="running", ) _atomic_json(config.dashboard_snapshot_path, partial) completed_at = clock() status = "completed" if not failed_shards else "partial" snapshot = _aggregate_snapshot( config=config, started_at=started_at, completed_at=completed_at, plan=plan, successful=successful, failed_shards=failed_shards, failure_codes=failure_codes, run_status=status, ) _atomic_json(config.dashboard_snapshot_path, snapshot) coverage = snapshot.get("coverage") counts = snapshot.get("counts") failures = snapshot.get("failures") report_hash = snapshot.get("report_hash") if ( not isinstance(coverage, dict) or not isinstance(counts, dict) or not isinstance(failures, dict) ): raise InvalidConfiguration("aggregate snapshot has an invalid internal shape") failure_code_counts = failures.get("by_code") if not isinstance(failure_code_counts, dict): raise InvalidConfiguration("aggregate failure evidence has an invalid internal shape") if not isinstance(report_hash, str): raise InvalidConfiguration("aggregate report hash is invalid") summary = { "completed_markets": coverage.get("completed_markets", 0), "completed_shards": len(successful), "event": ( "read_only_cycle_completed" if status == "completed" else "read_only_cycle_partial" ), "failed_markets": coverage.get("failed_markets", 0), "failed_shards": sum(failed_shards.values()), "failure_codes": failure_code_counts, "lead_count": counts.get("active", 0), "report_hash": report_hash, "status": status, "writes_enabled": False, } print(json.dumps(summary, sort_keys=True, separators=(",", ":")), flush=True) if failed_shards: raise PartialCycleError(summary) return summary def scheduler_loop( config: WorkerConfig, *, clock: Callable[[], datetime] = _utc_now, sleep: Callable[[float], None] = time.sleep, ) -> None: """Run once on startup when configured, then at the monthly UTC schedule.""" if config.run_on_start: try: run_worker_once(config, clock=clock) except Exception: pass while True: now = clock() target = next_monthly_run( now, day=config.schedule_day, hour_utc=config.schedule_hour_utc, ) sleep(max(0.0, (target - now).total_seconds())) try: run_worker_once(config, clock=clock) except Exception: continue def start_scheduler_from_env( environ: Mapping[str, str] | None = None, ) -> threading.Thread | None: """Start the worker only when request and approval flags are both enabled.""" source = os.environ if environ is None else environ scheduler_enabled = _parse_bool(source, _SCHEDULER_ENABLED_ENV, False) scheduler_approved = _parse_bool(source, _SCHEDULER_APPROVED_ENV, False) if not (scheduler_enabled and scheduler_approved): return None config = WorkerConfig.from_env(source) thread = threading.Thread( target=scheduler_loop, args=(config,), name="osm-read-only-worker", daemon=True, ) thread.start() return thread __all__ = [ "WorkerConfig", "build_market_plan", "dashboard_snapshot", "failure_snapshot", "next_monthly_run", "run_worker_once", "scheduler_loop", "start_scheduler_from_env", ]