59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""
|
|
JSON Schema validation service.
|
|
Validates Gemini output against the unified video analysis schema.
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
from jsonschema import Draft7Validator, ValidationError
|
|
|
|
# Load schema on module import
|
|
SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "video_analysis.schema.json"
|
|
|
|
with open(SCHEMA_PATH, "r") as f:
|
|
VIDEO_ANALYSIS_SCHEMA = json.load(f)
|
|
|
|
# Create validator instance
|
|
validator = Draft7Validator(VIDEO_ANALYSIS_SCHEMA)
|
|
|
|
|
|
def validate_analysis(data: Dict[str, Any]) -> tuple[bool, Optional[str]]:
|
|
"""
|
|
Validate analysis data against the unified schema.
|
|
|
|
Returns:
|
|
(is_valid, error_message)
|
|
- is_valid: True if valid, False otherwise
|
|
- error_message: First validation error if invalid, None if valid
|
|
"""
|
|
try:
|
|
validator.validate(data)
|
|
return True, None
|
|
except ValidationError as e:
|
|
# Format error message with path
|
|
path = " -> ".join(str(p) for p in e.path) if e.path else "root"
|
|
error_msg = f"Validation error at {path}: {e.message}"
|
|
return False, error_msg
|
|
|
|
|
|
def first_error(data: Dict[str, Any]) -> Optional[str]:
|
|
"""
|
|
Get the first validation error for the given data.
|
|
Returns None if data is valid.
|
|
"""
|
|
for err in validator.iter_errors(data):
|
|
path = " -> ".join(str(p) for p in err.path) if err.path else "root"
|
|
return f"{err.message} at {path}"
|
|
return None
|
|
|
|
|
|
def get_all_errors(data: Dict[str, Any]) -> list[str]:
|
|
"""
|
|
Get all validation errors for the given data.
|
|
Returns empty list if data is valid.
|
|
"""
|
|
errors = []
|
|
for err in validator.iter_errors(data):
|
|
path = " -> ".join(str(p) for p in err.path) if err.path else "root"
|
|
errors.append(f"{err.message} at {path}")
|
|
return errors
|