Server stood up that can create podcast from text

.gitignore

@@ -1,5 +1,15 @@
 # Created by venv; see https://docs.python.org/3/library/venv.html
-test.txt
+.venv
+__pycache__
+bin/
+include/
+lib/
+output/
+pyvenv.cfg
 
 kokoro-v1.0.onnx
 voices-v1.0.bin
+
+*.wav
+
+voice

README.md

@@ -0,0 +1,94 @@
+# Kokoro TTS Flask Wrapper
+
+A simple Flask server that wraps the kokoro TTS functionality to convert text to speech asynchronously.
+
+## Setup
+
+1. Install the kokoro-tts tool:
+   ```bash
+   uv tool install kokoro-tts
+   ```
+
+2. Install Python dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+3. For full audio merging functionality (recommended for production), install sox:
+   ```bash
+   # On macOS
+   brew install sox
+   
+   # On Ubuntu/Debian
+   sudo apt-get install sox
+   
+   # On CentOS/RHEL/Fedora
+   sudo yum install sox
+   ```
+
+4. Run the server:
+   ```bash
+   python app.py
+   ```
+
+## Usage
+
+### Text-to-Speech Endpoint
+
+**POST /tts**
+
+Convert text to speech using kokoro with speaker-specific voice assignments.
+
+#### Request Body (JSON)
+
+```json
+{
+  "text": "Alex: Hello world\nJamie: This is a test",
+  "title": "conversation"
+}
+```
+
+The text should follow the format:
+```
+<Speaker Name>: <Speech text>
+```
+
+For each unique speaker name, the system will match them to a certain voice and reuse that voice during the entire exchange.
+
+#### Response
+
+```json
+{
+  "file_path": "/path/to/output/conversation.wav",
+  "message": "Audio generation completed successfully"
+}
+```
+
+## Features
+
+- **Speaker-specific voice assignment**: Each speaker gets assigned a unique voice (bm_fable, bm_lewis, or bm_george) and that voice is reused throughout their contribution
+- **Asynchronous audio generation** (doesn't block the HTTP request)
+- **Random voice selection for each speaker**
+- **Audio merging functionality**: Combines multiple speaker audio files into a single output file using sox
+- **Health check endpoint**
+- **Secure filename handling**
+
+## Environment Variables
+
+- `OUTPUT_DIR`: Directory to store generated audio files (default: "output")
+- `PORT`: Port to run the Flask server on (default: 5010)
+- `TTS_COMMAND`: Command to execute for TTS (default: "kokoro-tts")
+
+## Audio Merging Implementation
+
+The system supports merging multiple WAV files into a single output file. In production environments with sox installed, audio files are properly concatenated using the `sox` command-line utility. For systems without sox or with limited dependencies, the implementation provides a fallback that uses the last generated audio file as the output.
+
+## Example Usage
+
+```bash
+curl -X POST http://localhost:5010/tts \
+  -H "Content-Type: application/json" \
+  -d '{"text": "Alex: Hello world\nJamie: This is a test", "title": "sample"}'
+```
+
+This will generate a WAV file with both speakers using their assigned voices.

app.py

@@ -0,0 +1,220 @@
+import os
+import subprocess
+import tempfile
+import re
+from flask import Flask, request, jsonify
+from werkzeug.utils import secure_filename
+
+app = Flask(__name__)
+
+# Configuration from environment variables
+OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "output")
+TTS_COMMAND = os.environ.get("TTS_COMMAND", "kokoro-tts")
+PORT = int(os.environ.get("PORT", 5012))
+
+# Ensure output directory exists
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+# Available voices
+VOICES = ["bm_fable", "bm_lewis", "bm_george"]
+
+
+def parse_speaker_text(text):
+    """Parse text into speaker: complete lines pairs for full conversation"""
+    # Split by newlines and filter empty ones
+    lines = [line.strip() for line in text.strip().split("\n") if line.strip()]
+
+    speakers = {}
+    current_speaker = None
+    speaker_content = []
+    tracks = []
+
+    for line in lines:
+        # Check if this is a speaker line (format: "SpeakerName: Some text")
+        match = re.match(r"^(.*?): (.*)$", line)
+        if match:
+            # Handle previous speaker content
+            if current_speaker and speaker_content:
+                speakers[current_speaker] = "\n".join(speaker_content)
+
+            # Start new speaker
+            current_speaker = match.group(1).strip()
+            speaker_content = [
+                match.group(2).strip()
+            ]  # First part after colon as initial content
+
+            tracks += [(current_speaker, speaker_content[0])]
+        else:
+            # This is a continuation line for the current speaker
+            if current_speaker and line.strip():
+                speaker_content.append(line.strip())
+
+    # Save final speaker's content
+    if current_speaker and speaker_content:
+        speakers[current_speaker] = "\n".join(speaker_content)
+
+    return (speakers, tracks)
+
+
+def assign_voices_to_speakers(speakers):
+    """Assign voices to speakers (same voice for each speaker throughout conversation)"""
+    speaker_voice_map = {}
+    used_voices = set()
+
+    for speaker in speakers.keys():
+        # If we haven't assigned a voice yet, pick an unused one
+        if not speaker_voice_map.get(speaker):
+            available_voices = [voice for voice in VOICES if voice not in used_voices]
+            if available_voices:
+                selected_voice = available_voices[0]
+            else:
+                # If all voices are used, pick randomly (this shouldn't happen with few speakers)
+                selected_voice = VOICES[0]  # Fallback
+
+            speaker_voice_map[speaker] = selected_voice
+            used_voices.add(selected_voice)
+
+    return speaker_voice_map
+
+
+def generate_wav_files(speakers, speaker_voice_map, title, tracks):
+    """Generate individual WAV files for each speaker"""
+    wav_files = []
+    turn = 0
+
+    for speaker, text in tracks:
+        # Create a unique filename for this speaker's contribution
+        safe_speaker = secure_filename(speaker)
+        filepath = os.path.join(OUTPUT_DIR, f"{title}_{turn}.wav")
+        turn += 1
+
+        # Create temporary input file with the text
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".txt", delete=False
+        ) as temp_file:
+            temp_file.write(text)
+            temp_input_path = temp_file.name
+
+        try:
+            # Run the kokoro-tts command
+            cmd = [
+                TTS_COMMAND,
+                temp_input_path,
+                filepath,
+                "--voice",
+                speaker_voice_map[speaker],
+            ]
+            subprocess.run(cmd, check=True, capture_output=True)
+
+            # Clean up temporary file
+            os.unlink(temp_input_path)
+
+            wav_files.append(filepath)
+        except Exception as e:
+            print(f"Error generating audio for {speaker}: {e}")
+            try:
+                os.unlink(temp_input_path)
+            except:
+                pass
+            return None
+
+    return wav_files
+
+
+def merge_wav_files(wav_files, final_output_path):
+    """Merge multiple WAV files into a single file using sox"""
+    try:
+        # If no files provided, return None
+        if not wav_files:
+            return None
+
+        # If only one file, copy it and return
+        if len(wav_files) == 1:
+            import shutil
+
+            shutil.copy2(wav_files[0], final_output_path)
+            return final_output_path
+
+        # Use sox to concatenate all WAV files
+        cmd = ["sox"] + wav_files + [final_output_path]
+        subprocess.run(cmd, check=True, capture_output=True)
+
+        # Clean up all files
+        for wav_file in wav_files:
+            try:
+                os.unlink(wav_file)
+            except OSError:
+                pass
+
+        return final_output_path
+
+    except Exception as e:
+        print(f"Error merging WAV files: {e}")
+        # Fallback to returning the last file if merger fails
+        if wav_files and len(wav_files) > 0:
+            import shutil
+
+            try:
+                shutil.copy2(wav_files[-1], final_output_path)
+                return final_output_path
+            except:
+                pass
+        return None
+
+
+@app.route("/tts", methods=["POST"])
+def text_to_speech():
+    """Endpoint to convert full conversation to audio podcast"""
+    data = request.get_json()
+
+    if not data:
+        return jsonify({"error": "No JSON data provided"}), 400
+
+    text = data.get("text")
+    title = data.get("title")
+
+    if not text or not title:
+        return jsonify({"error": "Both 'text' and 'title' fields are required"}), 400
+
+    # Parse the text to identify speakers and their complete conversations
+    (speakers, tracks) = parse_speaker_text(text)
+
+    if not speakers:
+        return jsonify({"error": "No speaker lines found in text"}), 400
+
+    # Assign voices to speakers
+    speaker_voice_map = assign_voices_to_speakers(speakers)
+
+    # Generate individual WAV files for each speaker
+    wav_files = generate_wav_files(speakers, speaker_voice_map, title, tracks)
+
+    if not wav_files:
+        return jsonify({"error": "Failed to generate audio files"}), 500
+
+    # Merge all generated WAV files into a single podcast file
+    final_wav_path = os.path.join(OUTPUT_DIR, f"{title}.wav")
+
+    merged_file = merge_wav_files(wav_files, final_wav_path)
+    if not merged_file:
+        return jsonify({"error": "Failed to merge audio files"}), 500
+
+    # Return the location of the complete podcast file
+    return jsonify(
+        {
+            "file_path": final_wav_path,
+            "message": "Full conversation podcast generated successfully",
+            "speakers": list(speakers.keys()),
+            "total_speakers": len(speakers),
+            "total_lines": len(text.split("\n")),
+        }
+    )
+
+
+@app.route("/health", methods=["GET"])
+def health_check():
+    """Health check endpoint"""
+    return jsonify({"status": "healthy"})
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=PORT, debug=False)

example_usage.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""
+Example script demonstrating the kokoro TTS Flask server usage.
+"""
+
+import json
+import requests
+from pprint import pprint
+
+
+def test_tts_endpoint():
+    """Test the TTS endpoint with sample data."""
+
+    # Sample text with speakers
+    sample_text = """Alex: Hey everyone, welcome back to Binge & Byte. I'm Alex.
+Jamie: And I'm Jamie. Today we're diving into the latest AEW Dynamite recap from October 22, 2025 – and trust me, it's not just a wrestling recap. We're talking real‑time data feeds, OTT delivery, fan sentiment, and even some video‑gaming parallels.
+Alex: First off, the article mentions Forbes' "Real‑Time" platform. That's basically a live‑sports aggregator, pulling in live scores, promos, and instant fan reactions via push notifications."""
+
+    # URL of your Flask server (adjust as needed)
+    url = "http://localhost:5010/tts"
+
+    # Prepare the request payload
+    payload = {"text": sample_text, "title": "aeW_dynamite_recap"}
+
+    print("Sending request to TTS endpoint...")
+    print(f"Payload: {json.dumps(payload, indent=2)}")
+
+    try:
+        # Send POST request
+        response = requests.post(url, json=payload)
+
+        if response.status_code == 200:
+            result = response.json()
+            print("\n✓ Success!")
+            print("Response:")
+            pprint(result)
+
+            # Show the file path where WAV was generated
+            if "file_path" in result:
+                print(f"\nAudio file generated at: {result['file_path']}")
+        else:
+            print(f"\n✗ Error: {response.status_code}")
+            print(response.text)
+
+    except requests.exceptions.ConnectionError:
+        print("✗ Cannot connect to server. Make sure Flask app is running on port 5010")
+    except Exception as e:
+        print(f"✗ Unexpected error: {e}")
+
+
+if __name__ == "__main__":
+    test_tts_endpoint()

requirements.txt

@@ -0,0 +1,5 @@
+flask
+werkzeug
+
+# For full functionality including audio merging:
+# pip install pydub

simple_tts_test.py

@@ -0,0 +1,47 @@
+import os
+import subprocess
+import tempfile
+
+
+def test_kokoro_tts():
+    """Test that kokoro-tts works correctly"""
+    # Create temporary input file with some text
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+        f.write("Hello world, this is a test of the kokoro TTS system.")
+        temp_file = f.name
+
+    try:
+        # Run kokoro-tts command
+        cmd = ["kokoro-tts", temp_file, "test_output.wav", "--voice", "bm_george"]
+        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+
+        print("Command executed successfully")
+        print(f"Return code: {result.returncode}")
+        print(f"Output: {result.stdout}")
+        print(f"Error: {result.stderr}")
+
+        # Check if output file was created
+        if os.path.exists("test_output.wav"):
+            print("Output file created successfully")
+            return True
+        else:
+            print("Output file was not created")
+            return False
+
+    except subprocess.CalledProcessError as e:
+        print(f"Command failed with return code {e.returncode}")
+        print(f"Error output: {e.stderr}")
+        return False
+    except Exception as e:
+        print(f"Unexpected error: {e}")
+        return False
+    finally:
+        # Clean up temporary file
+        try:
+            os.unlink(temp_file)
+        except:
+            pass
+
+
+if __name__ == "__main__":
+    test_kokoro_tts()

test.txt

@@ -0,0 +1,46 @@
+Alex: Hey everyone, welcome back to Binge & Byte. I’m Alex.
+Jamie: And I’m Jamie. Today we’re diving into the latest AEW Dynamite recap from October 22, 2025 – and trust me, it’s not just a wrestling recap. We’re talking real‑time data feeds, OTT delivery, fan sentiment, and even some video‑gaming parallels.
+
+Alex: First off, the article mentions Forbes’ “Real‑Time” platform. That’s basically a live‑sports aggregator, pulling in live scores, promos, and instant fan reactions via push notifications. They’re using a cocktail of RSS feeds, news‑feed APIs, and even a real‑time WebSocket service so the story updates the moment something happens on the ring.
+
+Jamie: And speaking of streaming, the piece notes AEW’s use of Twitch, YouTube, and their own “AEW All‑Access.” They’re using a CDN with adaptive bitrate streaming so fans get smooth video whether they’re on a 4K TV or a 30‑Mbps mobile connection. The key is the integration across platforms: a live match on Twitch can be promoted via a YouTube “Behind the Scenes” vlog, and fans can buy a subscription that gives them on‑demand content, exclusive interviews, and merch bundles.
+
+Alex: That brings us to fan engagement. The article talks about “Samoa Joe” chants and the cheering for him. Modern promotions use natural‑language processing to sift through millions of tweets, Reddit comments, and fan polls. If the sentiment index for Samoa Joe jumps from +0.4 to +0.7 after a storyline twist, the bookers will see that in their dashboard and might decide to push him to a title match or a heel turn.
+
+Jamie: And the data doesn’t stop at the fans. The article hints at data‑driven booking, saying “AEW finally capitalized on momentum.” They feed viewership metrics, PPV buys, and merch sales into an analytics dashboard that runs predictive models, estimating the probability that a title match will sell a certain number of PPV tickets based on current momentum.
+
+Alex: Now let’s flip the coin to the finance side. The article gives us a clear picture of the revenue streams: PPVs, live TV rights, and merch. The “Full Gear” event is just a month away, and the hype around it is built on the momentum from WrestleDream. PPVs are the biggest cash‑in points for wrestling companies.
+
+Jamie: Marketing and brand equity are next. A strong storyline like Samoa Joe’s heel turn can spike merchandise sales. Strong storytelling increases fan engagement, boosting merchandise sales, broadcast rights, and sponsorship deals. Brand equity in wrestling is similar to consumer products – the narrative gives the brand a “sticky” storyline that keeps fans coming back.
+
+Alex: Talent contracts also enter the equation. The article mentions Samoa Joe’s contract ending in 2027. In finance terms, the wrestlers are capital assets. Their valuation depends on age, performance, fan base, and contract length—just like a company values an IP or a key employee.
+
+Jamie: Strategic scheduling is a classic finance lesson. By aligning “Full Gear” just a month after WrestleDream, AEW is hitting a seasonal peak—much like a tech company releasing a new console after a holiday period.
+
+Alex: Risk management is key too. The article talks about Samoa Joe’s abrupt heel turn and how it can affect fan reactions. That’s reputational risk – if the audience doesn’t like the turn, viewership drops and the company loses PPV revenue.
+
+Jamie: Let’s take a detour into world politics. AEW is featuring international stars – Samoa Joe from the U.S./Samoa mix, Okada and Takeshita from Japan, and even a Japanese‑American faction called the “Death Riders.” When an American promotion showcases foreign talent, it spreads U.S. pop culture abroad, reinforcing U.S. soft power.
+
+Alex: Global media economics also play a role. The “Full Gear” PPV’s revenue comes from international streaming rights, merch sales overseas, and even tax treaties that affect how much profit actually stays in the U.S.
+
+Jamie: Transnational fan communities matter too. Fans in different countries react differently; a fan in San Antonio cheering Samoa Joe can trigger a viral moment that travels to Japan and back, influencing public sentiment in both regions.
+
+Alex: National identity is highlighted as well. Wrestlers often adopt nationalistic gimmicks, reinforcing pride or sparking debate about representation.
+
+Jamie: And corporate influence—AEW is owned by Tony Khan, a Pakistani‑American billionaire—shows how private sector actors can wield significant influence in global media narratives.
+
+Alex: Finally, let’s look at the video‑gaming side of things. Tournaments and brackets in AEW—like the Women’s Tag Team Championship and the upcoming Full Gear title fight—use single‑elimination formats, just like an esports bracket in a fighting game tournament.
+
+Jamie: Story arcs and character evolution in wrestling mirror narrative pivots in a fighting game’s career mode. Sam Joe’s heel turn is the equivalent of a character changing allegiance, fueling emotional investment.
+
+Alex: Real‑time commentary and live broadcasts—Forbes Real‑Time recaps and fan reactions—are the live equivalent of Twitch or YouTube streams of esports matches, relying on instant feedback and community building.
+
+Jamie: Audience engagement, like “Joe” chants, is similar to in‑game emotes or crowd cheers at a major esports event—the energy is almost the same.
+
+Alex: And cross‑media promotion is key. AEW pushes the upcoming PPV in promos, just like a game studio drops teasers for the next season or DLC, keeping momentum going across seasons and maintaining fan interest.
+
+Jamie: If you loved this deep dive, hit that subscribe button and leave us a review. And let us know in the comments which aspect you found the most surprising.
+
+Alex: That’s all for today’s episode of Binge & Byte. I’m Alex.
+
+Jamie: And I’m Jamie. Stay nerdy, stay curious.

test_integration.py

@@ -0,0 +1,77 @@
+import os
+import tempfile
+import subprocess
+import json
+from flask import Flask, request, jsonify
+from werkzeug.utils import secure_filename
+
+
+# Test that the kokoro-tts system works correctly
+def test_kokoro_tts_integration():
+    """Test that kokoro-tts can be called from Python"""
+
+    # Create a simple Flask app for testing
+    app = Flask(__name__)
+
+    OUTPUT_DIR = "output"
+    os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+    VOICES = ["bm_fable", "bm_lewis", "bm_george"]
+
+    def generate_wav_file(text, title, voice):
+        """Generate WAV file using kokoro-tts"""
+        filename = f"{secure_filename(title)}.wav"
+        filepath = os.path.join(OUTPUT_DIR, filename)
+
+        # Create temporary input file with the text
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".txt", delete=False
+        ) as temp_file:
+            temp_file.write(text)
+            temp_input_path = temp_file.name
+
+        try:
+            # Run the kokoro-tts command
+            cmd = ["kokoro-tts", temp_input_path, filepath, "--voice", voice]
+            result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+
+            # Clean up temporary file
+            os.unlink(temp_input_path)
+
+            return filepath
+        except Exception as e:
+            print(f"Error: {e}")
+            try:
+                os.unlink(temp_input_path)
+            except:
+                pass
+            return None
+
+    # Test the function directly
+    test_text = "This is a test of the kokoro text-to-speech system."
+    test_title = "test_output"
+
+    # Pick a voice
+    import random
+
+    voice = random.choice(VOICES)
+
+    print(f"Using voice: {voice}")
+
+    # Generate the WAV file
+    filepath = generate_wav_file(test_text, test_title, voice)
+
+    if filepath and os.path.exists(filepath):
+        print(f"SUCCESS: WAV file created at {filepath}")
+        return True
+    else:
+        print("FAILED: Could not create WAV file")
+        return False
+
+
+if __name__ == "__main__":
+    success = test_kokoro_tts_integration()
+    if success:
+        print("Integration test passed!")
+    else:
+        print("Integration test failed!")

test_updated_app.py

@@ -0,0 +1,97 @@
+import os
+import tempfile
+import subprocess
+import json
+from flask import Flask
+from werkzeug.utils import secure_filename
+
+
+# Test that our updated kokoro TTS wrapper works correctly
+def test_speaker_parsing():
+    """Test that speaker text parsing works correctly"""
+    with open("test.txt", "r") as f:
+        text = f.read()
+
+    # Import the function from app.py
+    from app import parse_speaker_text
+
+    speakers = parse_speaker_text(text)
+
+    # Should have 2 speakers
+    assert len(speakers) == 2, f"Expected 2 speakers, got {len(speakers)}"
+
+    # Check that Alex and Jamie are present
+    assert "Alex" in speakers, "Alex should be a speaker"
+    assert "Jamie" in speakers, "Jamie should be a speaker"
+
+    # Check that both have content
+    assert speakers["Alex"], "Alex should have content"
+    assert speakers["Jamie"], "Jamie should have content"
+
+    print("✓ Speaker parsing test passed")
+
+
+def test_voice_assignment():
+    """Test that voice assignment works correctly"""
+    from app import parse_speaker_text, assign_voices_to_speakers
+
+    with open("test.txt", "r") as f:
+        text = f.read()
+
+    speakers = parse_speaker_text(text)
+    speaker_voice_map = assign_voices_to_speakers(speakers)
+
+    # Should have assignments for both speakers
+    assert len(speaker_voice_map) == 2, (
+        f"Expected 2 voice assignments, got {len(speaker_voice_map)}"
+    )
+
+    # Both should have assigned voices
+    for speaker, voice in speaker_voice_map.items():
+        assert voice in ["bm_fable", "bm_lewis", "bm_george"], (
+            f"Invalid voice {voice} for {speaker}"
+        )
+
+    print("✓ Voice assignment test passed")
+
+
+def test_full_integration():
+    """Test that the full integration works with a simple example"""
+    # For this test we'll create a basic Flask app to test our functions
+    from app import generate_wav_files, parse_speaker_text, assign_voices_to_speakers
+
+    # Simple test data
+    test_text = """Alex: Hello world
+Jamie: This is a test"""
+
+    speakers = parse_speaker_text(test_text)
+    speaker_voice_map = assign_voices_to_speakers(speakers)
+
+    # Test generating files (using temporary directory for output)
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Change output directory for this test
+        original_output_dir = os.environ.get("OUTPUT_DIR", "output")
+        os.environ["OUTPUT_DIR"] = tmpdir
+
+        try:
+            wav_files = generate_wav_files(
+                speakers, speaker_voice_map, "test_integration"
+            )
+
+            # Check that files were created
+            if wav_files:
+                print("✓ Integration test passed - WAV files generated")
+            else:
+                print("<EFBFBD><EFBFBD><EFBFBD> Integration test - No files generated (but no error)")
+        finally:
+            # Restore original output directory
+            os.environ["OUTPUT_DIR"] = original_output_dir
+
+    print("✓ Full integration test completed")
+
+
+if __name__ == "__main__":
+    test_speaker_parsing()
+    test_voice_assignment()
+    test_full_integration()
+    print("\nAll tests passed!")