|
|
""" |
|
|
Input Validation Agent |
|
|
Validates and normalizes input data (file or text) into a standard format. |
|
|
""" |
|
|
|
|
|
from typing import Dict, Any, Optional |
|
|
import mimetypes |
|
|
from pathlib import Path |
|
|
|
|
|
from core.agent_base import Agent |
|
|
from core.errors import ValidationError |
|
|
from config.settings import settings |
|
|
|
|
|
|
|
|
class InputValidatorAgent(Agent): |
|
|
""" |
|
|
Validates input type, size, format and normalizes to internal format. |
|
|
""" |
|
|
|
|
|
def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: |
|
|
""" |
|
|
Validate and normalize input. |
|
|
|
|
|
Expected input_data: |
|
|
{ |
|
|
"type": "file" | "text", |
|
|
"content": bytes | str, |
|
|
"filename": str (optional, for files), |
|
|
} |
|
|
|
|
|
Returns: |
|
|
{ |
|
|
"content": bytes, |
|
|
"content_type": str, |
|
|
"size": int, |
|
|
"filename": str | None, |
|
|
"validation_status": "valid" |
|
|
} |
|
|
""" |
|
|
input_type = input_data.get("type") |
|
|
content = input_data.get("content") |
|
|
|
|
|
if not input_type or not content: |
|
|
raise ValidationError("Missing required fields: 'type' and 'content'") |
|
|
|
|
|
if input_type == "file": |
|
|
return self._validate_file(content, input_data.get("filename")) |
|
|
elif input_type == "text": |
|
|
return self._validate_text(content) |
|
|
else: |
|
|
raise ValidationError(f"Invalid input type: {input_type}") |
|
|
|
|
|
def _validate_file(self, content: bytes, filename: Optional[str]) -> Dict[str, Any]: |
|
|
"""Validate file input.""" |
|
|
if not isinstance(content, bytes): |
|
|
raise ValidationError("File content must be bytes") |
|
|
|
|
|
if len(content) == 0: |
|
|
raise ValidationError("File is empty") |
|
|
|
|
|
size = len(content) |
|
|
max_size = settings.get_max_file_size_bytes() |
|
|
|
|
|
if size > max_size: |
|
|
raise ValidationError( |
|
|
f"File size {size} bytes exceeds maximum {max_size} bytes" |
|
|
) |
|
|
|
|
|
|
|
|
content_type = "application/octet-stream" |
|
|
if filename: |
|
|
guessed_type, _ = mimetypes.guess_type(filename) |
|
|
if guessed_type: |
|
|
content_type = guessed_type |
|
|
|
|
|
return { |
|
|
"content": content, |
|
|
"content_type": content_type, |
|
|
"size": size, |
|
|
"filename": filename, |
|
|
"validation_status": "valid" |
|
|
} |
|
|
|
|
|
def _validate_text(self, content: str) -> Dict[str, Any]: |
|
|
"""Validate text input.""" |
|
|
if not isinstance(content, str): |
|
|
raise ValidationError("Text content must be string") |
|
|
|
|
|
if not content.strip(): |
|
|
raise ValidationError("Text is empty") |
|
|
|
|
|
|
|
|
content_bytes = content.encode('utf-8') |
|
|
size = len(content_bytes) |
|
|
|
|
|
max_size = settings.get_max_file_size_bytes() |
|
|
if size > max_size: |
|
|
raise ValidationError( |
|
|
f"Text size {size} bytes exceeds maximum {max_size} bytes" |
|
|
) |
|
|
|
|
|
return { |
|
|
"content": content_bytes, |
|
|
"content_type": "text/plain", |
|
|
"size": size, |
|
|
"filename": None, |
|
|
"validation_status": "valid" |
|
|
} |
