chore: remove local artifacts

Reviewed-on: https://git.home.ms/jarianc/kokorotts-server/pulls/32

app.py

@@ -2,15 +2,35 @@ import os
 import subprocess
 import tempfile
 import re
+import logging
+import glob
+from pathlib import Path
 from flask import Flask, request, jsonify
 from werkzeug.utils import secure_filename
 
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 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))
+API_KEY = os.environ.get("TTS_API_KEY", "")
+
+# Issue #10: Whitelist allowed TTS commands
+ALLOWED_TTS_COMMANDS = {"kokoro-tts", "/usr/local/bin/kokoro-tts", "/usr/bin/kokoro-tts"}
+_tts_command = os.environ.get("TTS_COMMAND", "kokoro-tts")
+if _tts_command not in ALLOWED_TTS_COMMANDS:
+    logger.warning(
+        f"TTS_COMMAND '{_tts_command}' not in whitelist, using default 'kokoro-tts'"
+    )
+TTS_COMMAND = "kokoro-tts"
+else:
+    TTS_COMMAND = _tts_command
+
+# Issue #12: Maximum number of output files to keep per title
+MAX_OUTPUT_FILES_PER_TITLE = 5
 
 # Ensure output directory exists
 os.makedirs(OUTPUT_DIR, exist_ok=True)
@@ -21,7 +41,6 @@ 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 = {}
@@ -30,26 +49,18 @@ def parse_speaker_text(text):
     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 = [
+            speaker_content = [match.group(2).strip()]
-                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)
 
@@ -62,14 +73,12 @@ def assign_voices_to_speakers(speakers):
     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]
-                selected_voice = VOICES[0]  # Fallback
 
             speaker_voice_map[speaker] = selected_voice
             used_voices.add(selected_voice)
@@ -77,26 +86,37 @@ def assign_voices_to_speakers(speakers):
     return speaker_voice_map
 
 
+def clean_old_outputs(title):
+    """Issue #12: Remove old output files for a given title, keeping only the newest."""
+    pattern = os.path.join(OUTPUT_DIR, f"{title}_*.wav")
+    files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True)
+    for old_file in files[MAX_OUTPUT_FILES_PER_TITLE:]:
+        try:
+            os.unlink(old_file)
+            logger.info(f"Removed old output file: {old_file}")
+        except OSError as e:
+            logger.warning(f"Failed to remove old output file {old_file}: {e}")
+
+
 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
+        temp_input_path = None
-        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
+            # Issue #11: Create temporary input file with proper cleanup
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".txt", delete=False
+            ) as temp_file:
+                temp_file.write(text)
+                temp_input_path = temp_file.name
+
             cmd = [
                 TTS_COMMAND,
                 temp_input_path,
@@ -106,17 +126,19 @@ def generate_wav_files(speakers, speaker_voice_map, title, tracks):
             ]
             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}")
+            logger.error(f"Error generating audio for {speaker}: {e}")
-            try:
-                os.unlink(temp_input_path)
-            except:
-                pass
             return None
+        finally:
+            # Issue #11: Always clean up temp file, using specific exception type
+            if temp_input_path and os.path.exists(temp_input_path):
+                try:
+                    os.unlink(temp_input_path)
+                except OSError as cleanup_err:
+                    logger.warning(
+                        f"Failed to clean up temp file {temp_input_path}: {cleanup_err}"
+                    )
 
     return wav_files
 
@@ -124,47 +146,61 @@ def generate_wav_files(speakers, speaker_voice_map, title, tracks):
 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
+        # Clean up intermediate files
         for wav_file in wav_files:
             try:
                 os.unlink(wav_file)
-            except OSError:
+            except OSError as e:
-                pass
+                logger.warning(f"Failed to remove intermediate file {wav_file}: {e}")
 
         return final_output_path
 
     except Exception as e:
-        print(f"Error merging WAV files: {e}")
+        logger.error(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:
+            except OSError as fallback_err:
-                pass
+                logger.error(
+                    f"Fallback copy failed for {wav_files[-1]}: {fallback_err}"
+                )
         return None
 
 
+def check_api_key():
+    """Issue #9: Validate API key from Authorization header."""
+    if not API_KEY:
+        return True  # No key configured = skip auth (dev mode)
+
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        return False
+    provided_key = auth_header[len("Bearer "):]
+    return provided_key == API_KEY
+
+
 @app.route("/tts", methods=["POST"])
 def text_to_speech():
     """Endpoint to convert full conversation to audio podcast"""
+    # Issue #9: API key authentication
+    if not check_api_key():
+        return jsonify({"error": "Unauthorized. Provide valid API key."}), 401
+
     data = request.get_json()
 
     if not data:
@@ -176,29 +212,32 @@ def text_to_speech():
     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
+    # Sanitize title to prevent path traversal
+    safe_title = secure_filename(title)
+    if not safe_title:
+        return jsonify({"error": "Invalid title format"}), 400
+
     (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
+    # Issue #12: Clean old outputs before generating new ones
-    wav_files = generate_wav_files(speakers, speaker_voice_map, title, tracks)
+    clean_old_outputs(safe_title)
+
+    wav_files = generate_wav_files(speakers, speaker_voice_map, safe_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"{safe_title}.wav")
-    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,

example_usage.py

@@ -17,7 +17,7 @@ Jamie: And I'm Jamie. Today we're diving into the latest AEW Dynamite recap from
 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"
+    url = "http://localhost:5012/tts"
 
     # Prepare the request payload
     payload = {"text": sample_text, "title": "aeW_dynamite_recap"}
@@ -43,7 +43,7 @@ Alex: First off, the article mentions Forbes' "Real‑Time" platform. That's bas
             print(response.text)
 
     except requests.exceptions.ConnectionError:
-        print("✗ Cannot connect to server. Make sure Flask app is running on port 5010")
+        print("✗ Cannot connect to server. Make sure Flask app is running on port 5012")
     except Exception as e:
         print(f"✗ Unexpected error: {e}")
 

requirements.txt

@@ -1,3 +1 @@
 flask
-werkzeug
-pydub

test.txt

@@ -1,46 +0,0 @@
-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

@@ -51,10 +51,8 @@ def test_kokoro_tts_integration():
     test_text = "This is a test of the kokoro text-to-speech system."
     test_title = "test_output"
 
-    # Pick a voice
+    # Pin a deterministic voice for reproducible testing
-    import random
+    voice = VOICES[0]
-
-    voice = random.choice(VOICES)
 
     print(f"Using voice: {voice}")
 

test_updated_app.py

@@ -12,10 +12,10 @@ def test_speaker_parsing():
     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)
+    # Issue #4: parse_speaker_text returns (speakers, tracks) tuple
+    speakers, tracks = parse_speaker_text(text)
 
     # Should have 2 speakers
     assert len(speakers) == 2, f"Expected 2 speakers, got {len(speakers)}"
@@ -28,7 +28,7 @@ def test_speaker_parsing():
     assert speakers["Alex"], "Alex should have content"
     assert speakers["Jamie"], "Jamie should have content"
 
-    print("✓ Speaker parsing test passed")
+    print("OK: Speaker parsing test passed")
 
 
 def test_voice_assignment():
@@ -38,7 +38,8 @@ def test_voice_assignment():
     with open("test.txt", "r") as f:
         text = f.read()
 
-    speakers = parse_speaker_text(text)
+    # Issue #14: Unpack tuple properly - parse_speaker_text returns (speakers, tracks)
+    speakers, tracks = parse_speaker_text(text)
     speaker_voice_map = assign_voices_to_speakers(speakers)
 
     # Should have assignments for both speakers
@@ -52,42 +53,38 @@ def test_voice_assignment():
             f"Invalid voice {voice} for {speaker}"
         )
 
-    print("✓ Voice assignment test passed")
+    print("OK: 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)
+    # Issue #4/#14: Unpack tuple properly
+    speakers, tracks = 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:
+            # Issue #5: Pass all 4 required args (speakers, speaker_voice_map, title, tracks)
             wav_files = generate_wav_files(
-                speakers, speaker_voice_map, "test_integration"
+                speakers, speaker_voice_map, "test_integration", tracks
             )
 
-            # Check that files were created
             if wav_files:
-                print("✓ Integration test passed - WAV files generated")
+                print("OK: Integration test passed - WAV files generated")
             else:
-                print("<EFBFBD><EFBFBD><EFBFBD> Integration test - No files generated (but no error)")
+                print("WARN: 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")
+    print("OK: Full integration test completed")
 
 
 if __name__ == "__main__":