diff --git a/backend/test_end_to_end_ai.py b/backend/test_end_to_end_ai.py new file mode 100644 index 0000000..e7bf04e --- /dev/null +++ b/backend/test_end_to_end_ai.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +""" +End-to-end test demonstrating AI-powered audiobook generation. +Shows the complete workflow from unmarked text to multi-voice audio. +""" +import asyncio +import sys +from services.role_detector import RoleDetector +from services.tts_service import TTSService +from services.ollama_service import OllamaService + + +async def test_complete_workflow(): + """Test the complete AI-powered audiobook generation workflow.""" + print("\n" + "="*70) + print("END-TO-END AI-POWERED AUDIOBOOK GENERATION TEST") + print("="*70 + "\n") + + # Sample story without any role markers + story = """The old wizard Merlin gazed into his crystal ball with concern. +"Dark times are coming," he whispered to himself. + +A young knight named Arthur entered the chamber. "Master Merlin, I heard your summons," he said with determination. + +Merlin turned to face him. "Arthur, the kingdom needs you now more than ever." + +Arthur drew his sword. "I am ready to defend our people." + +From the shadows, the sorceress Morgana watched silently. She knew the prophecy would unfold exactly as foretold.""" + + print("STORY TEXT (unmarked narrative):") + print("-" * 70) + print(story) + print("-" * 70 + "\n") + + # Initialize services + ollama_service = OllamaService() + role_detector = RoleDetector(ollama_service=ollama_service) + tts_service = TTSService(ollama_service=ollama_service) + + # Check Ollama availability + is_available = await ollama_service.is_available() + if not is_available: + print("❌ Ollama is not available. This test requires Ollama.") + print(" Please start Ollama with: ollama serve") + return False + + print("✓ Ollama service is available\n") + + # Step 1: Detect roles using AI + print("STEP 1: AI Role Detection") + print("-" * 70) + try: + roles = await role_detector.detect_roles(story) + print(f"✓ Detected {len(roles)} roles:\n") + for role in roles: + print(f" - {role['label']}") + print() + except Exception as e: + print(f"❌ Role detection failed: {e}") + return False + + # Step 2: User assigns voices (simulated) + print("STEP 2: Voice Assignment (simulated user input)") + print("-" * 70) + role_voice_map = {} + voice_options = ["af_bella", "am_adam", "am_michael", "af_sarah", "bf_emma"] + + for i, role in enumerate(roles): + voice = voice_options[i % len(voice_options)] + role_voice_map[role['id']] = voice + print(f" {role['label']} → {voice}") + print() + + # Step 3: AI text segmentation + print("STEP 3: AI Text Segmentation") + print("-" * 70) + try: + segments = await tts_service.segment_text_with_ai(story, role_voice_map) + + if segments is None: + print("❌ AI segmentation failed") + return False + + print(f"✓ AI segmented text into {len(segments)} segments:\n") + + for i, (speaker, voice, text) in enumerate(segments, 1): + # Truncate long text for display + display_text = text[:60] + "..." if len(text) > 60 else text + print(f" {i}. [{speaker}] ({voice})") + print(f" \"{display_text}\"") + print() + + # Verify intelligent segmentation + speakers_used = set(seg[0] for seg in segments) + print(f"Speakers used: {speakers_used}") + + # Check if dialogue was properly assigned + has_multiple_speakers = len(speakers_used) > 1 + has_narrator = "Narrator" in speakers_used or any("Merlin" in s or "Arthur" in s or "Morgana" in s for s in speakers_used) + + if has_multiple_speakers and has_narrator: + print("✓ AI successfully assigned dialogue to characters and narrative to appropriate speakers") + else: + print("⚠ Segmentation may not be optimal") + print() + + except Exception as e: + print(f"❌ Segmentation failed: {e}") + import traceback + traceback.print_exc() + return False + + # Step 4: Verify segmentation quality + print("STEP 4: Segmentation Quality Analysis") + print("-" * 70) + + # Count segments per speaker + speaker_counts = {} + for speaker, _, _ in segments: + speaker_counts[speaker] = speaker_counts.get(speaker, 0) + 1 + + print("Segments per speaker:") + for speaker, count in sorted(speaker_counts.items(), key=lambda x: x[1], reverse=True): + print(f" {speaker}: {count} segments") + print() + + # Check for dialogue detection + dialogue_keywords = ['"', "'", "said", "whispered", "replied"] + dialogue_segments = 0 + for _, _, text in segments: + if any(kw in text.lower() for kw in dialogue_keywords): + dialogue_segments += 1 + + print(f"Dialogue segments detected: {dialogue_segments}/{len(segments)}") + print() + + # Summary + print("="*70) + print("TEST SUMMARY") + print("="*70) + print(f"✓ Role Detection: {len(roles)} roles detected") + print(f"✓ Voice Assignment: {len(role_voice_map)} voices assigned") + print(f"✓ Text Segmentation: {len(segments)} segments created") + print(f"✓ Speaker Distribution: {len(speakers_used)} unique speakers used") + print(f"✓ Dialogue Detection: {dialogue_segments} dialogue segments") + print() + print("🎉 AI-POWERED AUDIOBOOK GENERATION: SUCCESS!") + print("="*70 + "\n") + + return True + + +async def main(): + """Run the end-to-end test.""" + try: + success = await test_complete_workflow() + return success + except Exception as e: + print(f"\n❌ Test failed with exception: {e}") + import traceback + traceback.print_exc() + return False + + +if __name__ == '__main__': + success = asyncio.run(main()) + sys.exit(0 if success else 1)