agentUniverse/examples/json_reader_demo.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

"""
JsonReader Demo - Demonstrates how to use JsonReader to load JSON files as knowledge.

This example shows:
1. Basic JSON object loading
2. JSON array loading
3. Loading with custom metadata
4. Integration with knowledge base
"""

import json
import tempfile
from pathlib import Path

from agentuniverse.agent.action.knowledge.reader.file.json_reader import JsonReader


def demo_basic_usage():
    """Basic usage of JsonReader"""
    print("=" * 60)
    print("Demo 1: Basic JSON Object Loading")
    print("=" * 60)

    # Create a sample JSON file
    sample_data = {
        "project": "agentUniverse",
        "version": "0.0.19",
        "features": ["knowledge", "reader", "json"],
        "config": {
            "enabled": True,
            "max_size": 1024
        }
    }

    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        json.dump(sample_data, f, ensure_ascii=False, indent=2)
        temp_path = f.name

    try:
        # Load JSON file using JsonReader
        reader = JsonReader()
        documents = reader.load_data(temp_path)

        print(f"Loaded {len(documents)} document(s)")
        print(f"\nDocument text (first 200 chars):")
        print(documents[0].text[:200] + "...")
        print(f"\nDocument metadata:")
        for key, value in documents[0].metadata.items():
            print(f"  {key}: {value}")

    finally:
        Path(temp_path).unlink()


def demo_array_loading():
    """Loading JSON array"""
    print("\n" + "=" * 60)
    print("Demo 2: JSON Array Loading")
    print("=" * 60)

    users = [
        {"id": 1, "name": "Alice", "role": "admin"},
        {"id": 2, "name": "Bob", "role": "user"},
        {"id": 3, "name": "Charlie", "role": "user"}
    ]

    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        json.dump(users, f, ensure_ascii=False, indent=2)
        temp_path = f.name

    try:
        reader = JsonReader()
        documents = reader.load_data(temp_path)

        print(f"Loaded {len(documents)} document(s)")
        print(f"JSON type: {documents[0].metadata['json_type']}")
        print(f"\nContent preview:")
        print(documents[0].text[:150] + "...")

    finally:
        Path(temp_path).unlink()


def demo_with_metadata():
    """Loading with custom metadata"""
    print("\n" + "=" * 60)
    print("Demo 3: Loading with Custom Metadata")
    print("=" * 60)

    config_data = {
        "database": {
            "host": "localhost",
            "port": 5432,
            "name": "agentuniverse_db"
        },
        "cache": {
            "enabled": True,
            "ttl": 3600
        }
    }

    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        json.dump(config_data, f, ensure_ascii=False, indent=2)
        temp_path = f.name

    try:
        reader = JsonReader()
        # Add custom metadata
        ext_info = {
            "source": "config_loader",
            "environment": "development",
            "priority": "high"
        }
        documents = reader.load_data(temp_path, ext_info=ext_info)

        print(f"Document metadata:")
        for key, value in documents[0].metadata.items():
            print(f"  {key}: {value}")

    finally:
        Path(temp_path).unlink()


def demo_unicode_support():
    """Demonstrating Unicode support"""
    print("\n" + "=" * 60)
    print("Demo 4: Unicode Support (中文支持)")
    print("=" * 60)

    multilingual_data = {
        "title": "多语言知识库",
        "description": "agentUniverse supports multiple languages",
        "languages": ["English", "中文", "日本語", "한국어"],
        "emoji": "🚀 🎉 ✨",
        "content": {
            "en": "Knowledge base system",
            "zh": "知识库系统",
            "ja": "知識ベースシステム"
        }
    }

    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        json.dump(multilingual_data, f, ensure_ascii=False, indent=2)
        temp_path = f.name

    try:
        reader = JsonReader()
        documents = reader.load_data(temp_path)

        print("Unicode characters are properly preserved:")
        print(documents[0].text[:300])

    finally:
        Path(temp_path).unlink()


def demo_error_handling():
    """Demonstrating error handling"""
    print("\n" + "=" * 60)
    print("Demo 5: Error Handling")
    print("=" * 60)

    reader = JsonReader()

    # Test 1: Invalid JSON
    print("\n1. Testing invalid JSON handling:")
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        f.write('{"invalid": json syntax}')
        temp_path = f.name

    try:
        reader.load_data(temp_path)
        print("  ❌ Should have raised JSONDecodeError")
    except json.JSONDecodeError as e:
        print(f"  ✓ Correctly caught JSONDecodeError")
    finally:
        Path(temp_path).unlink()

    # Test 2: Non-existent file
    print("\n2. Testing non-existent file handling:")
    try:
        reader.load_data("/tmp/nonexistent_file_12345.json")
        print("  ❌ Should have raised FileNotFoundError")
    except FileNotFoundError:
        print(f"  ✓ Correctly caught FileNotFoundError")


if __name__ == "__main__":
    print("\n" + "🚀 " * 20)
    print("JsonReader Demonstration")
    print("🚀 " * 20 + "\n")

    demo_basic_usage()
    demo_array_loading()
    demo_with_metadata()
    demo_unicode_support()
    demo_error_handling()

    print("\n" + "=" * 60)
    print("All demos completed successfully! ✨")
    print("=" * 60 + "\n")