import re import json import os from dataclasses import dataclass, asdict from typing import List, Dict, Any, Optional, Tuple from collections import Counter import math from pathlib import Path @dataclass class SlideContent: """Structure for individual slide content""" slide_number: int title: str main_points: List[str] supporting_details: List[str] key_quotes: List[str] timestamp_range: Tuple[float, float] # (start, end) in seconds slide_type: str # 'intro', 'content', 'conclusion', 'summary' @dataclass class AnalyzedContent: """Complete analyzed content ready for slide generation""" video_id: str video_title: str main_topic: str key_themes: List[str] slides: List[SlideContent] total_slides: int estimated_presentation_time: float # minutes content_summary: str analysis_metadata: Dict[str, Any] success: bool error_message: Optional[str] = None @dataclass class SavedTranscript: """Structure to match your saved transcript format""" video_id: str transcript: str duration: float language: str word_count: int success: bool video_title: str = "" extraction_method: str = "" extraction_timestamp: str = "" error_message: Optional[str] = None class YouTubeContentAnalyzer: """ Analyzes extracted content and structures it for slide generation """ def __init__(self, target_slides: Optional[int] = None, min_slides: int = 3, max_slides: int = 15): self.target_slides = target_slides self.min_slides = min_slides self.max_slides = max_slides # Common stop words to filter out self.stop_words = { 'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'from', 'up', 'about', 'into', 'through', 'during', 'before', 'after', 'above', 'below', 'between', 'among', 'throughout', 'despite', 'towards', 'upon', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should', 'may', 'might', 'must', 'can', 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'me', 'him', 'her', 'us', 'them', 'my', 'your', 'his', 'her', 'its', 'our', 'their', 'this', 'that', 'these', 'those' } def clean_text(self, text: str) -> str: """Clean and normalize text""" if not text: return "" # Remove music notation and special characters text = re.sub(r'[ā™Ŗā™«šŸŽµšŸŽ¶]', '', text) text = re.sub(r'\[.*?\]', '', text) # Remove bracketed content text = re.sub(r'\(.*?\)', '', text) # Remove parenthetical content # Remove timestamp markers like <00:00:00.359> text = re.sub(r'<\d+:\d+:\d+\.\d+>', '', text) # Remove "Kind: captions Language: en" type metadata text = re.sub(r'Kind:\s*\w+\s*Language:\s*\w+', '', text, flags=re.IGNORECASE) # Remove standalone numbers and timestamps text = re.sub(r'\b\d+\b', ' ', text) # Clean up whitespace text = re.sub(r'\s+', ' ', text) text = text.strip() return text def extract_sentences(self, text: str) -> List[str]: """Extract sentences from text""" if not text: return [] # First clean the text clean_text = self.clean_text(text) # Split on sentence endings sentences = re.split(r'[.!?]+', clean_text) # Clean and filter sentences clean_sentences = [] for sentence in sentences: sentence = sentence.strip() # More lenient filtering - accept shorter sentences if len(sentence) > 10 and not sentence.lower().startswith(('kind:', 'language:')): clean_sentences.append(sentence) print(f"Debug: Extracted {len(clean_sentences)} sentences from {len(text)} characters") if clean_sentences: print(f"Debug: First sentence: {clean_sentences[0][:100]}...") return clean_sentences def extract_key_phrases(self, text: str, top_n: int = 10) -> List[str]: """Extract key phrases and topics from text""" # Clean text clean_text = self.clean_text(text.lower()) # Extract words words = re.findall(r'\b[a-zA-Z]{3,}\b', clean_text) # Filter stop words filtered_words = [word for word in words if word not in self.stop_words] # Count word frequency word_freq = Counter(filtered_words) # Get most common words key_words = [word for word, count in word_freq.most_common(top_n * 2)] # Extract 2-3 word phrases phrases = [] words_list = clean_text.split() for i in range(len(words_list) - 1): # 2-word phrases phrase = f"{words_list[i]} {words_list[i+1]}" if (len(phrase) > 6 and words_list[i] not in self.stop_words and words_list[i+1] not in self.stop_words): phrases.append(phrase) # 3-word phrases if i < len(words_list) - 2: phrase = f"{words_list[i]} {words_list[i+1]} {words_list[i+2]}" if (len(phrase) > 10 and words_list[i] not in self.stop_words and words_list[i+2] not in self.stop_words): phrases.append(phrase) # Count phrase frequency phrase_freq = Counter(phrases) top_phrases = [phrase for phrase, count in phrase_freq.most_common(top_n)] # Combine words and phrases key_phrases = list(set(key_words[:top_n//2] + top_phrases[:top_n//2])) return key_phrases[:top_n] def identify_main_topic(self, text: str, video_title: str = "") -> str: """Identify the main topic of the content""" # Clean video title for hints if video_title: title_words = re.findall(r'\b[a-zA-Z]{3,}\b', video_title.lower()) title_keywords = [word for word in title_words if word not in self.stop_words] else: title_keywords = [] # Extract key phrases from content key_phrases = self.extract_key_phrases(text, top_n=5) # Combine title and content insights if title_keywords: main_topic = f"{' '.join(title_keywords[:3])}" if key_phrases: main_topic += f" - {key_phrases[0]}" elif key_phrases: main_topic = key_phrases[0].title() else: main_topic = "Content Analysis" return main_topic.title() def extract_key_themes(self, text: str, num_themes: int = 5) -> List[str]: """Extract main themes from the content""" sentences = self.extract_sentences(text) # Group sentences by similar content themes = [] # Extract themes based on key phrases key_phrases = self.extract_key_phrases(text, top_n=num_themes * 2) for phrase in key_phrases[:num_themes]: # Find sentences containing this phrase related_sentences = [] for sentence in sentences: if phrase.lower() in sentence.lower(): related_sentences.append(sentence[:100] + "..." if len(sentence) > 100 else sentence) if related_sentences: theme_title = phrase.title() themes.append(theme_title) # Ensure we have at least some themes if not themes: themes = ["Introduction", "Main Content", "Key Points", "Conclusion"] return themes[:num_themes] def determine_slide_count(self, content_length: int, duration: float) -> int: """Determine optimal number of slides based on content""" if self.target_slides: return max(self.min_slides, min(self.max_slides, self.target_slides)) # Calculate based on content density # Rule of thumb: 1 slide per 30-60 seconds, adjusted for word count time_based = max(3, min(15, duration // 45)) # 45 seconds per slide word_based = max(3, min(15, content_length // 400)) # 400 words per slide for longer content # Take average and round optimal_slides = int((time_based + word_based) / 2) return max(self.min_slides, min(self.max_slides, optimal_slides)) def segment_content(self, sentences: List[str], num_slides: int) -> List[List[str]]: """Segment content into slide-sized chunks""" if not sentences: return [] # Calculate sentences per slide sentences_per_slide = max(1, len(sentences) // num_slides) segments = [] current_segment = [] for i, sentence in enumerate(sentences): current_segment.append(sentence) # Create new segment when we reach target size or at natural breaks if (len(current_segment) >= sentences_per_slide and i < len(sentences) - 1) or i == len(sentences) - 1: if current_segment: segments.append(current_segment.copy()) current_segment = [] # Ensure we don't exceed target slide count while len(segments) > num_slides and len(segments) > 1: # Merge last two segments if len(segments) >= 2: segments[-2].extend(segments[-1]) segments.pop() return segments def create_slide_title(self, content_segment: List[str], slide_number: int, total_slides: int) -> str: """Generate title for a slide based on its content""" if not content_segment: return f"Slide {slide_number}" # Special cases for first and last slides if slide_number == 1: return "Introduction" elif slide_number == total_slides: return "Conclusion" # Extract key concepts from the segment combined_text = ' '.join(content_segment[:2]) # Use first 2 sentences key_phrases = self.extract_key_phrases(combined_text, top_n=3) if key_phrases: title = key_phrases[0].title() # Ensure title isn't too long if len(title) > 50: title = title[:47] + "..." return title return f"Key Point {slide_number - 1}" def extract_main_points(self, content_segment: List[str], max_points: int = 4) -> List[str]: """Extract main points from a content segment""" points = [] for sentence in content_segment[:max_points]: # Clean and format sentence as bullet point point = sentence.strip() # Remove redundant phrases point = re.sub(r'^(and|but|so|then|also|furthermore|moreover|however)\s+', '', point, flags=re.IGNORECASE) # Ensure point starts with capital letter if point: point = point[0].upper() + point[1:] if len(point) > 1 else point.upper() # Ensure point ends with period if not point.endswith(('.', '!', '?')): point += '.' points.append(point) return points def extract_supporting_details(self, content_segment: List[str], main_points: List[str]) -> List[str]: """Extract supporting details that complement main points""" details = [] used_sentences = set(main_points) for sentence in content_segment: if sentence not in used_sentences and len(details) < 3: # Look for sentences with specific details, numbers, examples if (re.search(r'\d+', sentence) or # Contains numbers 'example' in sentence.lower() or 'such as' in sentence.lower() or 'including' in sentence.lower() or 'like' in sentence.lower()): detail = sentence.strip() if len(detail) > 20: # Ensure it's substantial details.append(detail) return details def extract_key_quotes(self, content_segment: List[str], max_quotes: int = 2) -> List[str]: """Extract memorable quotes or key statements""" quotes = [] for sentence in content_segment: # Look for quotable content (direct speech, strong statements) if (len(sentence) > 30 and len(sentence) < 150 and # Good length for quotes ('said' in sentence.lower() or sentence.count('"') >= 2 or re.search(r'[!.]', sentence))): # Emphatic or complete statements quote = sentence.strip() quotes.append(quote) if len(quotes) >= max_quotes: break return quotes def analyze_content(self, extracted_content) -> AnalyzedContent: """Main analysis function that processes extracted content into slide structure""" try: # Extract basic info video_id = extracted_content.video_id transcript = extracted_content.transcript duration = extracted_content.duration word_count = extracted_content.word_count video_title = getattr(extracted_content, 'video_title', f"Video Analysis - {video_id}") print(f"Debug: Starting analysis for {video_id}") print(f"Debug: Transcript length: {len(transcript)} characters") print(f"Debug: Video title: {video_title}") print(f"Debug: Transcript preview: {transcript[:200]}...") # Clean and prepare text clean_transcript = self.clean_text(transcript) print(f"Debug: Clean transcript length: {len(clean_transcript)} characters") sentences = self.extract_sentences(clean_transcript) if not sentences: print(f"Debug: No sentences extracted. Raw transcript sample: {transcript[:500]}") # Try alternative sentence extraction alt_sentences = transcript.split('.') alt_sentences = [s.strip() for s in alt_sentences if len(s.strip()) > 10] if alt_sentences: print(f"Debug: Alternative extraction found {len(alt_sentences)} sentences") sentences = alt_sentences[:20] # Use first 20 sentences else: return AnalyzedContent( video_id=video_id, video_title=video_title, main_topic="", key_themes=[], slides=[], total_slides=0, estimated_presentation_time=0, content_summary="", analysis_metadata={}, success=False, error_message=f"No content to analyze. Transcript length: {len(transcript)}, Clean length: {len(clean_transcript)}" ) print(f"Debug: Processing {len(sentences)} sentences") # Analyze content structure main_topic = self.identify_main_topic(clean_transcript, video_title) key_themes = self.extract_key_themes(clean_transcript) # Determine slide structure num_slides = self.determine_slide_count(word_count, duration) content_segments = self.segment_content(sentences, num_slides) print(f"Debug: Created {len(content_segments)} content segments for {num_slides} slides") # Create slides slides = [] for i, segment in enumerate(content_segments): slide_number = i + 1 # Determine slide type if slide_number == 1: slide_type = "intro" elif slide_number == len(content_segments): slide_type = "conclusion" else: slide_type = "content" # Generate slide content title = self.create_slide_title(segment, slide_number, len(content_segments)) main_points = self.extract_main_points(segment) supporting_details = self.extract_supporting_details(segment, main_points) key_quotes = self.extract_key_quotes(segment) # Calculate timestamp range (rough estimate) segment_duration = duration / len(content_segments) start_time = (slide_number - 1) * segment_duration end_time = slide_number * segment_duration slide = SlideContent( slide_number=slide_number, title=title, main_points=main_points, supporting_details=supporting_details, key_quotes=key_quotes, timestamp_range=(start_time, end_time), slide_type=slide_type ) slides.append(slide) # Generate content summary content_summary = f"Analysis of {len(sentences)} sentences across {len(slides)} slides, focusing on {main_topic.lower()}." # Calculate presentation time (assuming 30 seconds per slide + 1 minute per quote/detail) base_time = len(slides) * 0.5 # 30 seconds per slide detail_time = sum(len(slide.supporting_details) + len(slide.key_quotes) for slide in slides) * 0.25 estimated_time = base_time + detail_time # Analysis metadata analysis_metadata = { 'original_word_count': word_count, 'sentences_analyzed': len(sentences), 'slides_created': len(slides), 'avg_points_per_slide': sum(len(slide.main_points) for slide in slides) / len(slides) if slides else 0, 'content_density': word_count / len(slides) if slides else 0, 'analysis_timestamp': __import__('datetime').datetime.now().isoformat() } return AnalyzedContent( video_id=video_id, video_title=video_title, main_topic=main_topic, key_themes=key_themes, slides=slides, total_slides=len(slides), estimated_presentation_time=estimated_time, content_summary=content_summary, analysis_metadata=analysis_metadata, success=True ) except Exception as e: import traceback print(f"Debug: Exception occurred: {str(e)}") print(f"Debug: Traceback: {traceback.format_exc()}") return AnalyzedContent( video_id=getattr(extracted_content, 'video_id', 'unknown'), video_title="Error", main_topic="", key_themes=[], slides=[], total_slides=0, estimated_presentation_time=0, content_summary="", analysis_metadata={}, success=False, error_message=f"Analysis failed: {str(e)}" ) class TranscriptAnalysisManager: """Manages analysis of saved transcripts""" def __init__(self, transcripts_dir: str = "transcripts", analysis_dir: str = "analysis"): self.transcripts_dir = Path(transcripts_dir) self.analysis_dir = Path(analysis_dir) self.analysis_dir.mkdir(exist_ok=True) # Initialize analyzer self.analyzer = YouTubeContentAnalyzer() def load_transcript(self, video_id: str) -> Optional[SavedTranscript]: """Load a specific saved transcript""" try: transcript_file = self.transcripts_dir / f"{video_id}_transcript.json" if not transcript_file.exists(): print(f"āŒ Transcript file not found: {transcript_file}") return None with open(transcript_file, 'r', encoding='utf-8') as f: data = json.load(f) # Convert to SavedTranscript object transcript = SavedTranscript(**data) print(f"āœ… Loaded transcript for {video_id}: {transcript.video_title}") return transcript except Exception as e: print(f"āŒ Error loading transcript for {video_id}: {e}") return None def list_available_transcripts(self) -> List[Dict[str, Any]]: """List all available saved transcripts""" transcripts = [] for file_path in self.transcripts_dir.glob("*_transcript.json"): try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) transcripts.append({ 'video_id': data.get('video_id', 'unknown'), 'title': data.get('video_title', 'Unknown'), 'word_count': data.get('word_count', 0), 'duration': data.get('duration', 0), 'success': data.get('success', False), 'extraction_method': data.get('extraction_method', 'unknown'), 'file_path': str(file_path) }) except Exception as e: print(f"Error reading {file_path}: {e}") return sorted(transcripts, key=lambda x: x.get('word_count', 0), reverse=True) def analyze_transcript(self, video_id: str, target_slides: Optional[int] = None, force_refresh: bool = False) -> Optional[AnalyzedContent]: """Analyze a specific saved transcript""" # Check if analysis already exists analysis_file = self.analysis_dir / f"{video_id}_analysis.json" if not force_refresh and analysis_file.exists(): try: print(f"šŸ“ Loading existing analysis for {video_id}") return self.load_analysis(video_id) except: print(f"āš ļø Failed to load existing analysis, creating new one") # Load transcript transcript = self.load_transcript(video_id) if not transcript or not transcript.success: print(f"āŒ Cannot analyze: transcript not available or failed") return None print(f"šŸ”„ Analyzing transcript for {video_id}...") # Run analysis analyzer = YouTubeContentAnalyzer(target_slides=target_slides) analyzed_content = analyzer.analyze_content(transcript) if analyzed_content.success: # Save analysis self.save_analysis(analyzed_content) print(f"āœ… Analysis completed for {video_id}") return analyzed_content else: print(f"āŒ Analysis failed: {analyzed_content.error_message}") return None def save_analysis(self, analyzed_content: AnalyzedContent) -> bool: """Save analysis results to JSON""" try: analysis_file = self.analysis_dir / f"{analyzed_content.video_id}_analysis.json" # Convert dataclass to dict for JSON serialization analysis_dict = asdict(analyzed_content) with open(analysis_file, 'w', encoding='utf-8') as f: json.dump(analysis_dict, f, indent=2, ensure_ascii=False) print(f"šŸ’¾ Analysis saved to: {analysis_file}") return True except Exception as e: print(f"āŒ Failed to save analysis: {e}") return False def load_analysis(self, video_id: str) -> Optional[AnalyzedContent]: """Load previously saved analysis""" try: analysis_file = self.analysis_dir / f"{video_id}_analysis.json" if not analysis_file.exists(): return None with open(analysis_file, 'r', encoding='utf-8') as f: data = json.load(f) # Reconstruct AnalyzedContent object slides = [] for slide_data in data.get('slides', []): slide = SlideContent( slide_number=slide_data['slide_number'], title=slide_data['title'], main_points=slide_data['main_points'], supporting_details=slide_data['supporting_details'], key_quotes=slide_data['key_quotes'], timestamp_range=tuple(slide_data['timestamp_range']), slide_type=slide_data['slide_type'] ) slides.append(slide) analyzed_content = AnalyzedContent( video_id=data['video_id'], video_title=data['video_title'], main_topic=data['main_topic'], key_themes=data['key_themes'], slides=slides, total_slides=data['total_slides'], estimated_presentation_time=data['estimated_presentation_time'], content_summary=data['content_summary'], analysis_metadata=data['analysis_metadata'], success=data['success'], error_message=data.get('error_message') ) return analyzed_content except Exception as e: print(f"āŒ Error loading analysis: {e}") return None def get_analysis_summary(self, video_id: str) -> Optional[str]: """Get a formatted summary of analysis results""" analyzed = self.load_analysis(video_id) if not analyzed or not analyzed.success: return None summary = f""" šŸ“Š **ANALYSIS SUMMARY: {analyzed.video_title}** šŸŽÆ **Overview:** ā€¢ Main Topic: {analyzed.main_topic} ā€¢ Total Slides: {analyzed.total_slides} ā€¢ Key Themes: {', '.join(analyzed.key_themes[:3])} ā€¢ Estimated Time: {analyzed.estimated_presentation_time:.1f} minutes šŸ“‹ **Slide Structure:**""" for slide in analyzed.slides: summary += f"\n {slide.slide_number}. {slide.title} ({len(slide.main_points)} points)" return summary def print_detailed_analysis(self, video_id: str): """Print detailed analysis with all slide content""" analyzed = self.load_analysis(video_id) if not analyzed or not analyzed.success: print(f"āŒ No analysis available for {video_id}") return print(f"\nšŸŽ¬ **DETAILED ANALYSIS: {analyzed.video_title}**") print("=" * 60) print(f"šŸŽÆ Main Topic: {analyzed.main_topic}") print(f"šŸ“Š Total Slides: {analyzed.total_slides}") print(f"ā±ļø Estimated Time: {analyzed.estimated_presentation_time:.1f} minutes") print(f"šŸ”‘ Key Themes: {', '.join(analyzed.key_themes)}") print(f"\nšŸ“‹ **SLIDE BREAKDOWN:**") print("-" * 40) for slide in analyzed.slides: print(f"\nšŸŽÆ **Slide {slide.slide_number}: {slide.title}** ({slide.slide_type})") if slide.main_points: print(" šŸ“Œ Main Points:") for point in slide.main_points: print(f" ā€¢ {point}") if slide.supporting_details: print(" šŸ“ Supporting Details:") for detail in slide.supporting_details: print(f" - {detail}") if slide.key_quotes: print(" šŸ’¬ Key Quotes:") for quote in slide.key_quotes: print(f" \"{quote}\"") timestamp_start = int(slide.timestamp_range[0]) timestamp_end = int(slide.timestamp_range[1]) print(f" ā° Timestamp: {timestamp_start//60}:{timestamp_start%60:02d} - {timestamp_end//60}:{timestamp_end%60:02d}") def main(): """Main function to demonstrate usage""" # Initialize manager manager = TranscriptAnalysisManager() # List available transcripts print("šŸ“ Available Transcripts:") print("=" * 40) transcripts = manager.list_available_transcripts() for i, transcript in enumerate(transcripts, 1): status = "āœ…" if transcript['success'] else "āŒ" print(f"{i}. {status} {transcript['video_id']}: {transcript['title']}") print(f" Words: {transcript['word_count']:,} | Duration: {transcript['duration']:.0f}s | Method: {transcript['extraction_method']}") if not transcripts: print("āŒ No transcripts found. Make sure you have saved some transcripts first.") return # Analyze the first successful transcript as example successful_transcripts = [t for t in transcripts if t['success']] if successful_transcripts: example_video_id = successful_transcripts[0]['video_id'] print(f"\nšŸ”„ Analyzing example transcript: {example_video_id}") analyzed = manager.analyze_transcript(example_video_id, target_slides=8) if analyzed: print(f"\n{manager.get_analysis_summary(example_video_id)}") # Show detailed analysis choice = input(f"\nShow detailed analysis for {example_video_id}? (y/N): ") if choice.lower() == 'y': manager.print_detailed_analysis(example_video_id) if __name__ == "__main__": main()