Add language-aware summarization and DOCX generation

.gitignore

@@ -1,18 +1,71 @@
-.ipynb_checkpoints/
-__pycache__/
-*.py[cod]
-*$py.class
+# Project-specific directories
+Documents/
+Image/
+Images/
+
+# Environment variables
 .env
+.env.*
+
+# Python virtual environments
+.venv/
 venv/
+ENV/
+env/
+
+# Python bytecode
 __pycache__/
-*.pyc
-*.pyo
+*.py[cod]
 *$py.class
-*~
+*.so
+.Python
+
+# Distribution / packaging
+dist/
+build/
+*.egg-info/
+*.egg
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IDE specific files
+.idea/
+.vscode/
 *.swp
+*.swo
+*~
+.DS_Store
+
+# Logs
+logs/
 *.log
-/instance
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
+
+# Local development settings
+instance/
+.webassets-cache
+
+# Dependency directories
+node_modules/
+
+# Compiled Python modules
+*.pyc
+*.pyo
+*.pyd
+
+# Temporary files
+*.bak
+*.tmp
+*.temp

app.py

@@ -1,33 +1,115 @@
 import gradio as gr
-from tool import YouTubeTranscriptExtractor, TranscriptSummarizer
+import os
+from dotenv import load_dotenv
+from tool import YouTubeTranscriptExtractor, TranscriptSummarizer, TranscriptToDocx
+
+# Load environment variables
+load_dotenv()
 
 youtube_tool = YouTubeTranscriptExtractor()
+docx_tool = TranscriptToDocx()
 #summarizer_tool = TranscriptSummarizer()
 
-def process_youtube_video(video_url, hf_api_key):
+def process_youtube_video(video_url, hf_api_key, existing_docx_path=None):
+    # Initialize tools
     summarizer_tool = TranscriptSummarizer(hf_api_key=hf_api_key)
-    transcript = youtube_tool.forward(video_url=video_url)
-    summary_and_blog = summarizer_tool.forward(transcript=transcript)
+
+    # Get video title
+    from pytubefix import YouTube
+    try:
+        yt = YouTube(video_url)
+        video_title = yt.title
+    except Exception:
+        video_title = "YouTube Video"
+
+    # Extract transcript and detect language
+    transcript_result = youtube_tool.forward(video_url=video_url)
+
+    # Parse the formatted string response
+    # Format: "LANGUAGE:lang||transcript_text"
+    try:
+        if "LANGUAGE:" in transcript_result and "||" in transcript_result:
+            parts = transcript_result.split("||", 1)
+            language = parts[0].replace("LANGUAGE:", "").strip()
+            transcript = parts[1]
+            print(f"Detected language: {language}")
+        else:
+            # Fallback if we didn't get the expected format
+            transcript = transcript_result
+            language = "en"
+            print("Warning: Could not detect language, using English as default")
+    except Exception as e:
+        transcript = transcript_result if isinstance(transcript_result, str) else "Error extracting transcript"
+        language = "en"
+        print(f"Warning: Error parsing transcript data: {str(e)}, using English as default")
+
+    # Generate summary and get image URL
+    summary_and_blog = summarizer_tool.forward(transcript=transcript, language=language)
     try:
-        summary, image_url = summary_and_blog.split("\n\nImage URL: ")
-    except ValueError:
+        if "\n\nImage URL: " in summary_and_blog:
+            summary, image_url = summary_and_blog.split("\n\nImage URL: ")
+        else:
+            summary = summary_and_blog
+            image_url = None
+    except Exception:
         summary = summary_and_blog
         image_url = None
-    return transcript, summary, image_url
-
-iface = gr.Interface(
-    fn=process_youtube_video,
-    inputs=[
-        gr.Textbox(label="YouTube Video URL"),
-        gr.Textbox(label="Hugging Face API Key", type="password")
-    ],
-    outputs=[
-        gr.Textbox(label="Transcript"),
-        gr.Textbox(label="Summary and Blog Content"),
-        gr.Image(label="Generated Image", image_mode="RGBA")
-    ],
-    title="YouTube Transcript Summarizer and Blog Content Generator",
-    description="Enter a YouTube video URL and Hugging Face API Key to extract the transcript, summarize it, and generate blog content with an image."
-)
+
+    # Generate or update DOCX file
+    # Handle the file path from Gradio
+    docx_file_path = None
+    if existing_docx_path is not None and existing_docx_path != "" and existing_docx_path != []:
+        # If it's a temporary file path from Gradio
+        if isinstance(existing_docx_path, str) and os.path.exists(existing_docx_path):
+            docx_file_path = existing_docx_path
+        # If it's a file object from Gradio
+        elif hasattr(existing_docx_path, 'name') and os.path.exists(existing_docx_path.name):
+            docx_file_path = existing_docx_path.name
+        # If it's a list (Gradio sometimes returns a list for file components)
+        elif isinstance(existing_docx_path, list) and len(existing_docx_path) > 0 and existing_docx_path[0] is not None:
+            if isinstance(existing_docx_path[0], str) and os.path.exists(existing_docx_path[0]):
+                docx_file_path = existing_docx_path[0]
+            elif hasattr(existing_docx_path[0], 'name') and os.path.exists(existing_docx_path[0].name):
+                docx_file_path = existing_docx_path[0].name
+
+    docx_path = docx_tool.forward(
+        transcript=transcript,
+        summary=summary,
+        video_title=video_title,
+        image_path=image_url,
+        existing_docx_path=docx_file_path
+    )
+
+    return transcript, summary, image_url, docx_path
+
+with gr.Blocks() as demo:
+    gr.Markdown("# YouTube Transcript Summarizer and Blog Content Generator")
+    gr.Markdown("Enter a YouTube video URL and Hugging Face API Key to extract the transcript, summarize it, and generate blog content with an image and DOCX file. Optionally, you can provide an existing DOCX file to update.")
+
+    # Check if Gemini API key is set
+    gemini_api_key = os.getenv("GEMINI_API_KEY")
+    if not gemini_api_key or gemini_api_key == "your_gemini_api_key_here":
+        gr.Markdown("⚠️ **Warning**: Gemini API key is not set in the .env file. Please add your Gemini API key to the .env file to use the summarization feature.")
+
+    with gr.Row():
+        with gr.Column():
+            video_url = gr.Textbox(label="YouTube Video URL")
+            hf_api_key = gr.Textbox(label="Hugging Face API Key", type="password")
+            existing_docx = gr.File(label="Existing DOCX file (optional)", file_types=[".docx"])
+            submit_btn = gr.Button("Process Video")
+
+        with gr.Column():
+            transcript_output = gr.Textbox(label="Transcript")
+            summary_output = gr.Textbox(label="Summary and Blog Content")
+            image_output = gr.Image(label="Generated Image", image_mode="RGBA")
+            docx_output = gr.File(label="Generated DOCX File")
+
+    submit_btn.click(
+        fn=process_youtube_video,
+        inputs=[video_url, hf_api_key, existing_docx],
+        outputs=[transcript_output, summary_output, image_output, docx_output]
+    )
+
+iface = demo
 
 iface.launch()

tool.py

@@ -1,71 +1,171 @@
 from smolagents.tools import Tool
-from typing import Optional
+from typing import Optional, Union, Dict, Any
 import os
-from transformers import pipeline
+import time
 import requests
 import io
 from PIL import Image
 from pytubefix import YouTube
-#from dotenv import load_dotenv
+import docx
+from docx.shared import Pt, RGBColor, Inches
+from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
+import google.generativeai as genai
+from dotenv import load_dotenv
 
-#load_dotenv()
+# Load environment variables
+load_dotenv()
 
 class TranscriptSummarizer(Tool):
-    description = "Summarizes a transcript and generates blog content using the transformers library and Hugging Face API for image generation."
+    description = "Summarizes a transcript and generates blog content using Google's Gemini model for summarization and Hugging Face API for image generation."
     name = "transcript_summarizer"
-    inputs = {'transcript': {'type': 'string', 'description': 'The transcript to summarize.'}}
+    inputs = {
+        'transcript': {'type': 'string', 'description': 'The transcript to summarize.'},
+        'language': {'type': 'string', 'description': 'The language of the transcript.', 'nullable': True}
+    }
     output_type = "string"
 
     def __init__(self, *args, hf_api_key: str = None, **kwargs):
         super().__init__(*args, **kwargs)
-        self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
+        # Get Gemini API key from environment variables
+        gemini_api_key = os.getenv("GEMINI_API_KEY")
+        if gemini_api_key:
+            # Configure the Gemini API
+            genai.configure(api_key=gemini_api_key)
+            # Set up the model
+            self.gemini_model = genai.GenerativeModel('gemini-2.0-flash')
+        else:
+            self.gemini_model = None
+
+        # Set up Hugging Face for image generation
         self.api_url = "https://api-inference.huggingface.co/models/ZB-Tech/Text-to-Image"
         self.hf_api_key = hf_api_key
         self.headers = {"Authorization": f"Bearer {self.hf_api_key}"}
 
-    def query(self, payload):
+    def query_image_api(self, payload):
         response = requests.post(self.api_url, headers=self.headers, json=payload)
         return response.content
 
-    def forward(self, transcript: str) -> str:
+    def summarize_with_gemini(self, text, language='en', max_tokens=1000):
+        """Use Gemini to summarize text in the specified language"""
+        # Map language codes to full language names for better prompting
+        language_map = {
+            'en': 'English',
+            'hi': 'Hindi',
+            'es': 'Spanish',
+            'fr': 'French',
+            'de': 'German',
+            'it': 'Italian',
+            'ja': 'Japanese',
+            'ko': 'Korean',
+            'pt': 'Portuguese',
+            'ru': 'Russian',
+            'zh': 'Chinese',
+            'ar': 'Arabic',
+            'bn': 'Bengali',
+            'ta': 'Tamil',
+            'te': 'Telugu',
+            'mr': 'Marathi',
+            'gu': 'Gujarati',
+            'kn': 'Kannada',
+            'ml': 'Malayalam',
+            'pa': 'Punjabi',
+            'ur': 'Urdu'
+            # Add more languages as needed
+        }
+
+        language_name = language_map.get(language, language)
+
+        prompt = f"""
+        Please summarize the following transcript in a concise but comprehensive way.
+        Focus on the main points and key information.
+
+        IMPORTANT: The transcript is in {language_name}. Please provide the summary in the SAME LANGUAGE ({language_name}).
+        Do not translate to any other language. Keep the summary in the original language of the transcript.
+
+        Transcript:
+        {text}
+        """
+
+        generation_config = {
+            "temperature": 0.4,
+            "top_p": 0.95,
+            "top_k": 40,
+            "max_output_tokens": max_tokens,
+        }
+
+        response = self.gemini_model.generate_content(
+            prompt,
+            generation_config=generation_config
+        )
+
+        return response.text
+
+    def forward(self, transcript: str, language: str = 'en') -> str:
         try:
             if not self.hf_api_key:
-                return "Hugging Face API key is required. Please provide it in the input field."
+                return "Hugging Face API key is required for image generation. Please provide it in the input field."
 
-            transcript_length = len(transcript)
+            if not self.gemini_model:
+                return "Gemini API key is required for summarization. Please add it to your .env file."
 
-            def get_summary_lengths(length):
-                # set the short maths formula
-                max_length = int(length * 0.8)
-                min_length = int(length * 0.2)
-                return max_length, min_length
+            transcript_length = len(transcript)
 
-            # Split the transcript into chunks of 500 characters make it dynamic according to the length of the transcript
-            if transcript_length < 500:
+            # Check if transcript is too short
+            if transcript_length < 100:
                 return "Transcript is too short to summarize."
-            chunk_size = 500
-            transcript_chunks = [transcript[i:i+chunk_size] for i in range(0, len(transcript), chunk_size)]
-
-            # Summarize each chunk of the transcript
-            summaries = []
-            for chunk in transcript_chunks:
-                max_length, min_length = get_summary_lengths(len(chunk))
-                summary = self.summarizer(chunk, max_length=max_length, min_length=min_length, do_sample=False)[0]['summary_text']
-                summaries.append(summary)
-
-            # Concatenate the summaries
-            full_summary = "\n".join(summaries)
-
-            key_entities = full_summary.split()[:15]  # Extract first 15 words as key entities
-            image_prompt = f"Generate an image related to: {' '.join(key_entities)}, cartoon style"
-            image_bytes = self.query({"inputs": image_prompt})
-            image = Image.open(io.BytesIO(image_bytes))
-            image_folder = "Image"
-            if not os.path.exists(image_folder):
-                os.makedirs(image_folder)
-            image_url = os.path.join(image_folder, "image.jpg")  # Specify the folder path
-            image.save(image_url) # Save the image to a file
-            return f"{full_summary}\n\nImage URL: {image_url}" # Return the file path
+
+            # For longer transcripts, split into chunks to handle context window limitations
+            if transcript_length > 30000:  # Gemini has a context window limit
+                chunk_size = 25000
+                transcript_chunks = [transcript[i:i+chunk_size] for i in range(0, len(transcript), chunk_size)]
+
+                # Summarize each chunk
+                chunk_summaries = []
+                for chunk in transcript_chunks:
+                    chunk_summary = self.summarize_with_gemini(chunk, language=language, max_tokens=1000)
+                    chunk_summaries.append(chunk_summary)
+
+                # Combine chunk summaries and create a final summary
+                combined_summary = "\n\n".join(chunk_summaries)
+                if len(combined_summary) > 25000:
+                    full_summary = self.summarize_with_gemini(combined_summary, language=language, max_tokens=2000)
+                else:
+                    full_summary = combined_summary
+            else:
+                # For shorter transcripts, summarize directly
+                full_summary = self.summarize_with_gemini(transcript, language=language, max_tokens=2000)
+
+            # Generate image based on summary
+            try:
+                key_entities = full_summary.split()[:15]  # Extract first 15 words as key entities
+                image_prompt = f"Generate an image related to: {' '.join(key_entities)}, cartoon style"
+                image_bytes = self.query_image_api({"inputs": image_prompt})
+
+                # Check if the response is valid
+                if not image_bytes or len(image_bytes) < 100:
+                    print("Warning: Received invalid or empty image response")
+                    return full_summary  # Return just the summary without image
+
+                try:
+                    # Try to open the image
+                    image = Image.open(io.BytesIO(image_bytes))
+
+                    # Save the image
+                    image_folder = "Image"
+                    if not os.path.exists(image_folder):
+                        os.makedirs(image_folder)
+                    image_url = os.path.join(image_folder, f"image_{int(time.time())}.jpg")  # Use timestamp for unique filename
+                    image.save(image_url)
+
+                    return f"{full_summary}\n\nImage URL: {image_url}"  # Return the file path with summary
+                except Exception as img_error:
+                    print(f"Error processing image: {str(img_error)}")
+                    # Return just the summary if image processing fails
+                    return full_summary
+            except Exception as img_gen_error:
+                print(f"Error generating image: {str(img_gen_error)}")
+                # Return just the summary if image generation fails
+                return full_summary
         except Exception as e:
             return f"An unexpected error occurred: {str(e)}"
 
@@ -73,24 +173,31 @@ class YouTubeTranscriptExtractor(Tool):
     description = "Extracts the transcript from a YouTube video."
     name = "youtube_transcript_extractor"
     inputs = {'video_url': {'type': 'string', 'description': 'The URL of the YouTube video.'}}
-    output_type = "string"
+    output_type = "string"  # Keep as string for compatibility with smolagents
 
     def forward(self, video_url: str) -> str:
         try:
             # Create a YouTube object
             yt = YouTube(video_url)
-            lang='en'
-           # Get the video transcript
+            lang = 'en'  # Default language
+
+            # Get the video transcript
             try:
-                if lang in yt.captions:
+                if 'en' in yt.captions:
                     transcript = yt.captions['en'].generate_srt_captions()
+                    lang = 'en'
                 else:
-                    transcript = yt.captions.all()[0].generate_srt_captions()
-                    lang = yt.captions.all()[0].code
+                    # Get the first available caption
+                    if len(yt.captions.all()) > 0:
+                        caption = yt.captions.all()[0]
+                        transcript = caption.generate_srt_captions()
+                        lang = caption.code
+                    else:
+                        return f"LANGUAGE:{lang}||No transcript available for this video."
             except StopIteration:
-                return "No transcript available for this video."
+                return f"LANGUAGE:{lang}||No transcript available for this video."
             except Exception as e:
-                return f"An unexpected error occurred while accessing captions: {str(e)}"
+                return f"LANGUAGE:{lang}||An unexpected error occurred while accessing captions: {str(e)}"
 
             # Clean up the transcript by removing timestamps and line numbers
             cleaned_transcript = ""
@@ -98,10 +205,95 @@ class YouTubeTranscriptExtractor(Tool):
                 if not line.strip().isdigit() and "-->" not in line:
                     cleaned_transcript += line + "\n"
 
-            print("transcript : ", cleaned_transcript)
-            return cleaned_transcript
+            print(f"Transcript language detected: {lang}")
+            print("Transcript sample: ", cleaned_transcript[:200] + "..." if len(cleaned_transcript) > 200 else cleaned_transcript)
+
+            # Return both the transcript and the language as a formatted string
+            # Format: "LANGUAGE:lang||transcript_text"
+            return f"LANGUAGE:{lang}||{cleaned_transcript}"
         except Exception as e:
-            return f"An unexpected error occurred: {str(e)}"
+            return f"LANGUAGE:en||An unexpected error occurred: {str(e)}"
 
     def __init__(self, *args, **kwargs):
         self.is_initialized = False
+
+class TranscriptToDocx(Tool):
+    description = "Creates or updates a DOCX file with YouTube transcript and summary."
+    name = "transcript_to_docx"
+    inputs = {
+        'transcript': {'type': 'string', 'description': 'The transcript to include in the document.'},
+        'summary': {'type': 'string', 'description': 'The summary to include in the document.'},
+        'video_title': {'type': 'string', 'description': 'The title of the YouTube video.'},
+        'image_path': {'type': 'string', 'description': 'Path to the image to include in the document.', 'nullable': True},
+        'existing_docx_path': {'type': 'string', 'description': 'Path to an existing DOCX file to update.', 'nullable': True}
+    }
+    output_type = "string"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.docx_folder = "Documents"
+        if not os.path.exists(self.docx_folder):
+            os.makedirs(self.docx_folder)
+
+    def forward(self, transcript: str, summary: str, video_title: str, image_path: Optional[str] = None, existing_docx_path: Optional[str] = None) -> str:
+        try:
+            # Determine if we're creating a new document or updating an existing one
+            if existing_docx_path and os.path.exists(existing_docx_path):
+                doc = docx.Document(existing_docx_path)
+                # Add a page break before adding new content
+                doc.add_paragraph().add_run().add_break(docx.enum.text.WD_BREAK.PAGE)
+            else:
+                doc = docx.Document()
+                # Set document properties
+                doc.core_properties.title = f"YouTube Transcript: {video_title}"
+                doc.core_properties.author = "YouTube Transcript Tool"
+
+            # Add title
+            title = doc.add_heading(video_title, level=1)
+            title.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
+
+            # Add summary section
+            doc.add_heading("Summary", level=2)
+            summary_para = doc.add_paragraph(summary)
+
+            # Add image if provided
+            if image_path and os.path.exists(image_path):
+                try:
+                    doc.add_picture(image_path, width=Inches(6))
+                    # Add caption for the image
+                    caption = doc.add_paragraph("Generated image based on transcript content")
+                    caption.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
+                    caption.runs[0].italic = True
+                except Exception as img_error:
+                    # If there's an error adding the image, just log it and continue
+                    print(f"Error adding image to document: {str(img_error)}")
+
+            # Add transcript section
+            doc.add_heading("Full Transcript", level=2)
+            transcript_para = doc.add_paragraph(transcript)
+
+            # Clean the video title for filename
+            safe_title = ''.join(c for c in video_title if c.isalnum() or c in ' _-')
+            safe_title = safe_title.replace(' ', '_')
+
+            # Save the document
+            output_filename = f"{safe_title}.docx"
+            output_path = os.path.join(self.docx_folder, output_filename)
+
+            try:
+                doc.save(output_path)
+                print(f"Document saved successfully at: {output_path}")
+                return output_path
+            except Exception as save_error:
+                error_msg = f"Error saving document: {str(save_error)}"
+                print(error_msg)
+                # Try with a simpler filename as fallback
+                try:
+                    fallback_path = os.path.join(self.docx_folder, f"youtube_transcript_{int(time.time())}.docx")
+                    doc.save(fallback_path)
+                    print(f"Document saved with fallback name at: {fallback_path}")
+                    return fallback_path
+                except:
+                    return error_msg
+        except Exception as e:
+            return f"An error occurred while creating the DOCX file: {str(e)}"