| """Load and validate pinned application metadata used by Orienter prompts.""" |
|
|
| import json |
| from collections.abc import Mapping |
| from pathlib import Path |
|
|
|
|
| def _reject_duplicate_keys(pairs): |
| result = {} |
| for key, value in pairs: |
| if key in result: |
| raise ValueError(f"Duplicate key in metadata cache: {key}") |
| result[key] = value |
| return result |
|
|
|
|
| def _loads_json(text: str): |
| return json.loads(text, object_pairs_hook=_reject_duplicate_keys) |
|
|
|
|
| def _validated_record(app_id, record): |
| normalized_id = str(app_id) |
| if not isinstance(record, Mapping): |
| raise ValueError( |
| f"metadata cache record for app_id {normalized_id} must be a JSON object" |
| ) |
| if "app_id" in record and str(record["app_id"]) != normalized_id: |
| raise ValueError( |
| f"metadata cache key {normalized_id} conflicts with record app_id {record['app_id']}" |
| ) |
| app_name = record.get("app_name") or record.get("name") or record.get("title") |
| app_description = record.get("app_description") or record.get("description") |
| if not isinstance(app_name, str) or not app_name.strip(): |
| raise ValueError(f"metadata cache record for app_id {normalized_id} is missing app_name") |
| if not isinstance(app_description, str) or not app_description.strip(): |
| raise ValueError( |
| f"metadata cache record for app_id {normalized_id} is missing app_description" |
| ) |
| return dict(record) |
|
|
|
|
| def _add_record(cache, app_id, record): |
| normalized_id = str(app_id) |
| if normalized_id in cache: |
| raise ValueError(f"Duplicate app_id in metadata cache: {normalized_id}") |
| cache[normalized_id] = _validated_record(normalized_id, record) |
|
|
|
|
| def load_app_metadata_cache(path): |
| """Return a validated app-id mapping from JSON, JSON-list, or JSONL input.""" |
| if path is None: |
| return None |
|
|
| path = Path(path) |
| if path.suffix.lower() == ".jsonl": |
| cache = {} |
| with path.open(encoding="utf-8") as file: |
| for line_number, line in enumerate(file, start=1): |
| if not line.strip(): |
| continue |
| record = _loads_json(line) |
| if not isinstance(record, Mapping) or "app_id" not in record: |
| raise ValueError( |
| f"metadata JSONL record on line {line_number} must include app_id" |
| ) |
| _add_record(cache, record["app_id"], record) |
| return cache |
|
|
| loaded = _loads_json(path.read_text(encoding="utf-8")) |
| if isinstance(loaded, list): |
| cache = {} |
| for index, record in enumerate(loaded): |
| if not isinstance(record, Mapping) or "app_id" not in record: |
| raise ValueError(f"metadata JSON list record {index} must include app_id") |
| _add_record(cache, record["app_id"], record) |
| return cache |
| if not isinstance(loaded, Mapping): |
| raise ValueError(f"Metadata cache must be a JSON object, list, or JSONL: {path}") |
|
|
| cache = {} |
| for app_id, record in loaded.items(): |
| _add_record(cache, app_id, record) |
| return cache |
|
|
|
|
| def get_app_metadata(app_id, metadata_cache): |
| """Return the validated app name and description for one app ID.""" |
| normalized_id = str(app_id) |
| record = metadata_cache.get(normalized_id) |
| if record is None: |
| raise KeyError(f"app_id {normalized_id} is missing from the app metadata cache") |
| record = _validated_record(normalized_id, record) |
| app_name = record.get("app_name") or record.get("name") or record.get("title") |
| app_description = record.get("app_description") or record.get("description") |
| return app_name, app_description |
|
|
