| """ |
| Main Entry Point |
| Demonstrates orchestrator usage with example workflows. |
| No business logic - just a thin interface layer. |
| """ |
| from dotenv import load_dotenv |
| load_dotenv() |
|
|
| from core.orchestrator import Orchestrator |
|
|
|
|
| print(">>> MAIN FILE LOADED <<<") |
|
|
|
|
| def example_text_proof(): |
| """Example: Create proof from text content.""" |
| print("\n=== Example 1: Text Proof Creation ===") |
| |
| orchestrator = Orchestrator() |
| |
| |
| result = orchestrator.create_proof({ |
| "type": "text", |
| "content": "This is a confidential document that needs timestamping." |
| }) |
| |
| if result["success"]: |
| print(f"✓ Proof created: {result['proof_id']}") |
| print(f" Hash: {result['proof'].content_hash}") |
| print(f" Timestamp: {result['proof'].timestamp}") |
| |
| |
| if "assistant" in result: |
| print(f"\n 🤖 AI Explanation:") |
| print(f" {result['assistant']['response']}") |
| else: |
| print(f"✗ Failed: {result['message']}") |
| |
| return result |
|
|
|
|
| def example_file_proof(): |
| """Example: Create proof from file content.""" |
| print("\n=== Example 2: File Proof Creation ===") |
| |
| orchestrator = Orchestrator() |
| |
| |
| file_content = b"Binary file content here" |
| |
| result = orchestrator.create_proof({ |
| "type": "file", |
| "content": file_content, |
| "filename": "document.pdf" |
| }) |
| |
| if result["success"]: |
| print(f"✓ Proof created: {result['proof_id']}") |
| print(f" Hash: {result['proof'].content_hash}") |
| print(f" File: {result['proof'].metadata['filename']}") |
| print(f" OCR Status: {result['proof'].ocr_status}") |
| if result['proof'].extracted_text: |
| print(f" Extracted Text: {result['proof'].extracted_text[:100]}...") |
| else: |
| print(f"✗ Failed: {result['message']}") |
| |
| return result |
|
|
|
|
| def example_image_ocr(): |
| """Example: Create proof from image with OCR.""" |
| print("\n=== Example 5: Image Proof with OCR ===") |
| |
| orchestrator = Orchestrator() |
| |
| |
| |
| print("Note: This example requires actual image bytes with text.") |
| print("Skipping OCR demo - install Tesseract and provide real image to test.") |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| print("✓ OCR integration ready for image inputs") |
|
|
|
|
| def example_ai_assistant(): |
| """Example: Ask AI assistant about proofs.""" |
| print("\n=== Example 6: AI Assistant Q&A ===") |
| |
| orchestrator = Orchestrator() |
| |
| if not orchestrator.ai_sidecar.enabled: |
| print("⚠️ AI assistant is disabled") |
| print(" Enable with: AI_ENABLED=true GEMINI_API_KEY=your-key") |
| return |
| |
| |
| create_result = orchestrator.create_proof({ |
| "type": "text", |
| "content": "Important contract signed on December 24, 2024" |
| }) |
| |
| if not create_result["success"]: |
| print("Failed to create proof for demo") |
| return |
| |
| proof_id = create_result["proof_id"] |
| |
| |
| questions = [ |
| "What does this proof guarantee?", |
| "How can I verify this proof later?", |
| "What should I do with this proof ID?" |
| ] |
| |
| for question in questions: |
| print(f"\n Q: {question}") |
| result = orchestrator.ask_assistant(question, proof_id) |
| |
| if result["success"]: |
| print(f" 🤖 A: {result['assistant']['response']}") |
| else: |
| print(f" ✗ {result['message']}") |
|
|
|
|
| def example_verification(): |
| """Example: Verify an existing proof.""" |
| print("\n=== Example 3: Proof Verification ===") |
| |
| orchestrator = Orchestrator() |
| |
| |
| original_content = "Verify this content" |
| create_result = orchestrator.create_proof({ |
| "type": "text", |
| "content": original_content |
| }) |
| |
| if not create_result["success"]: |
| print("Failed to create proof for verification") |
| return |
| |
| proof_id = create_result["proof_id"] |
| print(f"Created proof: {proof_id}") |
| |
| |
| verify_result = orchestrator.verify_proof( |
| proof_id, |
| original_content.encode('utf-8') |
| ) |
| |
| if verify_result["success"]: |
| vr = verify_result["verification_result"] |
| status = "✓ VALID" if vr.is_valid else "✗ INVALID" |
| print(f"{status}: {vr.message}") |
| |
| |
| if "assistant" in verify_result: |
| print(f"\n 🤖 AI Explanation:") |
| print(f" {verify_result['assistant']['response']}") |
| else: |
| print(f"✗ Verification failed: {verify_result['message']}") |
| |
| |
| print("\nAttempting verification with tampered content:") |
| tampered_result = orchestrator.verify_proof( |
| proof_id, |
| b"Tampered content" |
| ) |
| |
| if tampered_result["success"]: |
| vr = tampered_result["verification_result"] |
| status = "✓ VALID" if vr.is_valid else "✗ INVALID" |
| print(f"{status}: {vr.message}") |
|
|
|
|
| def example_retrieval(): |
| """Example: Retrieve a stored proof.""" |
| print("\n=== Example 4: Proof Retrieval ===") |
| |
| orchestrator = Orchestrator() |
| |
| |
| create_result = orchestrator.create_proof({ |
| "type": "text", |
| "content": "Retrieve this later" |
| }) |
| |
| if not create_result["success"]: |
| print("Failed to create proof") |
| return |
| |
| proof_id = create_result["proof_id"] |
| |
| |
| get_result = orchestrator.get_proof(proof_id) |
| |
| if get_result["success"]: |
| proof = get_result["proof"] |
| print(f"✓ Retrieved proof: {proof.proof_id}") |
| print(f" Hash: {proof.content_hash}") |
| print(f" Size: {proof.content_size} bytes") |
| print(f" Created: {proof.timestamp}") |
| else: |
| print(f"✗ Failed: {get_result['message']}") |
|
|
|
|
| def main(): |
| """Run all examples.""" |
| print("=" * 60) |
| print("PROOF-OF-EXISTENCE SYSTEM - Priority-3 MVP") |
| print("Deterministic Core + OCR + AI Sidecar") |
| print("=" * 60) |
| |
| try: |
| example_text_proof() |
| example_file_proof() |
| example_verification() |
| example_retrieval() |
| example_image_ocr() |
| example_ai_assistant() |
| |
| print("\n" + "=" * 60) |
| print("All examples completed successfully!") |
| print("=" * 60) |
| |
| except Exception as e: |
| print(f"\n✗ Error running examples: {str(e)}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
