80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
from .models import CampaignConfig
|
|
|
|
|
|
class CampaignLoadError(ValueError):
|
|
"""Raised when the campaign JSON cannot be loaded or parsed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SchemaValidationError:
|
|
path: str
|
|
message: str
|
|
|
|
|
|
class CampaignSchemaError(CampaignLoadError):
|
|
def __init__(self, errors: list[SchemaValidationError]) -> None:
|
|
self.errors = errors
|
|
details = "; ".join(f"{error.path}: {error.message}" for error in errors[:5])
|
|
if len(errors) > 5:
|
|
details += f"; ... and {len(errors) - 5} more"
|
|
super().__init__(f"campaign schema validation failed: {details}")
|
|
|
|
|
|
def load_campaign_json(path: str | Path) -> dict[str, Any]:
|
|
campaign_path = Path(path)
|
|
try:
|
|
with campaign_path.open("r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except OSError as exc:
|
|
raise CampaignLoadError(f"could not read campaign JSON {campaign_path}: {exc}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise CampaignLoadError(f"invalid campaign JSON {campaign_path}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise CampaignLoadError("campaign JSON root must be an object")
|
|
return data
|
|
|
|
|
|
def _default_schema_path() -> Path:
|
|
return Path(__file__).resolve().parents[1] / "schema" / "campaign.schema.json"
|
|
|
|
|
|
def load_campaign_schema(schema_path: str | Path | None = None) -> dict[str, Any]:
|
|
path = Path(schema_path) if schema_path else _default_schema_path()
|
|
return load_campaign_json(path)
|
|
|
|
|
|
def validate_against_schema(data: dict[str, Any], schema_path: str | Path | None = None) -> None:
|
|
schema = load_campaign_schema(schema_path)
|
|
validator = Draft202012Validator(schema, format_checker=FormatChecker())
|
|
errors = sorted(validator.iter_errors(data), key=lambda error: list(error.path))
|
|
if errors:
|
|
normalized = [
|
|
SchemaValidationError(
|
|
path="/" + "/".join(str(part) for part in error.absolute_path),
|
|
message=error.message,
|
|
)
|
|
for error in errors
|
|
]
|
|
raise CampaignSchemaError(normalized)
|
|
|
|
|
|
def load_campaign_config(
|
|
path: str | Path,
|
|
*,
|
|
validate_schema: bool = True,
|
|
schema_path: str | Path | None = None,
|
|
) -> CampaignConfig:
|
|
data = load_campaign_json(path)
|
|
if validate_schema:
|
|
validate_against_schema(data, schema_path=schema_path)
|
|
return CampaignConfig.model_validate(data)
|