|
|
""" |
|
|
Metadata Agent |
|
|
Generates structured metadata for proof objects. |
|
|
""" |
|
|
|
|
|
from typing import Dict, Any |
|
|
from datetime import datetime, timezone |
|
|
|
|
|
from core.agent_base import Agent |
|
|
from core.errors import MetadataError |
|
|
|
|
|
|
|
|
class MetadataAgent(Agent): |
|
|
""" |
|
|
Creates comprehensive metadata for proof generation. |
|
|
""" |
|
|
|
|
|
def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: |
|
|
""" |
|
|
Generate metadata object. |
|
|
|
|
|
Expected input_data: |
|
|
{ |
|
|
"content_type": str, |
|
|
"size": int, |
|
|
"filename": str | None, |
|
|
"content_hash": str, |
|
|
"hash_algorithm": str, |
|
|
"hash_timestamp": str, |
|
|
"validation_status": str, |
|
|
... |
|
|
} |
|
|
|
|
|
Returns: |
|
|
{ |
|
|
"metadata": { |
|
|
"content_type": str, |
|
|
"content_size": int, |
|
|
"filename": str | None, |
|
|
"hash_reference": str, |
|
|
"created_at": str, |
|
|
"validation_status": str, |
|
|
}, |
|
|
...passes through input_data... |
|
|
} |
|
|
""" |
|
|
try: |
|
|
metadata = { |
|
|
"content_type": input_data.get("content_type"), |
|
|
"content_size": input_data.get("size"), |
|
|
"filename": input_data.get("filename"), |
|
|
"hash_reference": input_data.get("content_hash"), |
|
|
"hash_algorithm": input_data.get("hash_algorithm"), |
|
|
"created_at": datetime.now(timezone.utc).isoformat(), |
|
|
"validation_status": input_data.get("validation_status"), |
|
|
} |
|
|
|
|
|
|
|
|
required = ["content_type", "content_size", "hash_reference"] |
|
|
missing = [k for k in required if metadata.get(k) is None] |
|
|
|
|
|
if missing: |
|
|
raise MetadataError(f"Missing required fields: {', '.join(missing)}") |
|
|
|
|
|
result = input_data.copy() |
|
|
result["metadata"] = metadata |
|
|
|
|
|
return result |
|
|
|
|
|
except Exception as e: |
|
|
if isinstance(e, MetadataError): |
|
|
raise |
|
|
raise MetadataError(f"Metadata generation failed: {str(e)}") from e |
