Function test_data_processing

test_python_pipeline.py:163–230

Test complete data processing pipeline

()

    return True

def test_data_processing():
    """Test complete data processing pipeline"""
    print("\n🔄 Testing complete data processing pipeline...")
    
    from processors.data_processor import OceanHazardDataProcessor
    
    # Create test social media posts
    test_posts = [
        {
            "id": "test_001",
            "platform": "twitter",
            "text": "Tsunami warning issued for coastal areas! Evacuate immediately! #tsunami #emergency",
            "user": "test_user",
            "timestamp": datetime.now().isoformat(),
            "likes": 150,
            "retweets": 45,
            "replies": 12,
            "followers": 1000
        },
        {
            "id": "test_002", 
            "platform": "facebook",
            "text": "Heavy rainfall causing flooding in the city. Stay safe everyone!",
            "user": "weather_authority",
            "timestamp": datetime.now().isoformat(),
            "likes": 89,
            "shares": 23,
            "comments": 8,
            "followers": 5000
        },
        {
            "id": "test_003",
            "platform": "instagram",
            "text": "Beautiful sunset at the beach today 🌅",
            "user": "beach_lover",
            "timestamp": datetime.now().isoformat(),
            "likes": 234,
            "comments": 15,
            "followers": 500
        }
    ]
    
    processor = OceanHazardDataProcessor()
    
    # Process the posts
    processed_posts = processor.process_batch(test_posts)
    
    print(f"  Processed {len(processed_posts)} posts")
    
    # Display results
    for i, post in enumerate(processed_posts):
        print(f"\n  Post {i+1} ({post['platform']}):")
        print(f"    Original: {post['original_text'][:50]}...")
        print(f"    Cleaned: {post['cleaned_text'][:50]}...")
        print(f"    Hazard: {post['hazard_analysis']['type']} ({post['hazard_analysis']['urgency']})")
        print(f"    Sentiment: {post['sentiment_analysis']['sentiment_label']}")
        print(f"    Confidence: {post['hazard_analysis']['confidence']:.2f}")