|
| 1 | +"""Translate installed Core capabilities into native text scan configuration.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from copy import deepcopy |
| 6 | +from functools import lru_cache |
| 7 | +from typing import Any |
| 8 | + |
| 9 | +_ASCII_LOWER = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz") |
| 10 | + |
| 11 | + |
| 12 | +def _locale_key(value: str) -> str: |
| 13 | + return value.strip(" \t\n\r\v\f").translate(_ASCII_LOWER) |
| 14 | + |
| 15 | + |
| 16 | +def _invalid(detail: str) -> RuntimeError: |
| 17 | + return RuntimeError(f"Incompatible datafog-core capabilities: {detail}") |
| 18 | + |
| 19 | + |
| 20 | +def _labels(value: Any, field: str) -> set[str]: |
| 21 | + if not isinstance(value, list) or any( |
| 22 | + not isinstance(label, str) or not label for label in value |
| 23 | + ): |
| 24 | + raise _invalid(f"{field} must be a list of entity labels") |
| 25 | + return set(value) |
| 26 | + |
| 27 | + |
| 28 | +@lru_cache(maxsize=4) |
| 29 | +def _read_capabilities(capabilities: Any) -> tuple[set[str], dict, dict]: |
| 30 | + """Snapshot one installed build; replacing its reader invalidates the cache. |
| 31 | +
|
| 32 | + The private snapshot never escapes into native scan configuration. Failed |
| 33 | + capability reads or validation are not cached by lru_cache. |
| 34 | + """ |
| 35 | + payload = deepcopy(capabilities()) |
| 36 | + if not isinstance(payload, dict): |
| 37 | + raise _invalid("capabilities() must return a dictionary") |
| 38 | + version = payload.get("contract_version") |
| 39 | + if not isinstance(version, int) or isinstance(version, bool) or version != 1: |
| 40 | + raise _invalid(f"unsupported contract_version {version!r}; expected 1") |
| 41 | + supported = _labels(payload.get("supported_entities"), "supported_entities") |
| 42 | + defaults = _labels(payload.get("default_entities"), "default_entities") |
| 43 | + if not defaults <= supported: |
| 44 | + raise _invalid("default_entities contains unsupported labels") |
| 45 | + advertised_locales = payload.get("locales") |
| 46 | + metadata = payload.get("entities") |
| 47 | + if not isinstance(advertised_locales, dict) or not isinstance(metadata, dict): |
| 48 | + raise _invalid("locales and entities must be dictionaries") |
| 49 | + |
| 50 | + return supported, _locale_lookup(advertised_locales, supported), metadata |
| 51 | + |
| 52 | + |
| 53 | +def _locale_lookup(advertised_locales: dict, supported: set[str]) -> dict[str, str]: |
| 54 | + locale_lookup: dict[str, str] = {} |
| 55 | + for locale, details in advertised_locales.items(): |
| 56 | + if not isinstance(locale, str) or not _locale_key(locale): |
| 57 | + raise _invalid("locale identifiers must be nonempty strings") |
| 58 | + if ( |
| 59 | + not isinstance(details, dict) |
| 60 | + or not _labels( |
| 61 | + details.get("enabled_entities"), f"locales[{locale!r}].enabled_entities" |
| 62 | + ) |
| 63 | + <= supported |
| 64 | + ): |
| 65 | + raise _invalid(f"invalid locale metadata for {locale!r}") |
| 66 | + locale_lookup[_locale_key(locale)] = locale |
| 67 | + |
| 68 | + return locale_lookup |
| 69 | + |
| 70 | + |
| 71 | +def _resolve_locale(value: Any, locale_lookup: dict[str, str]) -> str: |
| 72 | + if not isinstance(value, str) or _locale_key(value) not in locale_lookup: |
| 73 | + raise ValueError(f"Unsupported locale for the Rust backend: {value!r}") |
| 74 | + return value |
| 75 | + |
| 76 | + |
| 77 | +def _requested_locales(locales: Any, lookup: dict[str, str]) -> set[str]: |
| 78 | + if isinstance(locales, str): |
| 79 | + locales = [locales] |
| 80 | + if locales is not None and ( |
| 81 | + not isinstance(locales, (list, tuple)) |
| 82 | + or any(not isinstance(locale, str) for locale in locales) |
| 83 | + ): |
| 84 | + raise ValueError("locales must be a list of locale identifiers") |
| 85 | + return {_resolve_locale(locale, lookup) for locale in locales or []} |
| 86 | + |
| 87 | + |
| 88 | +def _activation_config(label: str, metadata: dict) -> dict: |
| 89 | + details = metadata.get(label) |
| 90 | + if not isinstance(details, dict): |
| 91 | + raise _invalid(f"missing entity metadata for {label!r}") |
| 92 | + scopes = details.get("scopes") |
| 93 | + if not isinstance(scopes, list) or any( |
| 94 | + not isinstance(scope, str) for scope in scopes |
| 95 | + ): |
| 96 | + raise _invalid(f"invalid scopes for {label!r}") |
| 97 | + if "text" not in scopes: |
| 98 | + raise ValueError( |
| 99 | + f"Entity {label!r} is not available for Rust text scanning " |
| 100 | + "(structured-only entity)" |
| 101 | + ) |
| 102 | + activation = details.get("activation") |
| 103 | + if not isinstance(activation, dict): |
| 104 | + raise _invalid(f"missing activation metadata for {label!r}") |
| 105 | + kind = activation.get("kind") |
| 106 | + if kind == "default": |
| 107 | + return {} |
| 108 | + if kind not in {"locale", "config"}: |
| 109 | + raise _invalid(f"unsupported activation kind {kind!r} for {label!r}") |
| 110 | + config = activation.get("scan_config") |
| 111 | + if not isinstance(config, dict) or not config: |
| 112 | + raise _invalid(f"missing scan_config for {label!r}") |
| 113 | + if kind == "locale" and "locale" not in config: |
| 114 | + raise _invalid(f"missing activation locale for {label!r}") |
| 115 | + return deepcopy(config) |
| 116 | + |
| 117 | + |
| 118 | +def scan_configs(core: Any, requested: set[str], locales: Any) -> list[dict]: |
| 119 | + """Build a union of singular-locale scans without duplicating inventories.""" |
| 120 | + capabilities = getattr(core, "capabilities", None) |
| 121 | + if not callable(capabilities): |
| 122 | + raise _invalid( |
| 123 | + "contract version 1 is required; install a compatible Core 0.4.x" |
| 124 | + ) |
| 125 | + supported, lookup, metadata = _read_capabilities(capabilities) |
| 126 | + selected_locales = _requested_locales(locales, lookup) |
| 127 | + common: dict[str, Any] = {} |
| 128 | + for label in sorted(requested & supported): |
| 129 | + config = _activation_config(label, metadata) |
| 130 | + for key, value in config.items(): |
| 131 | + if key == "locale": |
| 132 | + selected_locales.add(_resolve_locale(value, lookup)) |
| 133 | + elif key in common and common[key] != value: |
| 134 | + raise _invalid(f"conflicting activation values for {key!r}") |
| 135 | + else: |
| 136 | + common[key] = value |
| 137 | + if not selected_locales: |
| 138 | + return [common] |
| 139 | + return [ |
| 140 | + deepcopy(dict(common, locale=locale)) for locale in sorted(selected_locales) |
| 141 | + ] |
0 commit comments