diff --git a/tests/test_air_quality.py b/tests/test_air_quality.py index 4e7e6515..529e5adc 100644 --- a/tests/test_air_quality.py +++ b/tests/test_air_quality.py @@ -3,8 +3,8 @@ import pytest from weather_briefing.air_quality import AirQualityError, AQICNProvider, air_quality_to_document, health_guidance +from weather_briefing.data.resources import ReferenceDataError from weather_briefing.models import AirQualitySnapshot, AirQualityTimeKind -from weather_briefing.reference_data import ReferenceDataError async def test_aqicn_provider_labels_aqi_standard_without_converting_pm25() -> None: diff --git a/tests/test_allergen.py b/tests/test_allergen.py index 4096d072..9640928d 100644 --- a/tests/test_allergen.py +++ b/tests/test_allergen.py @@ -7,8 +7,8 @@ allergen_to_document, pollen_type_names, ) +from weather_briefing.data.resources import ReferenceDataError from weather_briefing.models import AllergenLevel, AllergenSnapshot -from weather_briefing.reference_data import ReferenceDataError def test_allergen_guidance_zero_is_none() -> None: diff --git a/tests/test_reference_data.py b/tests/test_reference_data.py index 65c22d8e..eaf11b84 100644 --- a/tests/test_reference_data.py +++ b/tests/test_reference_data.py @@ -6,15 +6,17 @@ from weather_briefing import __version__ from weather_briefing.air_quality import health_guidance -from weather_briefing.data.service_endpoints import NOMINATIM_USER_AGENT -from weather_briefing.reference_data import ( +from weather_briefing.data.resources import ( ReferenceDataError, load_reference_data, - localization_table, - open_meteo_weather_code_descriptions, reference_string, reference_string_tuple, reference_value, +) +from weather_briefing.data.service_endpoints import NOMINATIM_USER_AGENT +from weather_briefing.localization import localization_table +from weather_briefing.reference_data import ( + open_meteo_weather_code_descriptions, telegram_error_classification, ) @@ -157,7 +159,7 @@ def test_reference_string_tuple_rejects_non_list_value() -> None: @pytest.mark.parametrize("value", [None, "", " ", 7]) def test_reference_string_rejects_invalid_value(monkeypatch, value) -> None: - monkeypatch.setattr("weather_briefing.reference_data.reference_value", lambda *args: value) + monkeypatch.setattr("weather_briefing.data.resources.reference_value", lambda *args: value) with pytest.raises(ReferenceDataError, match="non-empty string"): reference_string("provider_defaults.json", "qweather_allergen_index_type") @@ -220,7 +222,7 @@ def test_telegram_error_classification_rejects_invalid_data(monkeypatch, value, def test_load_reference_data_rejects_non_dict_root(monkeypatch) -> None: - from weather_briefing.reference_data import load_reference_data + from weather_briefing.data.resources import load_reference_data class FakeResource: def joinpath(self, filename): @@ -230,16 +232,66 @@ def read_text(self, encoding=None): return "42" monkeypatch.setattr( - "weather_briefing.reference_data.resources.files", + "weather_briefing.data.resources.resources.files", lambda package: FakeResource(), ) - load_reference_data.cache_clear() - with pytest.raises(ReferenceDataError, match="must be an object"): load_reference_data("test.json") +def test_load_reference_data_returns_independent_values() -> None: + first = load_reference_data("localization.json") + tables = first["tables"] + assert isinstance(tables, dict) + tables.clear() + + second = load_reference_data("localization.json") + + assert second["tables"] + + +def test_reference_value_copies_only_the_selected_value(monkeypatch) -> None: + import weather_briefing.data.resources as resources_module + + class UnselectedValue: + def __deepcopy__(self, memo): + raise AssertionError( # pragma: no cover - reached only if root copying regresses + "reference_value must not copy the cached root" + ) + + selected = {"items": ["one"]} + root = {"selected": selected, "unselected": UnselectedValue()} + + monkeypatch.setattr(resources_module, "_load_reference_data", lambda filename: root) + + first = reference_value("test.json", "selected") + assert isinstance(first, dict) + items = first["items"] + assert isinstance(items, list) + items.append("changed") + + second = reference_value("test.json", "selected") + + assert second == {"items": ["one"]} + assert first is not selected + + +def test_reference_string_tuple_does_not_copy_the_cached_list(monkeypatch) -> None: + import weather_briefing.data.resources as resources_module + + class CachedStrings(list[str]): + def __deepcopy__(self, memo): + raise AssertionError( # pragma: no cover - reached only if list copying regresses + "reference_string_tuple must not copy the cached list" + ) + + cached = CachedStrings(["one", "two"]) + monkeypatch.setattr(resources_module, "_load_reference_data", lambda filename: {"selected": cached}) + + assert reference_string_tuple("test.json", "selected") == ("one", "two") + + @pytest.mark.parametrize( ("value", "message"), [ @@ -257,7 +309,7 @@ def read_text(self, encoding=None): ], ) def test_localization_table_rejects_incomplete_data(monkeypatch, value, message) -> None: - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match=message): @@ -268,7 +320,7 @@ def test_localization_table_rejects_missing_fields(monkeypatch) -> None: value, tables, _ = _mutable_localization_data() english = _localization_language(tables, "briefing", "en") del english["weather"] - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Invalid localization fields: briefing:en"): @@ -280,7 +332,7 @@ def test_localization_table_rejects_unknown_alias_target(monkeypatch) -> None: briefing = aliases["briefing"] assert _is_string_object_dict(briefing) briefing["zh-Hans"] = "missing" - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Invalid localization alias: briefing:zh-Hans"): @@ -290,7 +342,7 @@ def test_localization_table_rejects_unknown_alias_target(monkeypatch) -> None: def test_localization_table_rejects_whitespace_values(monkeypatch) -> None: value, tables, _ = _mutable_localization_data() _localization_language(tables, "briefing", "en")["weather"] = " " - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Invalid localization fields: briefing:en"): @@ -312,7 +364,7 @@ def test_localization_tables_are_immutable() -> None: def test_localization_table_rejects_invalid_alias_root(monkeypatch, aliases) -> None: value, _, _ = _mutable_localization_data() value["aliases"] = aliases - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="every supported table"): @@ -329,7 +381,7 @@ def test_localization_table_rejects_unknown_table() -> None: def test_localization_table_rejects_non_object_table(monkeypatch) -> None: value, tables, _ = _mutable_localization_data() tables["briefing"] = [] - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Unknown localization table: briefing"): @@ -341,7 +393,7 @@ def test_localization_table_rejects_non_object_labels(monkeypatch) -> None: briefing = tables["briefing"] assert _is_string_object_dict(briefing) briefing["en"] = [] - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Invalid localization fields: briefing:en"): @@ -351,7 +403,7 @@ def test_localization_table_rejects_non_object_labels(monkeypatch) -> None: def test_localization_table_rejects_non_object_table_aliases(monkeypatch) -> None: value, _, aliases = _mutable_localization_data() aliases["briefing"] = [] - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Localization aliases must be an object: briefing"): @@ -365,7 +417,7 @@ def test_localization_table_rejects_non_object_table_aliases(monkeypatch) -> Non def test_localization_table_rejects_invalid_aliases(monkeypatch, alias, target) -> None: value, _, aliases = _mutable_localization_data() aliases["briefing"] = {alias: target} - monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value) + monkeypatch.setattr("weather_briefing.data.resources._load_reference_data", lambda filename: value) localization_table.cache_clear() with pytest.raises(ReferenceDataError, match="Invalid localization alias: briefing"): diff --git a/weather_briefing/air_quality.py b/weather_briefing/air_quality.py index 6b03a532..d71fd1de 100644 --- a/weather_briefing/air_quality.py +++ b/weather_briefing/air_quality.py @@ -9,10 +9,11 @@ import pendulum from .api_client import api_call_extensions +from .data.resources import ReferenceDataError, reference_value from .data.service_endpoints import AQICN_BASE_URL from .languages import localized_labels +from .localization import localization_table from .models import AirQualitySnapshot, AirQualityTimeKind, SourceDocument -from .reference_data import ReferenceDataError, localization_table, reference_value from .time_utils import parse_datetime_with_default_timezone _AIR_QUALITY_FORMATS = localization_table("air_quality") diff --git a/weather_briefing/allergen.py b/weather_briefing/allergen.py index ab12b5b4..48afb9ac 100644 --- a/weather_briefing/allergen.py +++ b/weather_briefing/allergen.py @@ -5,9 +5,10 @@ from functools import cache from math import isfinite +from .data.resources import ReferenceDataError, reference_value from .languages import localized_labels +from .localization import localization_table from .models import AllergenSnapshot, SourceDocument -from .reference_data import ReferenceDataError, localization_table, reference_value _ALLERGEN_FORMATS = localization_table("allergen") diff --git a/weather_briefing/content_cleaners.py b/weather_briefing/content_cleaners.py index c9a870e4..588bf2d4 100644 --- a/weather_briefing/content_cleaners.py +++ b/weather_briefing/content_cleaners.py @@ -10,7 +10,7 @@ from bs4 import BeautifulSoup, Comment from soupsieve import SelectorSyntaxError -from .reference_data import reference_string_tuple +from .data.resources import reference_string_tuple class ContentCleaningError(ValueError): diff --git a/weather_briefing/data/resources.py b/weather_briefing/data/resources.py new file mode 100644 index 00000000..d22a0b4c --- /dev/null +++ b/weather_briefing/data/resources.py @@ -0,0 +1,74 @@ +"""Validated access to packaged JSON resources.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from functools import cache +from importlib import resources +from pathlib import PurePath +from typing import Any + +from .. import data + + +class ReferenceDataError(RuntimeError): + """Raised when packaged domain reference data is missing or malformed.""" + + +def _validate_reference_data_filename(filename: str) -> None: + if PurePath(filename).name != filename or not filename.endswith(".json"): + raise ReferenceDataError("Reference data filename must identify one JSON file") + + +def load_reference_data(filename: str) -> dict[str, object]: + """Load and validate one packaged JSON reference-data object.""" + _validate_reference_data_filename(filename) + return deepcopy(_load_reference_data(filename)) + + +@cache +def _load_reference_data(filename: str) -> dict[str, object]: + try: + text = resources.files(data).joinpath(filename).read_text(encoding="utf-8") + value = json.loads(text) + except (FileNotFoundError, OSError, json.JSONDecodeError) as exc: + raise ReferenceDataError(f"Unable to load reference data: {filename}") from exc + if not isinstance(value, dict): + raise ReferenceDataError(f"Reference data root must be an object: {filename}") + return value + + +def _cached_reference_value(filename: str, *path: str) -> Any: + _validate_reference_data_filename(filename) + value: Any = _load_reference_data(filename) + try: + for key in path: + value = value[key] + except (KeyError, TypeError) as exc: + joined_path = ".".join(path) + raise ReferenceDataError(f"Missing reference data field: {filename}:{joined_path}") from exc + return value + + +def reference_value(filename: str, *path: str) -> Any: + """Read an isolated nested value from a packaged reference-data file.""" + return deepcopy(_cached_reference_value(filename, *path)) + + +def reference_string(filename: str, *path: str) -> str: + """Read a non-empty string from packaged reference data.""" + value = reference_value(filename, *path) + if not isinstance(value, str) or not value.strip(): + joined_path = ".".join(path) + raise ReferenceDataError(f"Reference data field must be a non-empty string: {filename}:{joined_path}") + return value + + +def reference_string_tuple(filename: str, *path: str) -> tuple[str, ...]: + """Read a non-empty string sequence from packaged reference data.""" + value = _cached_reference_value(filename, *path) + if not isinstance(value, list) or not value or not all(isinstance(item, str) and item.strip() for item in value): + joined_path = ".".join(path) + raise ReferenceDataError(f"Reference data field must be a non-empty string list: {filename}:{joined_path}") + return tuple(value) diff --git a/weather_briefing/localization.py b/weather_briefing/localization.py new file mode 100644 index 00000000..780774b8 --- /dev/null +++ b/weather_briefing/localization.py @@ -0,0 +1,117 @@ +"""Validated localized scaffold tables.""" + +from __future__ import annotations + +from collections.abc import Mapping +from functools import cache +from types import MappingProxyType +from typing import TypeGuard + +from .data.resources import ReferenceDataError, reference_value +from .languages import normalize_language_tag + +_LOCALIZATION_FIELDS = { + "air_quality": frozenset( + { + "separator", + "unavailable", + "forecast_time", + "observation_time", + "observation", + "forecast", + "time_kind", + "aqi", + "aqi_summary", + "pm25_aqi", + "pm25_summary", + "pm25", + "health", + } + ), + "allergen": frozenset( + {"separator", "unavailable", "observed_at", "allergens", "overall", "health", "count", "level"} + ), + "briefing": frozenset( + { + "weather", + "warnings", + "disasters", + "advice", + "attribution", + "html_source_separator", + "plain_source_separator", + "status_open", + "status_close", + "detail_separator", + } + ), + "qweather": frozenset({"day", "lifestyle", "unknown", "no_details"}), + "weather_document": frozenset( + { + "separator", + "section_separator", + "unavailable", + "updated_at", + "forecast", + "lifestyle", + "summary", + "lifestyle_count", + } + ), +} +_LOCALIZATION_LANGUAGES = frozenset({"zh-CN", "zh-TW", "en", "ja"}) + + +def _is_normalized_language(value: object) -> bool: + if not isinstance(value, str): + return False + try: + return normalize_language_tag(value) == value + except ValueError: + return False + + +def _is_localization_labels(value: object, expected_fields: frozenset[str]) -> TypeGuard[dict[str, str]]: + return ( + isinstance(value, dict) + and set(value) == expected_fields + and all(isinstance(item, str) and item.strip() for item in value.values()) + ) + + +@cache +def localization_table(name: str) -> Mapping[str, Mapping[str, str]]: + """Return one fully validated localized scaffold table.""" + tables = reference_value("localization.json", "tables") + aliases = reference_value("localization.json", "aliases") + if ( + not isinstance(tables, dict) + or set(tables) != set(_LOCALIZATION_FIELDS) + or not isinstance(aliases, dict) + or not set(aliases).issubset(_LOCALIZATION_FIELDS) + ): + raise ReferenceDataError("Localization data must contain every supported table") + table = tables.get(name) + if name not in _LOCALIZATION_FIELDS or not isinstance(table, dict): + raise ReferenceDataError(f"Unknown localization table: {name}") + if set(table) != _LOCALIZATION_LANGUAGES: + raise ReferenceDataError(f"Localization table must contain every supported language: {name}") + expected_fields = _LOCALIZATION_FIELDS[name] + validated: dict[str, Mapping[str, str]] = {} + for language, labels in table.items(): + if not _is_localization_labels(labels, expected_fields): + raise ReferenceDataError(f"Invalid localization fields: {name}:{language}") + validated[language] = MappingProxyType(dict(labels)) + table_aliases = aliases.get(name, {}) + if not isinstance(table_aliases, dict): + raise ReferenceDataError(f"Localization aliases must be an object: {name}") + for alias, target in table_aliases.items(): + if ( + not _is_normalized_language(alias) + or not isinstance(target, str) + or target not in validated + or alias in validated + ): + raise ReferenceDataError(f"Invalid localization alias: {name}:{alias}") + validated[alias] = validated[target] + return MappingProxyType(validated) diff --git a/weather_briefing/reference_data.py b/weather_briefing/reference_data.py index 10fb7037..8c4ff459 100644 --- a/weather_briefing/reference_data.py +++ b/weather_briefing/reference_data.py @@ -1,94 +1,25 @@ -"""Validated access to packaged domain reference data.""" +"""Compatibility exports for reference data awaiting feature migration.""" from __future__ import annotations -import json import re from collections.abc import Mapping from dataclasses import dataclass from functools import cache -from importlib import resources -from pathlib import PurePath from types import MappingProxyType -from typing import Any, TypeGuard -from . import data -from .languages import normalize_language_tag +from .data.resources import ( + ReferenceDataError, + load_reference_data, + reference_string, + reference_string_tuple, + reference_value, +) +from .localization import localization_table -_LOCALIZATION_FIELDS = { - "air_quality": frozenset( - { - "separator", - "unavailable", - "forecast_time", - "observation_time", - "observation", - "forecast", - "time_kind", - "aqi", - "aqi_summary", - "pm25_aqi", - "pm25_summary", - "pm25", - "health", - } - ), - "allergen": frozenset( - {"separator", "unavailable", "observed_at", "allergens", "overall", "health", "count", "level"} - ), - "briefing": frozenset( - { - "weather", - "warnings", - "disasters", - "advice", - "attribution", - "html_source_separator", - "plain_source_separator", - "status_open", - "status_close", - "detail_separator", - } - ), - "qweather": frozenset({"day", "lifestyle", "unknown", "no_details"}), - "weather_document": frozenset( - { - "separator", - "section_separator", - "unavailable", - "updated_at", - "forecast", - "lifestyle", - "summary", - "lifestyle_count", - } - ), -} -_LOCALIZATION_LANGUAGES = frozenset({"zh-CN", "zh-TW", "en", "ja"}) _CLASSIFICATION_REASON = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*") -def _is_normalized_language(value: object) -> bool: - if not isinstance(value, str): - return False - try: - return normalize_language_tag(value) == value - except ValueError: - return False - - -def _is_localization_labels(value: object, expected_fields: frozenset[str]) -> TypeGuard[dict[str, str]]: - return ( - isinstance(value, dict) - and set(value) == expected_fields - and all(isinstance(item, str) and item.strip() for item in value.values()) - ) - - -class ReferenceDataError(RuntimeError): - """Raised when packaged domain reference data is missing or malformed.""" - - @dataclass(frozen=True, slots=True) class TelegramErrorClassification: """Validated Telegram API error mappings and delivery metadata.""" @@ -99,51 +30,6 @@ class TelegramErrorClassification: channel_unavailable_reasons: frozenset[str] -@cache -def load_reference_data(filename: str) -> dict[str, object]: - """Load and validate one packaged JSON reference-data object.""" - if PurePath(filename).name != filename or not filename.endswith(".json"): - raise ReferenceDataError("Reference data filename must identify one JSON file") - try: - text = resources.files(data).joinpath(filename).read_text(encoding="utf-8") - value = json.loads(text) - except (FileNotFoundError, OSError, json.JSONDecodeError) as exc: - raise ReferenceDataError(f"Unable to load reference data: {filename}") from exc - if not isinstance(value, dict): - raise ReferenceDataError(f"Reference data root must be an object: {filename}") - return value - - -def reference_value(filename: str, *path: str) -> Any: - """Read a nested value from a packaged reference-data file.""" - value: Any = load_reference_data(filename) - try: - for key in path: - value = value[key] - except (KeyError, TypeError) as exc: - joined_path = ".".join(path) - raise ReferenceDataError(f"Missing reference data field: {filename}:{joined_path}") from exc - return value - - -def reference_string(filename: str, *path: str) -> str: - """Read a non-empty string from packaged reference data.""" - value = reference_value(filename, *path) - if not isinstance(value, str) or not value.strip(): - joined_path = ".".join(path) - raise ReferenceDataError(f"Reference data field must be a non-empty string: {filename}:{joined_path}") - return value - - -def reference_string_tuple(filename: str, *path: str) -> tuple[str, ...]: - """Read a non-empty string sequence from packaged reference data.""" - value = reference_value(filename, *path) - if not isinstance(value, list) or not value or not all(isinstance(item, str) and item.strip() for item in value): - joined_path = ".".join(path) - raise ReferenceDataError(f"Reference data field must be a non-empty string list: {filename}:{joined_path}") - return tuple(value) - - @cache def open_meteo_weather_code_descriptions() -> Mapping[int, str]: """Return validated English descriptions for Open-Meteo WMO weather codes.""" @@ -245,39 +131,14 @@ def telegram_error_classification() -> TelegramErrorClassification: ) -@cache -def localization_table(name: str) -> Mapping[str, Mapping[str, str]]: - """Return one fully validated localized scaffold table.""" - tables = reference_value("localization.json", "tables") - aliases = reference_value("localization.json", "aliases") - if ( - not isinstance(tables, dict) - or set(tables) != set(_LOCALIZATION_FIELDS) - or not isinstance(aliases, dict) - or not set(aliases).issubset(_LOCALIZATION_FIELDS) - ): - raise ReferenceDataError("Localization data must contain every supported table") - table = tables.get(name) - if name not in _LOCALIZATION_FIELDS or not isinstance(table, dict): - raise ReferenceDataError(f"Unknown localization table: {name}") - if set(table) != _LOCALIZATION_LANGUAGES: - raise ReferenceDataError(f"Localization table must contain every supported language: {name}") - expected_fields = _LOCALIZATION_FIELDS[name] - validated: dict[str, Mapping[str, str]] = {} - for language, labels in table.items(): - if not _is_localization_labels(labels, expected_fields): - raise ReferenceDataError(f"Invalid localization fields: {name}:{language}") - validated[language] = MappingProxyType(dict(labels)) - table_aliases = aliases.get(name, {}) - if not isinstance(table_aliases, dict): - raise ReferenceDataError(f"Localization aliases must be an object: {name}") - for alias, target in table_aliases.items(): - if ( - not _is_normalized_language(alias) - or not isinstance(target, str) - or target not in validated - or alias in validated - ): - raise ReferenceDataError(f"Invalid localization alias: {name}:{alias}") - validated[alias] = validated[target] - return MappingProxyType(validated) +__all__ = [ + "ReferenceDataError", + "TelegramErrorClassification", + "load_reference_data", + "localization_table", + "open_meteo_weather_code_descriptions", + "reference_string", + "reference_string_tuple", + "reference_value", + "telegram_error_classification", +]