| """JSON Schema helpers for dataset-only output contracts.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
| from pydantic import BaseModel |
|
|
| from gcmd_classifier.datasets.application_service import AwaitingReadmeSelection |
| from gcmd_classifier.datasets.comparison_models import ( |
| DatasetComparisonResult, |
| HumanReviewRecord, |
| ) |
| from gcmd_classifier.datasets.deterministic_evidence import ( |
| DeterministicEvidencePacket, |
| DeterministicEvidenceSelection, |
| ) |
| from gcmd_classifier.datasets.models import ( |
| CoverageResolutionFailure, |
| DatasetClassificationResult, |
| DatasetCoverageResolution, |
| DatasetEvidencePacket, |
| DatasetExtractionManifest, |
| ) |
| from gcmd_classifier.datasets.runtime_models import ( |
| ExpertSourceBinding, |
| SealedExpertArtifactEnvelope, |
| VerifiedPersistedBlindResultReference, |
| ) |
| from gcmd_classifier.datasets.structural_scope import ( |
| DeterministicScopeResult, |
| StructuralPassageIndex, |
| ) |
|
|
| DATASET_SCHEMA_MODELS: dict[str, type[BaseModel]] = { |
| "dataset_structural_passage_index.schema.json": StructuralPassageIndex, |
| "dataset_deterministic_product_scope.schema.json": DeterministicScopeResult, |
| "dataset_deterministic_evidence_selection.schema.json": DeterministicEvidenceSelection, |
| "dataset_deterministic_evidence_packet.schema.json": DeterministicEvidencePacket, |
| "dataset_awaiting_readme_selection.schema.json": AwaitingReadmeSelection, |
| "dataset_classification_result.schema.json": DatasetClassificationResult, |
| "dataset_coverage_resolution.schema.json": DatasetCoverageResolution, |
| "dataset_coverage_resolution_failure.schema.json": CoverageResolutionFailure, |
| "dataset_extraction_manifest.schema.json": DatasetExtractionManifest, |
| "dataset_evidence_packet.schema.json": DatasetEvidencePacket, |
| "dataset_comparison.schema.json": DatasetComparisonResult, |
| "dataset_human_review.schema.json": HumanReviewRecord, |
| "dataset_sealed_expert_artifact.schema.json": SealedExpertArtifactEnvelope, |
| "dataset_expert_source_binding.schema.json": ExpertSourceBinding, |
| "dataset_verified_blind_reference.schema.json": VerifiedPersistedBlindResultReference, |
| } |
|
|
|
|
| def schema_for(model_type: type[BaseModel]) -> dict[str, Any]: |
| """Generate and validate a Draft 2020-12-compatible Pydantic schema.""" |
| schema = model_type.model_json_schema() |
| if model_type is DatasetClassificationResult: |
| schema.setdefault("allOf", []).append( |
| { |
| "if": { |
| "properties": {"classification_attempted": {"const": False}}, |
| "required": ["classification_attempted"], |
| }, |
| "then": { |
| "properties": { |
| "classification_outcome": {"type": "null"}, |
| "classifications": {"maxItems": 0}, |
| "blind_result": {"type": "null"}, |
| "comparison_reference": {"type": "null"}, |
| "processing_metadata": { |
| "properties": {"model_calls": {"const": 0}}, |
| "required": ["model_calls"], |
| }, |
| } |
| }, |
| } |
| ) |
| return schema |
|
|
|
|
| def validate_serialized(model_type: type[BaseModel], payload: dict[str, Any]) -> BaseModel: |
| """Validate a serialized dataset payload through its authoritative typed contract.""" |
| return model_type.model_validate(payload) |
|
|
|
|
| def write_dataset_schemas(directory: Path) -> tuple[Path, ...]: |
| """Write formal dataset schemas deterministically.""" |
| directory.mkdir(parents=True, exist_ok=True) |
| written: list[Path] = [] |
| for filename, model_type in DATASET_SCHEMA_MODELS.items(): |
| path = directory / filename |
| path.write_text(json.dumps(schema_for(model_type), indent=2) + "\n") |
| written.append(path) |
| return tuple(written) |
|
|
