Improvements

IMPROVEMENTS.md

@@ -0,0 +1 @@
+# My Improvements

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())

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

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