a good start

.gitignore

@@ -3,3 +3,4 @@ __pycache__/
 */__pycache__/
 .DS_Store
 *.pyc
+.venv/

README.md

@@ -1,8 +1,32 @@
-# Run the reader app
-python apps/reader.py
+# Podcast Content Generator
 
-# Run the producer app
-python apps/producer.py
+A Gradio app that helps podcast producers generate preview clips, timestamps, descriptions, and more from podcast transcripts or YouTube videos.
 
-# Run a script
-python scripts/transcript.py audio_file.mp3
+## Features
+
+- Generate preview clips suggestions
+- Create Twitter/social media clips
+- Generate episode descriptions
+- Create timestamps
+- Get title and thumbnail suggestions
+- Support for YouTube URLs or raw transcript text
+- Customizable prompts for each type of content
+
+## Usage
+
+1. Paste a YouTube URL or transcript text into the input box
+2. Click "Generate Content" to process
+3. Get generated content in various formats
+4. Optionally customize the prompts used for generation
+
+## Environment Variables
+
+The app requires the following environment variable:
+- `ANTHROPIC_API_KEY`: Your Anthropic API key for Claude
+
+## Credits
+
+Built with:
+- Gradio
+- Claude AI (Anthropic)
+- YouTube Transcript API

app.py

@@ -0,0 +1,272 @@
+import gradio as gr
+import asyncio
+from pathlib import Path
+import anthropic
+import os
+from dataclasses import dataclass
+from typing import Dict
+from youtube_transcript_api import YouTubeTranscriptApi
+import re
+import pandas as pd
+
+# Move relevant classes and functions into app.py
+@dataclass
+class ContentRequest:
+    prompt_key: str
+
+class ContentGenerator:
+    def __init__(self):
+        self.current_prompts = self._load_default_prompts()
+        self.client = anthropic.Anthropic()
+        
+    def _load_default_prompts(self) -> Dict[str, str]:
+        """Load default prompts and examples from files and CSVs."""
+        
+        # Load CSV examples
+        try:
+            timestamps_df = pd.read_csv("data/Timestamps.csv")
+            titles_df = pd.read_csv("data/Titles & Thumbnails.csv")
+            descriptions_df = pd.read_csv("data/Viral Episode Descriptions.csv")
+            clips_df = pd.read_csv("data/Viral Twitter Clips.csv")
+            
+            # Format timestamp examples
+            timestamp_examples = "\n\n".join(timestamps_df['Timestamps'].dropna().tolist())
+            
+            # Format title examples
+            title_examples = "\n".join([
+                f'Title: "{row.Titles}"\nThumbnail: "{row.Thumbnail}"'
+                for _, row in titles_df.iterrows()
+            ])
+            
+            # Format description examples
+            description_examples = "\n".join([
+                f'Tweet: "{row["Tweet Text"]}"'
+                for _, row in descriptions_df.iterrows()
+            ])
+            
+            # Format clip examples
+            clip_examples = "\n\n".join([
+                f'Tweet Text: "{row["Tweet Text"]}"\nClip Transcript: "{row["Clip Transcript"]}"'
+                for _, row in clips_df.iterrows() if pd.notna(row["Tweet Text"])
+            ])
+            
+        except Exception as e:
+            print(f"Warning: Error loading CSV examples: {e}")
+            timestamp_examples = ""
+            title_examples = ""
+            description_examples = ""
+            clip_examples = ""
+
+        # Load base prompts and inject examples
+        prompts = {}
+        for key in ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"]:
+            prompt = Path(f"prompts/{key}.txt").read_text()
+            
+            # Inject relevant examples
+            if key == "timestamps":
+                prompt = prompt.replace("{timestamps_examples}", timestamp_examples)
+            elif key == "titles_and_thumbnails":
+                prompt = prompt.replace("{title_examples}", title_examples)
+            elif key == "description":
+                prompt = prompt.replace("{description_examples}", description_examples)
+            elif key == "clips":
+                prompt = prompt.replace("{clip_examples}", clip_examples)
+            
+            prompts[key] = prompt
+
+        return prompts
+
+    async def generate_content(self, request: ContentRequest, transcript: str) -> str:
+        """Generate content using Claude asynchronously."""
+        try:
+            print(f"\nFull prompt for {request.prompt_key}:")
+            print("=== SYSTEM PROMPT ===")
+            print(self.current_prompts[request.prompt_key])
+            print("=== END SYSTEM PROMPT ===\n")
+            
+            response = self.client.messages.create(
+                model="claude-3-5-sonnet-20241022",
+                max_tokens=8192,
+                system=self.current_prompts[request.prompt_key],
+                messages=[{"role": "user", "content": f"Process this transcript:\n\n{transcript}"}]
+            )
+            
+            if response and hasattr(response, 'content'):
+                return response.content[0].text
+            else:
+                return f"Error: Unexpected response structure for {request.prompt_key}"
+                
+        except Exception as e:
+            return f"Error generating content: {str(e)}"
+
+def extract_video_id(url: str) -> str:
+    """Extract video ID from various YouTube URL formats."""
+    match = re.search(
+        r"(?:youtube\.com\/watch\?v=|youtu\.be\/|youtube\.com\/embed\/|youtube\.com\/v\/)([A-Za-z0-9_-]+)",
+        url
+    )
+    return match.group(1) if match else None
+
+def get_transcript(video_id: str) -> str:
+    """Get transcript from YouTube video ID."""
+    try:
+        transcript = YouTubeTranscriptApi.list_transcripts(video_id).find_transcript(["en"])
+        return " ".join(entry["text"] for entry in transcript.fetch())
+    except Exception as e:
+        return f"Error fetching transcript: {str(e)}"
+
+class TranscriptProcessor:
+    def __init__(self):
+        self.generator = ContentGenerator()
+
+    def _get_youtube_transcript(self, url: str) -> str:
+        """Get transcript from YouTube URL."""
+        try:
+            if video_id := extract_video_id(url):
+                return get_transcript(video_id)
+            raise Exception("Invalid YouTube URL")
+        except Exception as e:
+            raise Exception(f"Error fetching YouTube transcript: {str(e)}")
+
+    async def process_transcript(self, input_text: str):
+        """Process input and generate all content."""
+        try:
+            transcript = (
+                self._get_youtube_transcript(input_text)
+                if any(x in input_text for x in ["youtube.com", "youtu.be"])
+                else input_text
+            )
+
+            # Process each type sequentially
+            sections = {}
+            for key in ["titles_and_thumbnails", "description", "previews", "clips", "timestamps"]:
+                result = await self.generator.generate_content(ContentRequest(key), transcript)
+                sections[key] = result
+
+            # Combine into markdown with H2 headers
+            markdown = f"""
+## Titles and Thumbnails
+
+{sections['titles_and_thumbnails']}
+
+## Twitter Description
+
+{sections['description']}
+
+## Preview Clips
+
+{sections['previews']}
+
+## Twitter Clips
+
+{sections['clips']}
+
+## Timestamps
+
+{sections['timestamps']}
+"""
+            return markdown
+
+        except Exception as e:
+            return f"Error processing input: {str(e)}"
+
+    def update_prompts(self, *values) -> str:
+        """Update the current session's prompts."""
+        self.generator.current_prompts.update(zip(
+            ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"],
+            values
+        ))
+        return "Prompts updated for this session!"
+
+def create_interface():
+    """Create the Gradio interface."""
+    processor = TranscriptProcessor()
+    
+    with gr.Blocks(title="Podcast Content Generator") as app:
+        gr.Markdown(
+            """
+            # Podcast Content Generator
+            Generate preview clips, timestamps, descriptions and more from podcast transcripts or YouTube videos.
+            
+            Simply paste a YouTube URL or raw transcript text to get started!
+            """
+        )
+        
+        with gr.Tab("Generate Content"):
+            input_text = gr.Textbox(
+                label="Input", 
+                placeholder="YouTube URL or transcript text...",
+                lines=10
+            )
+            submit_btn = gr.Button("Generate Content")
+            
+            output = gr.Markdown()  # Single markdown output
+
+            async def process_wrapper(text):
+                print("Process wrapper started")
+                print(f"Input text: {text[:100]}...")
+                
+                try:
+                    result = await processor.process_transcript(text)
+                    print("Process completed, got results")
+                    return result
+                except Exception as e:
+                    print(f"Error in process_wrapper: {str(e)}")
+                    return f"# Error\n\n{str(e)}"
+
+            submit_btn.click(
+                fn=process_wrapper,
+                inputs=input_text,
+                outputs=output,
+                queue=True
+            )
+
+        with gr.Tab("Customize Prompts"):
+            gr.Markdown(
+                """
+                ## Customize Generation Prompts
+                Here you can experiment with different prompts during your session.
+                Changes will remain active until you reload the page.
+                
+                Tip: Copy your preferred prompts somewhere safe if you want to reuse them later!
+                """
+            )
+
+            prompt_inputs = [
+                gr.Textbox(
+                    label=f"{key.replace('_', ' ').title()} Prompt",
+                    lines=10,
+                    value=processor.generator.current_prompts[key]
+                )
+                for key in [
+                    "previews",
+                    "clips", 
+                    "description",
+                    "timestamps",
+                    "titles_and_thumbnails"
+                ]
+            ]
+            status = gr.Textbox(label="Status", interactive=False)
+
+            # Update prompts when they change
+            for prompt in prompt_inputs:
+                prompt.change(
+                    fn=processor.update_prompts,
+                    inputs=prompt_inputs,
+                    outputs=[status]
+                )
+
+            # Reset button
+            reset_btn = gr.Button("Reset to Default Prompts")
+            reset_btn.click(
+                fn=lambda: (
+                    processor.update_prompts(*processor.generator.current_prompts.values()),
+                    *processor.generator.current_prompts.values(),
+                ),
+                outputs=[status] + prompt_inputs,
+            )
+
+    return app
+
+if __name__ == "__main__":
+    create_interface().launch() 

apps/producer.py

@@ -1,128 +0,0 @@
-import gradio as gr
-import asyncio
-from pathlib import Path
-from ..utils.content_generator import ContentGenerator, ContentRequest
-from ..utils.youtube_utils import get_transcript, extract_video_id
-
-class TranscriptProcessor:
-    def __init__(self):
-        self.generator = ContentGenerator()
-
-    def _get_youtube_transcript(self, url: str) -> str:
-        """Get transcript from YouTube URL."""
-        try:
-            if video_id := extract_video_id(url):
-                return get_transcript(video_id)
-            raise Exception("Invalid YouTube URL")
-        except Exception as e:
-            raise Exception(f"Error fetching YouTube transcript: {str(e)}")
-
-    async def process_transcript(self, input_text: str):
-        """Process input and generate all content."""
-        try:
-            # Get transcript from URL or use direct input
-            transcript = (
-                self._get_youtube_transcript(input_text)
-                if any(x in input_text for x in ["youtube.com", "youtu.be"])
-                else input_text
-            )
-
-            # Define content generation requests
-            requests = [
-                ContentRequest("previews", max_tokens=8192),
-                ContentRequest("clips", max_tokens=8192),
-                ContentRequest("description"),
-                ContentRequest("timestamps"),
-                ContentRequest("titles_and_thumbnails"),
-            ]
-
-            # Generate all content concurrently
-            results = await asyncio.gather(
-                *[self.generator.generate_content(req, transcript) for req in requests]
-            )
-            return tuple(results)
-
-        except Exception as e:
-            return (f"Error processing input: {str(e)}",) * 5
-
-    def update_prompts(self, *values) -> str:
-        """Update the current session's prompts."""
-        self.generator.current_prompts.update(zip(
-            ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"],
-            values
-        ))
-        return "Prompts updated for this session!"
-
-def create_interface():
-    """Create the Gradio interface."""
-    processor = TranscriptProcessor()
-    
-    with gr.Blocks(title="Podcast Transcript Analyzer") as app:
-        with gr.Tab("Generate Content"):
-            gr.Markdown("# Podcast Content Generator")
-            input_text = gr.Textbox(label="Input", placeholder="YouTube URL or transcript...", lines=10)
-            submit_btn = gr.Button("Generate Content")
-            outputs = [
-                gr.Textbox(label=label, lines=10, interactive=False)
-                for label in ["Preview Clips", "Twitter Clips", "Twitter Description", "Timestamps", "Title & Thumbnail Suggestions"]
-            ]
-            
-            async def process_wrapper(text):
-                return await processor.process_transcript(text)
-            
-            submit_btn.click(fn=process_wrapper, inputs=[input_text], outputs=outputs)
-
-        with gr.Tab("Experiment with Prompts"):
-            gr.Markdown("# Experiment with Prompts")
-            gr.Markdown(
-                """
-            Here you can experiment with different prompts during your session. 
-            Changes will remain active until you reload the page.
-            
-            Tip: Copy your preferred prompts somewhere safe if you want to reuse them later!
-            """
-            )
-
-            prompt_inputs = [
-                gr.Textbox(
-                    label="Preview Clips Prompt", lines=10, value=processor.generator.current_prompts["previews"]
-                ),
-                gr.Textbox(
-                    label="Clips Prompt", lines=10, value=processor.generator.current_prompts["clips"]
-                ),
-                gr.Textbox(
-                    label="Description Prompt",
-                    lines=10,
-                    value=processor.generator.current_prompts["description"],
-                ),
-                gr.Textbox(
-                    label="Timestamps Prompt",
-                    lines=10,
-                    value=processor.generator.current_prompts["timestamps"],
-                ),
-                gr.Textbox(
-                    label="Titles & Thumbnails Prompt",
-                    lines=10,
-                    value=processor.generator.current_prompts["titles_and_thumbnails"],
-                ),
-            ]
-            status = gr.Textbox(label="Status", interactive=False)
-
-            # Update prompts when they change
-            for prompt in prompt_inputs:
-                prompt.change(fn=processor.update_prompts, inputs=prompt_inputs, outputs=[status])
-
-            # Reset button
-            reset_btn = gr.Button("Reset to Default Prompts")
-            reset_btn.click(
-                fn=lambda: (
-                    processor.update_prompts(*processor.generator.current_prompts.values()),
-                    *processor.generator.current_prompts.values(),
-                ),
-                outputs=[status] + prompt_inputs,
-            )
-
-    return app
-
-if __name__ == "__main__":
-    create_interface().launch() 

apps/reader.py

@@ -1,364 +0,0 @@
-import sys
-from pathlib import Path
-
-# Add project root to Python path
-project_root = str(Path(__file__).parent.parent)
-if project_root not in sys.path:
-    sys.path.append(project_root)
-
-import gradio as gr
-import asyncio
-import os
-import json
-import requests
-from anthropic import Anthropic
-from utils.document_parser import DocumentParser
-from dotenv import load_dotenv
-
-# Load environment variables
-env_path = Path(project_root) / ".env"
-load_dotenv(env_path)
-
-# Mochi deck IDs
-DECK_CATEGORIES = {
-    "CS/Hardware": "rhGqR9SK",
-    "Math/Physics": "Dm5vczZg",
-    "AI": "SS9QEfiy",
-    "History/Military": "3nJYp7Zh",
-    "Quotes/Random": "rWUzSu8t",
-    "Bio": "BspzxaUJ",
-    "Econ/Finance": "mvvJ27Q1"
-}
-
-class CardGenerator:
-    """Handles card generation and Mochi integration."""
-    
-    def __init__(self):
-        self.parser = DocumentParser()
-        self.claude = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
-        self.mochi_key = os.getenv("MOCHI_API_KEY")
-        
-        # Load prompts
-        self.prompts = {
-            key: Path(f"prompts/{key}.txt").read_text()
-            for key in ["card_generation", "commentary"]
-        }
-        
-        # State
-        self.current_cards = []
-        self.current_index = 0
-        self.approved_cards = []
-    
-    def get_chapter_list(self, file_data) -> list[str]:
-        """Get list of chapters from document.
-        
-        Args:
-            file_data: File data from Gradio
-        """
-        try:
-            if not file_data:
-                return []
-            
-            # Attempt to extract filename from file_data
-            filename = getattr(file_data, 'name', None)
-            if not filename:
-                filename = "uploaded_file"
-                print("DEBUG: No filename attribute found, using default.")
-            else:
-                print(f"DEBUG: Filename extracted: {filename}")
-            
-            # Check file extension
-            file_ext = Path(filename).suffix.lower()
-            if not file_ext:
-                print("DEBUG: No file extension found, checking content type.")
-                # Attempt to determine file type from content
-                if file_data.startswith(b'%PDF-'):
-                    file_ext = '.pdf'
-                elif file_data.startswith(b'PK'):
-                    file_ext = '.epub'
-                else:
-                    raise ValueError("Unsupported file type")
-            print(f"DEBUG: File extension: {file_ext}")
-            
-            return self.parser.load_document(file_data, filename)
-        except Exception as e:
-            return [f"Error: {str(e)}"]
-    
-    async def process_chapter(self, file_data, chapter_idx: int) -> tuple:
-        """Process chapter and generate cards + commentary.
-        
-        Args:
-            file_data: File data from Gradio
-            chapter_idx: Index of chapter to process
-        """
-        try:
-            if not file_data:
-                return None, "No file provided"
-                
-            # Get chapter content
-            content = self.parser.get_chapter_content(chapter_idx)
-            
-            # Generate cards and commentary
-            cards, commentary = await asyncio.gather(
-                self._generate_cards(content),
-                self._generate_commentary(content)
-            )
-            
-            # Parse and store cards
-            self.current_cards = json.loads(cards)
-            self.current_index = 0
-            self.approved_cards = []
-            
-            # Return first card and commentary
-            return self._get_current_card(), commentary
-            
-        except Exception as e:
-            return None, f"Error: {str(e)}"
-        finally:
-            self.parser.cleanup()
-    
-    async def _generate_cards(self, content: str) -> str:
-        """Generate flashcards using Claude."""
-        response = await self.claude.messages.create(
-            model="claude-3-opus-20240229",
-            max_tokens=4000,
-            system=self.prompts["card_generation"],
-            messages=[{"role": "user", "content": content}]
-        )
-        return response.content[0].text
-    
-    async def _generate_commentary(self, content: str) -> str:
-        """Generate commentary using Claude."""
-        response = await self.claude.messages.create(
-            model="claude-3-opus-20240229",
-            max_tokens=4000,
-            system=self.prompts["commentary"],
-            messages=[{"role": "user", "content": content}]
-        )
-        return response.content[0].text
-    
-    def _get_current_card(self) -> dict:
-        """Get current card with UI state."""
-        if not self.current_cards or self.current_index >= len(self.current_cards):
-            return {
-                'front': "",
-                'back': "",
-                'category': "",
-                'status': "No more cards to review",
-                'show_buttons': False,
-                'show_upload': True
-            }
-        
-        card = self.current_cards[self.current_index]
-        return {
-            'front': card['front'],
-            'back': card['back'],
-            'category': card['category'],
-            'status': f"Card {self.current_index + 1} of {len(self.current_cards)}",
-            'show_buttons': True,
-            'show_upload': False
-        }
-    
-    def accept_card(self, front: str, back: str, category: str) -> dict:
-        """Accept current card and move to next."""
-        if self.current_index < len(self.current_cards):
-            self.approved_cards.append({
-                'front': front,
-                'back': back,
-                'category': category
-            })
-        
-        self.current_index += 1
-        return self._get_current_card()
-    
-    def reject_card(self) -> dict:
-        """Reject current card and move to next."""
-        if self.current_index < len(self.current_cards):
-            self.current_cards.pop(self.current_index)
-        return self._get_current_card()
-    
-    def upload_to_mochi(self) -> str:
-        """Upload approved cards to Mochi."""
-        if not self.approved_cards:
-            return "No cards to upload!"
-        
-        results = []
-        for card in self.approved_cards:
-            try:
-                # Format card for Mochi
-                mochi_card = {
-                    "deck-id": DECK_CATEGORIES[card["category"]],
-                    "fields": {
-                        "name": {"id": "name", "value": card["front"]},
-                        "back": {"id": "back", "value": card["back"]}
-                    }
-                }
-                
-                # Upload to Mochi
-                response = requests.post(
-                    "https://app.mochi.cards/api/cards",
-                    json=mochi_card,
-                    auth=(self.mochi_key, "")
-                )
-                
-                if response.status_code != 200:
-                    results.append(f"Error: {response.text}")
-                
-            except Exception as e:
-                results.append(f"Error: {str(e)}")
-        
-        # Clear approved cards
-        success_count = len(self.approved_cards) - len(results)
-        self.approved_cards = []
-        
-        if results:
-            return f"Uploaded {success_count} cards with {len(results)} errors:\n" + "\n".join(results)
-        return f"Successfully uploaded {success_count} cards to Mochi!"
-
-def create_interface():
-    """Create the Gradio interface."""
-    generator = CardGenerator()
-    
-    with gr.Blocks(title="Document Reader & Card Generator") as app:
-        # Document upload and chapter selection
-        with gr.Row():
-            file_input = gr.File(
-                label="Upload EPUB Document",
-                type="binary",
-                file_types=[".epub"]
-            )
-        
-        chapter_select = gr.Dropdown(
-            label="Select Chapter",
-            choices=[],
-            interactive=True,
-            visible=False
-        )
-        
-        def update_chapters(file):
-            if not file:
-                return gr.update(choices=[], visible=False)
-            chapters = generator.get_chapter_list(file)
-            return gr.update(choices=chapters, visible=True, value=chapters[0] if chapters else None)
-        
-        file_input.change(
-            fn=update_chapters,
-            inputs=[file_input],
-            outputs=[chapter_select]
-        )
-        
-        process_btn = gr.Button("Process Chapter")
-        
-        # Commentary section
-        commentary = gr.Textbox(
-            label="Commentary",
-            lines=10,
-            interactive=False
-        )
-        
-        # Card review section
-        gr.Markdown("## Review Cards")
-        
-        with gr.Row():
-            card_front = gr.Textbox(
-                label="Front",
-                lines=3,
-                interactive=True
-            )
-            card_back = gr.Textbox(
-                label="Back",
-                lines=3,
-                interactive=True
-            )
-        
-        with gr.Row():
-            deck_category = gr.Dropdown(
-                choices=list(DECK_CATEGORIES.keys()),
-                label="Deck Category",
-                value="AI"
-            )
-            card_status = gr.Textbox(
-                label="Status",
-                interactive=False
-            )
-        
-        with gr.Row():
-            accept_btn = gr.Button("Accept & Next", visible=False)
-            reject_btn = gr.Button("Reject & Next", visible=False)
-            upload_btn = gr.Button("Upload to Mochi", visible=False)
-        
-        upload_status = gr.Textbox(
-            label="Upload Status",
-            interactive=False
-        )
-        
-        # Event handlers
-        async def process_chapter(file, chapter_idx):
-            card, comment = await generator.process_chapter(file, chapter_idx)
-            if not card:  # Error occurred
-                return [
-                    "", "", comment, gr.update(visible=False),
-                    gr.update(visible=False), "", gr.update(visible=False)
-                ]
-            
-            return [
-                card['front'],
-                card['back'],
-                comment,
-                gr.update(visible=card['show_buttons']),
-                gr.update(visible=card['show_buttons']),
-                card['status'],
-                gr.update(visible=card['show_upload'])
-            ]
-        
-        def handle_card_action(action, front, back, category):
-            card = (generator.accept_card(front, back, category) 
-                   if action == 'accept' else 
-                   generator.reject_card())
-            
-            return [
-                card['front'],
-                card['back'],
-                card['status'],
-                gr.update(visible=card['show_buttons']),
-                gr.update(visible=card['show_buttons']),
-                card['category'],
-                gr.update(visible=card['show_upload'])
-            ]
-        
-        # Connect events
-        process_btn.click(
-            fn=process_chapter,
-            inputs=[file_input, chapter_select],
-            outputs=[
-                card_front, card_back, commentary,
-                accept_btn, reject_btn, card_status, upload_btn
-            ]
-        )
-        
-        accept_btn.click(
-            fn=lambda f, b, c: handle_card_action('accept', f, b, c),
-            inputs=[card_front, card_back, deck_category],
-            outputs=[
-                card_front, card_back, card_status,
-                accept_btn, reject_btn, deck_category, upload_btn
-            ]
-        )
-        
-        reject_btn.click(
-            fn=lambda: handle_card_action('reject', None, None, None),
-            outputs=[
-                card_front, card_back, card_status,
-                accept_btn, reject_btn, deck_category, upload_btn
-            ]
-        )
-        
-        upload_btn.click(
-            fn=generator.upload_to_mochi,
-            outputs=[upload_status]
-        )
-    
-    return app
-
-if __name__ == "__main__":
-    create_interface().launch() 

prompts/card_generation.txt

@@ -1,38 +0,0 @@
-You are an expert at creating high-quality spaced repetition flashcards that promote deep understanding and retention. Your task is to generate flashcards from the given text that are:
-
-1. Clear and concise
-2. Focus on one concept per card
-3. Test understanding rather than just recall
-4. Avoid overly complex or compound questions
-5. Use precise language
-
-Each card must be assigned to one of these categories:
-- CS/Hardware
-- Math/Physics
-- AI
-- History/Military
-- Quotes/Random
-- Bio
-- Econ/Finance
-
-Format each card as a JSON object:
-{
-    "category": "Category name from the list above",
-    "front": "Question or prompt",
-    "back": "Answer or explanation"
-}
-
-Example cards:
-{
-    "category": "Bio",
-    "front": "What is the key difference between procedural and declarative memory?",
-    "back": "Procedural memory is for skills and procedures (how to ride a bike), while declarative memory is for facts and events (what you had for breakfast)."
-}
-
-{
-    "category": "Bio",
-    "front": "What role does the hippocampus play in memory formation?",
-    "back": "The hippocampus is crucial for converting short-term memories into long-term memories through a process called consolidation. It acts as a temporary storage and processing center before memories are distributed to other parts of the cortex."
-}
-
-Please generate 5-10 high-quality flashcards from the provided text. Focus on the most important concepts, insights, and relationships. Format the output as a JSON array containing the card objects. 

prompts/clips.txt

@@ -1,5 +1,9 @@
 You are a social media expert for the Dwarkesh Podcast. Generate 10 viral-worthy clips from the transcript.
-Format as:
+
+Here are examples of successful viral clips from previous episodes:
+{clip_examples}
+
+Format your output as:
 Tweet 1
 Tweet Text: [text]
 Clip Transcript: [45-120 seconds of transcript] 

prompts/commentary.txt

@@ -1,25 +0,0 @@
-You are an expert researcher and critical thinker. Your task is to analyze the provided text and generate insightful commentary that:
-
-1. Identifies the key arguments, insights, and novel ideas
-2. Highlights connections to other important concepts or fields
-3. Points out particularly interesting or counterintuitive points
-4. Suggests areas that merit further exploration
-5. Notes any potential weaknesses or areas of uncertainty in the arguments
-
-Your commentary should be scholarly but engaging, helping the reader develop a deeper understanding of the material. Focus on substance over style, and be specific rather than general.
-
-Structure your response as follows:
-
-Key Insights:
-- [2-3 bullet points highlighting the most important takeaways]
-
-Interesting Connections:
-- [2-3 bullet points noting connections to other fields/concepts]
-
-Worth Exploring Further:
-- [1-2 bullet points suggesting related areas for deeper investigation]
-
-Critical Notes:
-- [1-2 bullet points on potential weaknesses or areas needing clarification]
-
-Then provide 2-3 paragraphs of integrated analysis that weaves these points together into a coherent commentary. 

prompts/description.txt

@@ -1,4 +1,7 @@
 Create an engaging episode description tweet (280 chars max) that:
 1. Highlights compelling aspects
 2. Includes topic areas and handles
-3. Ends with "Links below" or "Enjoy!" 
+3. Ends with "Links below" or "Enjoy!" 
+
+Here are examples of successful episode descriptions:
+{description_examples} 

prompts/enhance.txt

@@ -1,46 +0,0 @@
-You are an expert transcript editor. Your task is to enhance this transcript for maximum readability while maintaining the core message.
-
-IMPORTANT: Respond ONLY with the enhanced transcript. Do not include any explanations, headers, or phrases like "Here is the transcript."
-
-Note: Below you'll find an auto-generated transcript that may help with speaker identification, but focus on creating your own high-quality transcript from the audio.
-
-Think about your job as if you were transcribing an interview for a print book where the priority is the reading audience. It should just be a total pleasure to read this as a written artifact where all the flubs and repetitions and conversational artifacts and filler words and false starts are removed, where a bunch of helpful punctuation is added. It should basically read like somebody wrote it specifically for reading rather than just something somebody said extemporaneously. 
-
-Please:
-1. Fix speaker attribution errors, especially at segment boundaries. Watch for incomplete thoughts that were likely from the previous speaker.
-
-2. Optimize AGGRESSIVELY for readability over verbatim accuracy:
-   - Readability is the most important thing!!
-   - Remove ALL conversational artifacts (yeah, so, I mean, etc.)
-   - Remove ALL filler words (um, uh, like, you know)
-   - Remove false starts and self-corrections completely
-   - Remove redundant phrases and hesitations
-   - Convert any indirect or rambling responses into direct statements
-   - Break up run-on sentences into clear, concise statements
-   - Maintain natural conversation flow while prioritizing clarity and directness
-
-3. Format the output consistently:
-   - Keep the "Speaker X 00:00:00" format (no brackets, no other formatting)
-   - DO NOT change the timestamps. You're only seeing a chunk of the full transcript, which is why your 0:00:00 is not the true beginning. Keep the timestamps as they are.
-   - Add TWO line breaks between speaker/timestamp and the text
-   - Use proper punctuation and capitalization
-   - Add paragraph breaks for topic changes
-   - When you add paragraph breaks between the same speaker's remarks, no need to restate the speaker attribution
-   - Don't go more than four sentences without adding a paragraph break. Be liberal with your paragraph breaks. 
-   - Preserve distinct speaker turns
-
-Example input:
-Speaker A 00:01:15
-
-Um, yeah, so like, I've been working on this new project at work, you know? And uh, what's really interesting is that, uh, we're seeing these amazing results with the new approach we're taking. Like, it's just, you know, it's really transforming how we do things.
-
-And then, I mean, the thing is, uh, when we showed it to the client last week, they were just, you know, completely blown away by what we achieved. Like, they couldn't even believe it was the same system they had before.
-
-Example output:
-Speaker A 00:01:15
-
-I've been working on this new project at work, and we're seeing amazing results with our new approach. It's really transforming how we do things.
-
-When we showed it to the client last week, they were completely blown away by what we achieved. They couldn't believe it was the same system they had before.
-
-Enhance the following transcript, starting directly with the speaker format: 

prompts/find_links.txt

@@ -1,35 +0,0 @@
-You are an expert at identifying key terms, concepts, and references in text that would benefit from having reference links. Your task is to analyze the provided transcript text and identify terms that would genuinely help readers understand important context they might miss otherwise.
-
-Focus ONLY on these types of terms:
-1. Technical concepts and jargon that a general audience might not be familiar with
-2. Research papers or academic works mentioned or referenced
-3. Blog posts, articles, or online resources that are specifically cited
-4. Books that are discussed (to link to Goodreads/Amazon)
-5. Specific projects, tools, or technologies that are central to the discussion
-6. Names of lesser-known people who made significant contributions being discussed
-
-DO NOT identify:
-1. Common words or general concepts (like "short", "editor", "polymath")
-2. Basic technical terms that most people would know
-3. Generic job titles or roles
-4. Common industry terms
-5. Basic scientific concepts
-6. Well-known companies or organizations
-
-Remember: Only identify terms where having a reference would genuinely add value by providing important context or deeper understanding that the audience might otherwise miss.
-
-Respond in this format for each term:
-TERM: <the exact term as it appears in text>
-REASON: <1-2 sentences explaining why this term should be linked>
-
-Example input:
-"We used GPT-4 to implement the RLHF technique described in the Constitutional AI paper, similar to what Anthropic did with Claude."
-
-Example output:
-TERM: RLHF
-REASON: A complex technical concept (Reinforcement Learning from Human Feedback) that's crucial to understanding modern AI development but might be unfamiliar to general audiences.
-
-TERM: Constitutional AI paper
-REASON: A specific research paper that introduced important concepts being referenced; readers might want to read the original source.
-
-Analyze the following transcript text and identify key terms that should be linked: 

prompts/timestamps.txt

@@ -7,5 +7,5 @@ You are a podcast timestamp generator. Create 5-7 timestamps for this episode, f
 
 Output the timestamps in chronological order, one per line. 
 
-Previous examples:
+Here are examples from previous episodes:
 {timestamps_examples} 

prompts/titles_and_thumbnails.txt

@@ -13,6 +13,9 @@ Thumbnail: 2-4 ALL CAPS words that amplify the intrigue
 - Create intellectual curiosity without sensationalism
 - Make the viewer wonder "What's the story here?"
 
+Here are examples of successful title-thumbnail combinations from previous episodes:
+{title_examples}
+
 Example:
 Title: "David Reich – How One Small Tribe Conquered the World 70,000 Years Ago"
 Thumbnail: "LAST HUMANS STANDING"

requirements.txt

@@ -1,13 +1,4 @@
-gradio
-deepgram-sdk
-google-generativeai
-anthropic
-pandas
+gradio>=4.0.0
 youtube-transcript-api
-pydub 
-assemblyai
-pytube
-PyPDF2
-EbookLib
-beautifulsoup4
+anthropic
 python-dotenv

scripts/add_links.py

@@ -1,209 +0,0 @@
-import argparse
-from pathlib import Path
-import os
-import re
-from typing import List, Dict, Tuple
-from dataclasses import dataclass
-import anthropic
-from exa_py import Exa
-
-@dataclass
-class Term:
-    """A term identified for linking with its explanation"""
-    term: str
-    reason: str
-
-@dataclass
-class Link:
-    """A link found for a term"""
-    term: str
-    url: str
-    title: str
-
-def chunk_text(text: str, max_chunk_size: int = 2000) -> List[str]:
-    """Split text into chunks of roughly equal size at paragraph boundaries"""
-    paragraphs = text.split("\n\n")
-    chunks = []
-    current_chunk = []
-    current_size = 0
-    
-    for para in paragraphs:
-        para_size = len(para)
-        if current_size + para_size > max_chunk_size and current_chunk:
-            chunks.append("\n\n".join(current_chunk))
-            current_chunk = [para]
-            current_size = para_size
-        else:
-            current_chunk.append(para)
-            current_size += para_size
-    
-    if current_chunk:
-        chunks.append("\n\n".join(current_chunk))
-    
-    return chunks
-
-def parse_claude_response(response: str) -> List[Term]:
-    """Parse Claude's response to extract terms and reasons"""
-    terms = []
-    current_term = None
-    current_reason = None
-    
-    for line in response.split("\n"):
-        line = line.strip()
-        if not line:
-            continue
-            
-        if line.startswith("TERM: "):
-            # Save previous term if exists
-            if current_term and current_reason:
-                terms.append(Term(current_term, current_reason))
-            current_term = line[6:].strip()
-            current_reason = None
-        elif line.startswith("REASON: "):
-            current_reason = line[8:].strip()
-            
-    # Add final term
-    if current_term and current_reason:
-        terms.append(Term(current_term, current_reason))
-        
-    return terms
-
-def find_links_for_terms(exa: Exa, terms: List[Term]) -> Dict[str, Link]:
-    """Find best link for each term using Exa search"""
-    links = {}
-    
-    for term in terms:
-        # Construct a search query that looks for authoritative sources
-        # query = f"The best explanation or overview of {term.term} is (site: wikipedia.org OR site: .edu OR site: .gov):"
-        
-        try:
-            # Search with Exa
-            results = exa.search(term.term, num_results=1, type="auto")
-            if results.results:
-                result = results.results[0]
-                links[term.term] = Link(
-                    term=term.term,
-                    url=result.url,
-                    title=result.title
-                )
-        except Exception as e:
-            print(f"Error finding link for {term.term}: {e}")
-            continue
-            
-    return links
-
-def add_links_to_text(text: str, links: Dict[str, Link]) -> str:
-    """Add markdown links to text for all terms we have links for"""
-    # Sort terms by length (descending) to handle overlapping terms correctly
-    terms = sorted(links.keys(), key=len, reverse=True)
-    
-    # Create regex pattern that matches whole words only
-    patterns = [re.compile(fr'\b{re.escape(term)}\b') for term in terms]
-    
-    # Track which terms we've linked to avoid duplicate links
-    linked_terms = set()
-    
-    # Process each term
-    result = text
-    for term, pattern in zip(terms, patterns):
-        if term in linked_terms:
-            continue
-            
-        # Only replace first occurrence
-        link = links[term]
-        replacement = f"[{term}]({link.url})"
-        result = pattern.sub(replacement, result, count=1)
-        linked_terms.add(term)
-        
-    return result
-
-def process_transcript(
-    transcript_path: Path,
-    claude_client: anthropic.Client,
-    exa_client: Exa,
-    prompt_template: str
-) -> str:
-    """Process a transcript file to add reference links"""
-    # Read transcript
-    text = transcript_path.read_text()
-    
-    # Split into chunks
-    chunks = chunk_text(text)
-    
-    # Process each chunk
-    all_terms = []
-    for chunk in chunks:
-        # Get Claude's suggestions
-        prompt = prompt_template + "\n\n" + chunk
-        response = claude_client.messages.create(
-            model="claude-3-5-sonnet-20241022",
-            max_tokens=1024,
-            system="You are a helpful AI assistant.",
-            messages=[{"role": "user", "content": prompt}]
-        )
-        
-        # Parse response
-        terms = parse_claude_response(response.content[0].text)
-        all_terms.extend(terms)
-        
-    # Find links for all terms
-    links = find_links_for_terms(exa_client, all_terms)
-    
-    # Add links to text
-    linked_text = add_links_to_text(text, links)
-    
-    return linked_text
-
-def main():
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "transcript", 
-        nargs="?",  # Make the argument optional
-        default="output/transcripts/transcript.md",
-        help="Path to transcript file (default: output/transcripts/transcript.md)"
-    )
-    parser.add_argument("--output", help="Output file path (default: input path with -linked suffix)")
-    args = parser.parse_args()
-    
-    transcript_path = Path(args.transcript)
-    if not transcript_path.exists():
-        raise FileNotFoundError(f"Transcript file not found: {transcript_path}")
-        
-    # Set up output path
-    if args.output:
-        output_path = Path(args.output)
-    else:
-        stem = transcript_path.stem
-        output_path = transcript_path.parent / f"{stem}-linked{transcript_path.suffix}"
-        
-    # Read prompt template
-    prompt_path = Path("prompts/find_links.txt")
-    if not prompt_path.exists():
-        raise FileNotFoundError(f"Prompt file not found: {prompt_path}")
-    prompt_template = prompt_path.read_text()
-    
-    # Initialize clients
-    claude_client = anthropic.Client(api_key=os.getenv("ANTHROPIC_API_KEY"))
-    exa_client = Exa(api_key=os.getenv("EXA_API_KEY"))
-    
-    try:
-        # Process transcript
-        linked_text = process_transcript(
-            transcript_path,
-            claude_client,
-            exa_client,
-            prompt_template
-        )
-        
-        # Save output
-        output_path.write_text(linked_text)
-        print(f"Processed transcript saved to: {output_path}")
-        
-    except Exception as e:
-        print(f"Error processing transcript: {e}")
-        return 1
-        
-    return 0
-
-if __name__ == "__main__":
-    main() 

scripts/preview_generator.py

@@ -1,92 +0,0 @@
-import argparse
-from pathlib import Path
-import os
-from google import generativeai
-from pydub import AudioSegment
-
-
-class PreviewGenerator:
-    """Handles generating preview suggestions using Gemini"""
-
-    def __init__(self, api_key: str):
-        generativeai.configure(api_key=api_key)
-        self.model = generativeai.GenerativeModel("gemini-exp-1206")
-        self.prompt = Path("prompts/previews.txt").read_text()
-
-    async def generate_previews(self, audio_path: Path, transcript_path: Path = None) -> str:
-        """Generate preview suggestions for the given audio file and optional transcript"""
-        print("Generating preview suggestions...")
-        
-        # Load and compress audio for Gemini
-        audio = AudioSegment.from_file(audio_path)
-        
-        # Create a buffer for the compressed audio
-        import io
-        buffer = io.BytesIO()
-        # Use lower quality MP3 for faster processing
-        audio.export(buffer, format="mp3", parameters=["-q:a", "9"])
-        buffer.seek(0)
-        
-        # Use the File API to upload the audio
-        audio_file = generativeai.upload_file(buffer, mime_type="audio/mp3")
-        
-        # Prepare content for Gemini
-        content = [self.prompt]
-        content.append(audio_file)  # Add the uploaded file reference
-        
-        # Add transcript if provided
-        if transcript_path and transcript_path.exists():
-            print("Including transcript in analysis...")
-            # Upload transcript as a file too
-            transcript_file = generativeai.upload_file(transcript_path)
-            content.append(transcript_file)
-        
-        # Generate suggestions using Gemini
-        response = await self.model.generate_content_async(content)
-        
-        return response.text
-
-
-async def main():
-    parser = argparse.ArgumentParser(description="Generate podcast preview suggestions")
-    parser.add_argument("audio_file", help="Audio file to analyze")
-    parser.add_argument("--transcript", "-t", help="Optional transcript file")
-    args = parser.parse_args()
-    
-    audio_path = Path(args.audio_file)
-    if not audio_path.exists():
-        raise FileNotFoundError(f"File not found: {audio_path}")
-    
-    transcript_path = Path(args.transcript) if args.transcript else None
-    if transcript_path and not transcript_path.exists():
-        print(f"Warning: Transcript file not found: {transcript_path}")
-        transcript_path = None
-    
-    # Ensure output directory exists
-    output_dir = Path("output")
-    output_dir.mkdir(exist_ok=True)
-    output_path = output_dir / "previews.txt"
-    
-    try:
-        generator = PreviewGenerator(os.getenv("GOOGLE_API_KEY"))
-        suggestions = await generator.generate_previews(audio_path, transcript_path)
-        
-        # Save output
-        output_path.write_text(suggestions)
-        print(f"\nPreview suggestions saved to: {output_path}")
-        
-        # Also print to console
-        print("\nPreview Suggestions:")
-        print("-" * 40)
-        print(suggestions)
-            
-    except Exception as e:
-        print(f"Error: {e}")
-        return 1
-    
-    return 0
-
-
-if __name__ == "__main__":
-    import asyncio
-    asyncio.run(main())

scripts/process_playlist.py

@@ -1,77 +0,0 @@
-import asyncio
-from pathlib import Path
-import sys
-import time
-from typing import List
-
-# Add the project root to Python path
-sys.path.append(str(Path(__file__).parent.parent))
-
-from utils.youtube_utils import get_transcript, get_playlist_video_ids
-from utils.content_generator import ContentGenerator, ContentRequest
-
-PLAYLIST_URL = "https://www.youtube.com/playlist?list=PLd7-bHaQwnthaNDpZ32TtYONGVk95-fhF"
-MAX_CONCURRENT = 3  # Limit concurrent requests
-RETRY_DELAY = 65  # Seconds to wait before retrying after rate limit
-
-async def process_video(video_id: str, generator: ContentGenerator, retry_count: int = 0) -> str:
-    """Process a single video and return the formatted result."""
-    try:
-        print(f"Processing video {video_id}...")
-        
-        # Get transcript
-        transcript = get_transcript(video_id)
-        if not transcript:
-            print(f"No transcript available for {video_id}")
-            return ""
-            
-        # Generate suggestions
-        request = ContentRequest("titles_and_thumbnails")
-        result = await generator.generate_content(request, transcript)
-        return f"Video ID: {video_id}\n{result}\n{'='*50}\n"
-        
-    except Exception as e:
-        if "rate_limit_error" in str(e) and retry_count < 3:
-            print(f"Rate limit hit for {video_id}, waiting {RETRY_DELAY}s before retry {retry_count + 1}")
-            await asyncio.sleep(RETRY_DELAY)
-            return await process_video(video_id, generator, retry_count + 1)
-        print(f"Error processing {video_id}: {e}")
-        return ""
-
-async def process_batch(video_ids: List[str], generator: ContentGenerator) -> List[str]:
-    """Process a batch of videos with rate limiting."""
-    tasks = [process_video(video_id, generator) for video_id in video_ids]
-    return await asyncio.gather(*tasks)
-
-async def process_playlist():
-    """Process all videos in playlist with batching."""
-    generator = ContentGenerator()
-    output_file = Path("output/playlist-titles-thumbnails.txt")
-    
-    # Get videos from playlist
-    print("Getting videos from playlist...")
-    video_ids = get_playlist_video_ids(PLAYLIST_URL)
-    print(f"Found {len(video_ids)} videos")
-    
-    # Process videos in batches
-    results = []
-    for i in range(0, len(video_ids), MAX_CONCURRENT):
-        batch = video_ids[i:i + MAX_CONCURRENT]
-        print(f"\nProcessing batch {i//MAX_CONCURRENT + 1}")
-        batch_results = await process_batch(batch, generator)
-        results.extend(batch_results)
-        
-        # Add delay between batches to avoid rate limits
-        if i + MAX_CONCURRENT < len(video_ids):
-            delay = 5  # Short delay between successful batches
-            print(f"Waiting {delay}s before next batch...")
-            await asyncio.sleep(delay)
-    
-    # Filter out empty results and save
-    results = [r for r in results if r]
-    output_file.parent.mkdir(parents=True, exist_ok=True)
-    output_file.write_text("\n".join(results))
-    print(f"\nResults written to {output_file}")
-
-if __name__ == "__main__":
-    asyncio.run(process_playlist())

scripts/transcript.py

@@ -1,311 +0,0 @@
-import argparse
-from dataclasses import dataclass
-from pathlib import Path
-import json
-import hashlib
-import os
-from typing import List, Tuple, Iterator
-import assemblyai as aai
-from google import generativeai
-from pydub import AudioSegment
-import asyncio
-import io
-from multiprocessing import Pool
-from functools import partial
-from itertools import groupby
-
-
-@dataclass
-class Utterance:
-    """A single utterance from a speaker"""
-    speaker: str
-    text: str
-    start: int  # timestamp in ms from AssemblyAI
-    end: int    # timestamp in ms from AssemblyAI
-
-    @property
-    def timestamp(self) -> str:
-        """Format start time as HH:MM:SS"""
-        seconds = int(self.start // 1000)
-        hours = seconds // 3600
-        minutes = (seconds % 3600) // 60
-        seconds = seconds % 60
-        return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
-
-
-class Transcriber:
-    """Handles getting and caching transcripts from AssemblyAI"""
-
-    def __init__(self, api_key: str):
-        aai.settings.api_key = api_key
-        self.cache_dir = Path("output/transcripts/.cache")
-        self.cache_dir.mkdir(parents=True, exist_ok=True)
-
-    def get_transcript(self, audio_path: Path) -> List[Utterance]:
-        """Get transcript, using cache if available"""
-        cache_file = self.cache_dir / f"{audio_path.stem}.json"
-        
-        if cache_file.exists():
-            with open(cache_file) as f:
-                data = json.load(f)
-                if data["hash"] == self._get_file_hash(audio_path):
-                    print("Using cached AssemblyAI transcript...")
-                    # Create proper Utterance objects from cached data
-                    return [
-                        Utterance(
-                            speaker=u["speaker"],
-                            text=u["text"],
-                            start=u["start"],
-                            end=u["end"]
-                        )
-                        for u in data["utterances"]
-                    ]
-
-        print("Getting new transcript from AssemblyAI...")
-        config = aai.TranscriptionConfig(speaker_labels=True, language_code="en")
-        transcript = aai.Transcriber().transcribe(str(audio_path), config=config)
-        
-        utterances = [
-            Utterance(
-                speaker=u.speaker,
-                text=u.text,
-                start=u.start,
-                end=u.end
-            )
-            for u in transcript.utterances
-        ]
-        
-        # Cache the raw utterance data
-        cache_data = {
-            "hash": self._get_file_hash(audio_path),
-            "utterances": [
-                {
-                    "speaker": u.speaker,
-                    "text": u.text,
-                    "start": u.start,
-                    "end": u.end
-                }
-                for u in utterances
-            ]
-        }
-        with open(cache_file, "w") as f:
-            json.dump(cache_data, f, indent=2)
-            
-        return utterances
-
-    def _get_file_hash(self, file_path: Path) -> str:
-        """Calculate MD5 hash of a file"""
-        hash_md5 = hashlib.md5()
-        with open(file_path, "rb") as f:
-            for chunk in iter(lambda: f.read(4096), b""):
-                hash_md5.update(chunk)
-        return hash_md5.hexdigest()
-
-
-class Enhancer:
-    """Handles enhancing transcripts using Gemini"""
-
-    def __init__(self, api_key: str):
-        generativeai.configure(api_key=api_key)
-        self.model = generativeai.GenerativeModel("gemini-exp-1206")
-        self.prompt = Path("prompts/enhance.txt").read_text()
-
-    async def enhance_chunks(self, chunks: List[Tuple[str, io.BytesIO]]) -> List[str]:
-        """Enhance multiple transcript chunks concurrently with concurrency control"""
-        print(f"Enhancing {len(chunks)} chunks...")
-        
-        # Create a semaphore to limit concurrent requests
-        semaphore = asyncio.Semaphore(3)  # Allow up to 3 concurrent requests
-        
-        async def process_chunk(i: int, chunk: Tuple[str, io.BytesIO]) -> str:
-            text, audio = chunk
-            async with semaphore:
-                audio.seek(0)
-                response = await self.model.generate_content_async(
-                    [self.prompt, text, {"mime_type": "audio/mp3", "data": audio.read()}]
-                )
-                print(f"Completed chunk {i+1}/{len(chunks)}")
-                return response.text
-
-        # Create tasks for all chunks and run them concurrently
-        tasks = [
-            process_chunk(i, chunk) 
-            for i, chunk in enumerate(chunks)
-        ]
-        
-        # Wait for all tasks to complete
-        results = await asyncio.gather(*tasks)
-        return results
-
-
-@dataclass
-class SpeakerDialogue:
-    """Represents a continuous section of speech from a single speaker"""
-    speaker: str
-    utterances: List[Utterance]
-    
-    @property
-    def start(self) -> int:
-        """Start time of first utterance"""
-        return self.utterances[0].start
-    
-    @property
-    def end(self) -> int:
-        """End time of last utterance"""
-        return self.utterances[-1].end
-    
-    @property
-    def timestamp(self) -> str:
-        """Format start time as HH:MM:SS"""
-        return self.utterances[0].timestamp
-    
-    def format(self, markdown: bool = False) -> str:
-        """Format this dialogue as text with newlines between utterances
-        Args:
-            markdown: If True, add markdown formatting for speaker and timestamp
-        """
-        texts = [u.text + "\n\n" for u in self.utterances]  # Add two newlines after each utterance
-        combined_text = ''.join(texts).rstrip()  # Remove trailing whitespace at the end
-        if markdown:
-            return f"**Speaker {self.speaker}** *{self.timestamp}*\n\n{combined_text}"
-        return f"Speaker {self.speaker} {self.timestamp}\n\n{combined_text}"
-
-
-def group_utterances_by_speaker(utterances: List[Utterance]) -> Iterator[SpeakerDialogue]:
-    """Group consecutive utterances by the same speaker"""
-    for speaker, group in groupby(utterances, key=lambda u: u.speaker):
-        yield SpeakerDialogue(speaker=speaker, utterances=list(group))
-
-
-def estimate_tokens(text: str, chars_per_token: int = 4) -> int:
-    """
-    Estimate number of tokens in text
-    Args:
-        text: The text to estimate tokens for
-        chars_per_token: Estimated characters per token (default 4)
-    """
-    return (len(text) + chars_per_token - 1) // chars_per_token
-
-
-def chunk_dialogues(
-    dialogues: Iterator[SpeakerDialogue], 
-    max_tokens: int = 2000, 
-    chars_per_token: int = 4
-) -> List[List[SpeakerDialogue]]:
-    """
-    Split dialogues into chunks that fit within token limit
-    Args:
-        dialogues: Iterator of SpeakerDialogues
-        max_tokens: Maximum tokens per chunk
-        chars_per_token: Estimated characters per token (default 4)
-    """
-    chunks = []
-    current_chunk = []
-    current_text = ""
-    
-    for dialogue in dialogues:
-        # Format this dialogue
-        formatted = dialogue.format()
-        
-        # If adding this dialogue would exceed token limit, start new chunk
-        new_text = current_text + "\n\n" + formatted if current_text else formatted
-        if current_chunk and estimate_tokens(new_text, chars_per_token) > max_tokens:
-            chunks.append(current_chunk)
-            current_chunk = [dialogue]
-            current_text = formatted
-        else:
-            current_chunk.append(dialogue)
-            current_text = new_text
-    
-    if current_chunk:
-        chunks.append(current_chunk)
-    
-    return chunks
-
-
-def format_chunk(dialogues: List[SpeakerDialogue], markdown: bool = False) -> str:
-    """Format a chunk of dialogues into readable text
-    Args:
-        dialogues: List of dialogues to format
-        markdown: If True, add markdown formatting for speaker and timestamp
-    """
-    return "\n\n".join(dialogue.format(markdown=markdown) for dialogue in dialogues)
-
-
-def prepare_audio_chunks(audio_path: Path, utterances: List[Utterance]) -> List[Tuple[str, io.BytesIO]]:
-    """Prepare audio chunks and their corresponding text"""
-    # Group utterances by speaker and split into chunks
-    dialogues = group_utterances_by_speaker(utterances)
-    chunks = chunk_dialogues(dialogues)
-    print(f"Preparing {len(chunks)} audio segments...")
-    
-    # Load audio once
-    audio = AudioSegment.from_file(audio_path)
-    
-    # Process each chunk
-    prepared = []
-    for chunk in chunks:
-        # Extract just the needed segment
-        segment = audio[chunk[0].start:chunk[-1].end]
-        buffer = io.BytesIO()
-        # Use lower quality MP3 for faster processing
-        segment.export(buffer, format="mp3", parameters=["-q:a", "9"])
-        # Use non-markdown format for Gemini
-        prepared.append((format_chunk(chunk, markdown=False), buffer))
-    
-    return prepared
-
-
-def main():
-    parser = argparse.ArgumentParser()
-    parser.add_argument("audio_file", help="Audio file to transcribe")
-    args = parser.parse_args()
-    
-    audio_path = Path(args.audio_file)
-    if not audio_path.exists():
-        raise FileNotFoundError(f"File not found: {audio_path}")
-        
-    out_dir = Path("output/transcripts")
-    out_dir.mkdir(parents=True, exist_ok=True)
-    
-    try:
-        # Get transcript
-        transcriber = Transcriber(os.getenv("ASSEMBLYAI_API_KEY"))
-        utterances = transcriber.get_transcript(audio_path)
-        
-        # Save original transcript
-        dialogues = list(group_utterances_by_speaker(utterances))  # Convert iterator to list
-        original = format_chunk(dialogues, markdown=True)  # Use markdown for final output
-        (out_dir / "autogenerated-transcript.md").write_text(original)
-        
-        # Enhance transcript
-        enhancer = Enhancer(os.getenv("GOOGLE_API_KEY"))
-        chunks = prepare_audio_chunks(audio_path, utterances)
-        enhanced = asyncio.run(enhancer.enhance_chunks(chunks))
-        
-        # Save enhanced transcript with markdown
-        merged = "\n\n".join(chunk.strip() for chunk in enhanced)
-        # Apply markdown formatting to the final enhanced transcript
-        merged = apply_markdown_formatting(merged)
-        (out_dir / "transcript.md").write_text(merged)
-        
-        print("\nTranscripts saved to:")
-        print(f"- {out_dir}/autogenerated-transcript.md")
-        print(f"- {out_dir}/transcript.md")
-        
-    except Exception as e:
-        print(f"Error: {e}")
-        return 1
-    
-    return 0
-
-
-def apply_markdown_formatting(text: str) -> str:
-    """Apply markdown formatting to speaker and timestamp in the transcript"""
-    import re
-    pattern = r"(Speaker \w+) (\d{2}:\d{2}:\d{2})"
-    return re.sub(pattern, r"**\1** *\2*", text)
-
-
-if __name__ == "__main__":
-    main()

utils/__init__.py

@@ -1 +0,0 @@
-# Empty file to make utils a package 

utils/content_generator.py

@@ -1,79 +0,0 @@
-import anthropic
-from dataclasses import dataclass
-from pathlib import Path
-import asyncio
-import concurrent.futures
-import time
-from typing import Dict, List
-import pandas as pd
-
-client = anthropic.Anthropic()
-
-@dataclass
-class ContentRequest:
-    prompt_key: str
-    max_tokens: int = 2000
-    temperature: float = 1.0
-
-class ContentGenerator:
-    def __init__(self):
-        self.current_prompts = self._load_default_prompts()
-        
-    def _load_default_prompts(self) -> Dict[str, str]:
-        """Load default prompts from files."""
-        return {
-            key: Path(f"prompts/{key}.txt").read_text()
-            for key in ["previews", "clips", "description", "timestamps", "titles_and_thumbnails"]
-        }
-
-    def _load_examples(self, filename: str, columns: List[str]) -> str:
-        """Load examples from CSV file."""
-        try:
-            df = pd.read_csv(f"data/{filename}")
-            if len(columns) == 1:
-                return "\n\n".join(df[columns[0]].dropna().tolist())
-            
-            examples = []
-            for _, row in df.iterrows():
-                if all(pd.notna(row[col]) for col in columns):
-                    example = "\n".join(f"{col}: {row[col]}" for col in columns)
-                    examples.append(example)
-            return "\n\n".join(examples)
-        except Exception as e:
-            print(f"Error loading {filename}: {str(e)}")
-            return ""
-
-    async def generate_content(self, request: ContentRequest, transcript: str) -> str:
-        """Generate content using Claude asynchronously."""
-        print(f"Starting {request.prompt_key} generation...")
-        start_time = time.time()
-        
-        example_configs = {
-            "clips": ("Viral Twitter Clips.csv", ["Tweet Text", "Clip Transcript"]),
-            "description": ("Viral Episode Descriptions.csv", ["Tweet Text"]),
-            "timestamps": ("Timestamps.csv", ["Timestamps"]),
-            "titles_and_thumbnails": ("Titles & Thumbnails.csv", ["Titles", "Thumbnail"]),
-        }
-        
-        # Build prompt with examples
-        full_prompt = self.current_prompts[request.prompt_key]
-        if config := example_configs.get(request.prompt_key):
-            if examples := self._load_examples(*config):
-                full_prompt += f"\n\nPrevious examples:\n{examples}"
-
-        # Run API call in thread pool
-        loop = asyncio.get_event_loop()
-        with concurrent.futures.ThreadPoolExecutor() as pool:
-            message = await loop.run_in_executor(
-                pool,
-                lambda: client.messages.create(
-                    model="claude-3-5-sonnet-20241022",
-                    max_tokens=request.max_tokens,
-                    temperature=request.temperature,
-                    system=full_prompt,
-                    messages=[{"role": "user", "content": [{"type": "text", "text": f"Process this transcript:\n\n{transcript}"}]}]
-                )
-            )
-        result = message.content[0].text
-        print(f"Finished {request.prompt_key} in {time.time() - start_time:.2f} seconds")
-        return result 

utils/document_parser.py

@@ -1,191 +0,0 @@
-from pathlib import Path
-import tempfile
-import os
-from ebooklib import epub
-from bs4 import BeautifulSoup
-
-class DocumentParser:
-    """Simple EPUB document parser that extracts chapters and their content."""
-    
-    def __init__(self):
-        self._temp_file = None
-        self._book = None
-        self._chapters = []
-    
-    def load_document(self, file_data, filename=None) -> list[str]:
-        """Load an EPUB document and extract chapter titles.
-        
-        Args:
-            file_data: File data from Gradio (FileData object with read() method)
-            filename: Optional filename (not used)
-        """
-        # Clean up any previous temp file
-        self.cleanup()
-        
-        # Get the raw bytes from the Gradio file data
-        content = file_data.read() if hasattr(file_data, 'read') else file_data
-        
-        # Save to temp file
-        with tempfile.NamedTemporaryFile(delete=False, suffix='.epub') as temp:
-            temp.write(content)
-            self._temp_file = temp.name
-        
-        # Read the EPUB
-        try:
-            self._book = epub.read_epub(self._temp_file)
-            print("DEBUG: Successfully read EPUB file")
-        except Exception as e:
-            print(f"DEBUG: Error reading EPUB: {str(e)}")
-            raise ValueError(f"Failed to read EPUB: {str(e)}")
-        
-        # Extract chapters
-        self._chapters = self._extract_chapters()
-        print(f"DEBUG: Extracted {len(self._chapters)} chapters")
-        
-        # Return chapter titles
-        return [chapter['title'] for chapter in self._chapters]
-    
-    def get_chapter_content(self, chapter_idx: int) -> str:
-        """Get the content of a specific chapter."""
-        if not self._book or not self._chapters:
-            raise ValueError("No document loaded")
-        
-        if not 0 <= chapter_idx < len(self._chapters):
-            raise ValueError(f"Invalid chapter index: {chapter_idx}")
-        
-        chapter = self._chapters[chapter_idx]
-        self._current_chapter_title = chapter['title'].strip()  # Store for _get_chapter_text
-        
-        print(f"DEBUG: Getting content for chapter: {self._current_chapter_title}")
-        content = self._get_chapter_text(chapter['item'])
-        print(f"DEBUG: Extracted {len(content)} characters of content")
-        
-        return content
-    
-    def _extract_chapters(self) -> list[dict]:
-        """Extract chapters from the EPUB file."""
-        chapters = []
-        
-        # First try to get chapters from the table of contents
-        print("DEBUG: Checking table of contents...")
-        if hasattr(self._book, 'toc'):
-            # Debug the TOC structure
-            print("DEBUG: TOC structure:")
-            for item in self._book.toc:
-                print(f"DEBUG: TOC item type: {type(item)}")
-                if isinstance(item, tuple):
-                    print(f"DEBUG: Tuple length: {len(item)}")
-                    if len(item) > 1:
-                        print(f"DEBUG: Second item type: {type(item[1])}")
-                        if isinstance(item[1], (list, tuple)):
-                            print(f"DEBUG: Sub-items count: {len(item[1])}")
-            
-            def process_toc_entries(entries, level=0):
-                for item in entries:
-                    # Handle both Link objects and tuples
-                    if hasattr(item, 'title') and hasattr(item, 'href'):
-                        # Direct Link object
-                        doc = self._book.get_item_with_href(item.href)
-                        if doc:
-                            prefix = "  " * level if level > 0 else ""
-                            chapters.append({
-                                'title': prefix + item.title,
-                                'item': doc
-                            })
-                    elif isinstance(item, tuple):
-                        section = item[0]
-                        # Process the section
-                        if hasattr(section, 'title') and hasattr(section, 'href'):
-                            doc = self._book.get_item_with_href(section.href)
-                            if doc:
-                                prefix = "  " * level if level > 0 else ""
-                                chapters.append({
-                                    'title': prefix + section.title,
-                                    'item': doc
-                                })
-                        
-                        # Process sub-items if they exist
-                        if len(item) > 1:
-                            if isinstance(item[1], (list, tuple)):
-                                process_toc_entries(item[1], level + 1)
-                            elif hasattr(item[1], 'title'):  # Single sub-item
-                                process_toc_entries([item[1]], level + 1)
-            
-            process_toc_entries(self._book.toc)
-            print(f"DEBUG: Found {len(chapters)} chapters in TOC")
-            print("DEBUG: Chapter titles found:")
-            for ch in chapters:
-                print(f"  - {ch['title']}")
-        
-        # If no chapters found in TOC, scan the documents
-        if not chapters:
-            print("DEBUG: No chapters in TOC, scanning documents...")
-            # Get all HTML documents
-            docs = [item for item in self._book.get_items() 
-                   if item.get_type() == epub.ITEM_DOCUMENT]
-            
-            print(f"DEBUG: Found {len(docs)} documents to scan")
-            
-            for doc in docs:
-                soup = BeautifulSoup(doc.get_content(), 'html.parser')
-                
-                # Look for chapter headings
-                headings = (
-                    soup.find_all(['h1', 'h2']) + 
-                    soup.find_all(class_=lambda x: x and ('chapter' in x.lower() or 'title' in x.lower()))
-                )
-                
-                for heading in headings:
-                    # Clean up the text
-                    title = ' '.join(heading.get_text().split())
-                    if title:  # Only add if we have a title
-                        chapters.append({
-                            'title': title,
-                            'item': doc
-                        })
-        
-        if not chapters:
-            print("DEBUG: No chapters found, using documents as chapters")
-            # If still no chapters found, treat each document as a chapter
-            for doc in self._book.get_items():
-                if doc.get_type() == epub.ITEM_DOCUMENT:
-                    chapters.append({
-                        'title': f"Chapter {len(chapters) + 1}",
-                        'item': doc
-                    })
-        
-        return chapters
-    
-    def _get_chapter_text(self, item) -> str:
-        """Extract text content from a chapter."""
-        try:
-            soup = BeautifulSoup(item.get_content(), 'html.parser')
-            
-            # Remove script and style elements
-            for element in soup(['script', 'style']):
-                element.decompose()
-            
-            # Get main content area (usually in body or main tags)
-            content_area = soup.find('body') or soup.find('main') or soup
-            
-            # Get all text blocks, excluding navigation elements
-            text_blocks = []
-            for element in content_area.find_all(text=True, recursive=True):
-                if (element.parent.name not in ['script', 'style', 'nav', 'header'] and
-                    element.strip()):
-                    text_blocks.append(element.strip())
-            
-            return '\n\n'.join(text_blocks)
-            
-        except Exception as e:
-            print(f"DEBUG: Error extracting text: {str(e)}")
-            # Fallback to simple text extraction
-            return soup.get_text(separator='\n\n', strip=True)
-    
-    def cleanup(self):
-        """Clean up temporary files."""
-        if self._temp_file and os.path.exists(self._temp_file):
-            os.unlink(self._temp_file)
-            self._temp_file = None
-        self._book = None
-        self._chapters = [] 

utils/youtube_utils.py

@@ -1,26 +0,0 @@
-from youtube_transcript_api import YouTubeTranscriptApi
-from pytube import Playlist
-import re
-from typing import Optional, List
-
-def extract_video_id(url: str) -> Optional[str]:
-    """Extract video ID from various YouTube URL formats."""
-    match = re.search(
-        r"(?:youtube\.com\/watch\?v=|youtu\.be\/|youtube\.com\/embed\/|youtube\.com\/v\/)([A-Za-z0-9_-]+)",
-        url
-    )
-    return match.group(1) if match else None
-
-def get_transcript(video_id: str) -> str:
-    """Get transcript from YouTube video ID."""
-    try:
-        transcript = YouTubeTranscriptApi.list_transcripts(video_id).find_transcript(["en"])
-        return " ".join(entry["text"] for entry in transcript.fetch())
-    except Exception as e:
-        print(f"Error fetching transcript for {video_id}: {str(e)}")
-        return ""
-
-def get_playlist_video_ids(playlist_url: str) -> List[str]:
-    """Get all video IDs from a YouTube playlist."""
-    playlist = Playlist(playlist_url)
-    return [url.split("watch?v=")[1] for url in playlist.video_urls]