From 4a98b2e4c0399a8177bb82591a62d4cf54ff24ef Mon Sep 17 00:00:00 2001 From: dan22k Date: Thu, 23 Oct 2025 16:28:49 -0400 Subject: [PATCH 1/3] Add content analysis, parallel fetching, and export features --- enhanced_channel_analyzer.py | 179 +++++++++++++++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 enhanced_channel_analyzer.py diff --git a/enhanced_channel_analyzer.py b/enhanced_channel_analyzer.py new file mode 100644 index 0000000..6eedd39 --- /dev/null +++ b/enhanced_channel_analyzer.py @@ -0,0 +1,179 @@ +""" +Enhanced YouTube Channel Analyzer +Forked and improved from vmeylan/youtube_video_and_transcript_downloader +Additions: +- Content analysis (sentiment, complexity) +- Parallel transcript fetching +- Export to multiple formats +- Progress tracking +""" + +import os +import asyncio +from concurrent.futures import ThreadPoolExecutor +from typing import List, Dict +from datetime import datetime +import json +import pandas as pd +from tqdm import tqdm # Progress bars + +# Original imports +from youtube_transcript_api import YouTubeTranscriptApi +from googleapiclient.discovery import build +from dotenv import load_dotenv + +# New analysis imports +from textblob import TextBlob # Sentiment analysis +import nltk +from textstat import flesch_reading_ease + +class EnhancedChannelAnalyzer: + """ + Enhanced version with analysis features + """ + + def __init__(self, api_key: str): + self.youtube = build('youtube', 'v3', developerKey=api_key) + self.executor = ThreadPoolExecutor(max_workers=5) # Parallel fetching + + def analyze_sentiment(self, text: str) -> Dict: + """Add sentiment analysis""" + blob = TextBlob(text) + return { + 'polarity': blob.sentiment.polarity, + 'subjectivity': blob.sentiment.subjectivity, + 'sentiment': 'positive' if blob.sentiment.polarity > 0 else 'negative' + } + + def analyze_complexity(self, text: str) -> Dict: + """Analyze text complexity""" + return { + 'reading_ease': flesch_reading_ease(text), + 'word_count': len(text.split()), + 'avg_word_length': sum(len(word) for word in text.split()) / len(text.split()) + } + + def extract_topics(self, text: str, num_topics: int = 5) -> List[str]: + """Extract key topics using TF-IDF""" + # Simple version - can be enhanced with sklearn + words = nltk.word_tokenize(text.lower()) + stop_words = set(nltk.corpus.stopwords.words('english')) + words = [w for w in words if w.isalnum() and w not in stop_words and len(w) > 3] + + from collections import Counter + word_freq = Counter(words) + return [word for word, _ in word_freq.most_common(num_topics)] + + async def fetch_transcript_async(self, video_id: str) -> Dict: + """Async transcript fetching for speed""" + loop = asyncio.get_event_loop() + + try: + # Run in thread pool to not block + transcript = await loop.run_in_executor( + self.executor, + YouTubeTranscriptApi.get_transcript, + video_id + ) + + text = ' '.join([seg['text'] for seg in transcript]) + + # Add analysis + analysis = { + 'text': text, + 'word_count': len(text.split()), + **self.analyze_sentiment(text), + **self.analyze_complexity(text), + 'topics': self.extract_topics(text) + } + + return analysis + + except Exception as e: + return {'error': str(e)} + + async def analyze_channel_parallel(self, channel_id: str, max_videos: int = 10): + """Fetch and analyze videos in parallel""" + + # Get video list (sync) + videos = self.get_channel_videos(channel_id, max_videos) + + # Create progress bar + pbar = tqdm(total=len(videos), desc="Fetching transcripts") + + # Fetch all transcripts in parallel + tasks = [] + for video in videos: + task = self.fetch_transcript_async(video['video_id']) + tasks.append(task) + + results = await asyncio.gather(*tasks) + + # Combine with video metadata + for video, analysis in zip(videos, results): + video['analysis'] = analysis + pbar.update(1) + + pbar.close() + return videos + + def get_channel_videos(self, channel_id: str, max_videos: int) -> List[Dict]: + """Get video list from channel""" + # Implementation from original repo + # ... (reuse existing code) + pass + + def export_results(self, results: Dict, format: str = 'all'): + """Export in multiple formats""" + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + + if format in ['json', 'all']: + with open(f'analysis_{timestamp}.json', 'w') as f: + json.dump(results, f, indent=2) + + if format in ['csv', 'all']: + df = pd.DataFrame(results) + df.to_csv(f'analysis_{timestamp}.csv', index=False) + + if format in ['markdown', 'all']: + self.export_markdown_report(results, f'report_{timestamp}.md') + + def export_markdown_report(self, results: Dict, filename: str): + """Create a nice markdown report""" + with open(filename, 'w') as f: + f.write("# YouTube Channel Analysis Report\n\n") + f.write(f"Generated: {datetime.now()}\n\n") + + # Summary statistics + total_words = sum(v.get('analysis', {}).get('word_count', 0) for v in results) + avg_sentiment = sum(v.get('analysis', {}).get('polarity', 0) for v in results) / len(results) + + f.write("## Summary\n") + f.write(f"- Videos analyzed: {len(results)}\n") + f.write(f"- Total words: {total_words:,}\n") + f.write(f"- Average sentiment: {avg_sentiment:.2f}\n\n") + + # Per-video analysis + f.write("## Video Analysis\n\n") + for video in results: + f.write(f"### {video['title']}\n") + if 'analysis' in video and 'error' not in video['analysis']: + f.write(f"- Words: {video['analysis']['word_count']:,}\n") + f.write(f"- Sentiment: {video['analysis']['sentiment']}\n") + f.write(f"- Topics: {', '.join(video['analysis']['topics'])}\n") + f.write(f"- Reading ease: {video['analysis']['reading_ease']:.1f}\n\n") + +# Usage +async def main(): + load_dotenv() + analyzer = EnhancedChannelAnalyzer(os.getenv('YOUTUBE_API_KEY')) + + results = await analyzer.analyze_channel_parallel( + channel_id='UC_channel_id', + max_videos=10 + ) + + analyzer.export_results(results, format='all') + +if __name__ == "__main__": + asyncio.run(main()) From ac5eabd7db863cc1ffb7af147d6f1bbb736728f5 Mon Sep 17 00:00:00 2001 From: dan22k Date: Thu, 23 Oct 2025 16:30:32 -0400 Subject: [PATCH 2/3] Add content analysis, parallel fetching, and export features --- enhanced_channel_analyzer.py | 58 ++++++++++++++++++------------------ requirements.txt | 8 +++++ 2 files changed, 37 insertions(+), 29 deletions(-) diff --git a/enhanced_channel_analyzer.py b/enhanced_channel_analyzer.py index 6eedd39..24caead 100644 --- a/enhanced_channel_analyzer.py +++ b/enhanced_channel_analyzer.py @@ -31,11 +31,11 @@ class EnhancedChannelAnalyzer: """ Enhanced version with analysis features """ - + def __init__(self, api_key: str): self.youtube = build('youtube', 'v3', developerKey=api_key) self.executor = ThreadPoolExecutor(max_workers=5) # Parallel fetching - + def analyze_sentiment(self, text: str) -> Dict: """Add sentiment analysis""" blob = TextBlob(text) @@ -44,7 +44,7 @@ def analyze_sentiment(self, text: str) -> Dict: 'subjectivity': blob.sentiment.subjectivity, 'sentiment': 'positive' if blob.sentiment.polarity > 0 else 'negative' } - + def analyze_complexity(self, text: str) -> Dict: """Analyze text complexity""" return { @@ -52,22 +52,22 @@ def analyze_complexity(self, text: str) -> Dict: 'word_count': len(text.split()), 'avg_word_length': sum(len(word) for word in text.split()) / len(text.split()) } - + def extract_topics(self, text: str, num_topics: int = 5) -> List[str]: """Extract key topics using TF-IDF""" # Simple version - can be enhanced with sklearn words = nltk.word_tokenize(text.lower()) stop_words = set(nltk.corpus.stopwords.words('english')) words = [w for w in words if w.isalnum() and w not in stop_words and len(w) > 3] - + from collections import Counter word_freq = Counter(words) return [word for word, _ in word_freq.most_common(num_topics)] - + async def fetch_transcript_async(self, video_id: str) -> Dict: """Async transcript fetching for speed""" loop = asyncio.get_event_loop() - + try: # Run in thread pool to not block transcript = await loop.run_in_executor( @@ -75,9 +75,9 @@ async def fetch_transcript_async(self, video_id: str) -> Dict: YouTubeTranscriptApi.get_transcript, video_id ) - + text = ' '.join([seg['text'] for seg in transcript]) - + # Add analysis analysis = { 'text': text, @@ -86,73 +86,73 @@ async def fetch_transcript_async(self, video_id: str) -> Dict: **self.analyze_complexity(text), 'topics': self.extract_topics(text) } - + return analysis - + except Exception as e: return {'error': str(e)} - + async def analyze_channel_parallel(self, channel_id: str, max_videos: int = 10): """Fetch and analyze videos in parallel""" - + # Get video list (sync) videos = self.get_channel_videos(channel_id, max_videos) - + # Create progress bar pbar = tqdm(total=len(videos), desc="Fetching transcripts") - + # Fetch all transcripts in parallel tasks = [] for video in videos: task = self.fetch_transcript_async(video['video_id']) tasks.append(task) - + results = await asyncio.gather(*tasks) - + # Combine with video metadata for video, analysis in zip(videos, results): video['analysis'] = analysis pbar.update(1) - + pbar.close() return videos - + def get_channel_videos(self, channel_id: str, max_videos: int) -> List[Dict]: """Get video list from channel""" # Implementation from original repo # ... (reuse existing code) pass - + def export_results(self, results: Dict, format: str = 'all'): """Export in multiple formats""" timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') - + if format in ['json', 'all']: with open(f'analysis_{timestamp}.json', 'w') as f: json.dump(results, f, indent=2) - + if format in ['csv', 'all']: df = pd.DataFrame(results) df.to_csv(f'analysis_{timestamp}.csv', index=False) - + if format in ['markdown', 'all']: self.export_markdown_report(results, f'report_{timestamp}.md') - + def export_markdown_report(self, results: Dict, filename: str): """Create a nice markdown report""" with open(filename, 'w') as f: f.write("# YouTube Channel Analysis Report\n\n") f.write(f"Generated: {datetime.now()}\n\n") - + # Summary statistics total_words = sum(v.get('analysis', {}).get('word_count', 0) for v in results) avg_sentiment = sum(v.get('analysis', {}).get('polarity', 0) for v in results) / len(results) - + f.write("## Summary\n") f.write(f"- Videos analyzed: {len(results)}\n") f.write(f"- Total words: {total_words:,}\n") f.write(f"- Average sentiment: {avg_sentiment:.2f}\n\n") - + # Per-video analysis f.write("## Video Analysis\n\n") for video in results: @@ -167,12 +167,12 @@ def export_markdown_report(self, results: Dict, filename: str): async def main(): load_dotenv() analyzer = EnhancedChannelAnalyzer(os.getenv('YOUTUBE_API_KEY')) - + results = await analyzer.analyze_channel_parallel( channel_id='UC_channel_id', max_videos=10 ) - + analyzer.export_results(results, format='all') if __name__ == "__main__": diff --git a/requirements.txt b/requirements.txt index c80bddb..2ca72dc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -34,3 +34,11 @@ uritemplate==4.1.1 urllib3==1.26.15 youtube-transcript-api==0.5.0 yt-dlp==2024.3.10 + +# Enhanced features +textblob>=0.17.0 +nltk>=3.8 +textstat>=0.7.3 +pandas>=2.0.0 +tqdm>=4.65.0 +asyncio>=3.4.3 From 3d79bb945eec33d01644c59c67f6567bcd8e9f6f Mon Sep 17 00:00:00 2001 From: dan22k Date: Thu, 23 Oct 2025 16:33:59 -0400 Subject: [PATCH 3/3] Add initial improvements --- IMPROVEMENTS.md | 1 + readme.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 IMPROVEMENTS.md diff --git a/IMPROVEMENTS.md b/IMPROVEMENTS.md new file mode 100644 index 0000000..8ffc503 --- /dev/null +++ b/IMPROVEMENTS.md @@ -0,0 +1 @@ +# My Improvements diff --git a/readme.md b/readme.md index 0ff5317..0e1fec3 100644 --- a/readme.md +++ b/readme.md @@ -75,3 +75,4 @@ This will use the API key and other parameters specified in the `.env` file. If you have suggestions or improvements, feel free to submit a pull request or open an issue. +# Enhanced version