Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from dotenv import load_dotenv | |
| from langchain_community.document_loaders import WebBaseLoader | |
| from langchain.embeddings import HuggingFaceEmbeddings | |
| from langchain_community.vectorstores.faiss import FAISS | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| from langchain_core.documents import Document | |
| import os | |
| import json | |
| from langchain_groq import ChatGroq | |
| from langchain.chains.qa_with_sources.retrieval import RetrievalQAWithSourcesChain | |
| from langchain.prompts import PromptTemplate | |
| from bs4 import SoupStrainer | |
| from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound | |
| import yt_dlp | |
| import re | |
| from googleapiclient.discovery import build | |
| from googleapiclient.errors import HttpError | |
| from google_auth_oauthlib.flow import InstalledAppFlow | |
| from google.auth.transport.requests import Request | |
| from google.oauth2.credentials import Credentials | |
| # Load environment variables (optional) | |
| load_dotenv() | |
| # Hardcoded Groq API key | |
| GROQ_API_KEY = "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" | |
| # YouTube API key (to be set in Hugging Face Spaces secrets, optional if using OAuth) | |
| YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY") | |
| # Path to store OAuth credentials | |
| CREDENTIALS_FILE = "youtube_credentials.json" | |
| CLIENT_SECRETS_FILE = "client_secrets.json" | |
| # Custom CSS | |
| st.markdown(""" | |
| <style> | |
| body { | |
| background: linear-gradient(135deg, #1e3c72, #2a5298); | |
| color: #ffffff; | |
| font-family: 'Arial', sans-serif; | |
| } | |
| .stSidebar, .main .block-container { | |
| background: rgba(255, 255, 255, 0.1); | |
| border-radius: 15px; | |
| backdrop-filter: blur(10px); | |
| -webkit-backdrop-filter: blur(10px); | |
| border: 1px solid rgba(255, 255, 255, 0.2); | |
| box-shadow: 0 8px 32px rgba(0, 0, 0, 0.2); | |
| padding: 20px; | |
| } | |
| .stTextInput > div > input { | |
| background: rgba(255, 255, 255, 0.15); | |
| color: #ffffff; | |
| border: 1px solid rgba(255, 255, 255, 0.3); | |
| border-radius: 10px; | |
| padding: 10px; | |
| box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1); | |
| } | |
| .stButton > button { | |
| background: linear-gradient(45deg, #6b48ff, #00ddeb); | |
| color: #ffffff; | |
| border: none; | |
| border-radius: 10px; | |
| padding: 10px 20px; | |
| font-weight: bold; | |
| box-shadow: 0 4px 12px rgba(0, 0, 0, 0.2); | |
| transition: transform 0.2s; | |
| } | |
| .stButton > button:hover { | |
| transform: translateY(-2px); | |
| box-shadow: 0 6px 16px rgba(0, 0, 0, 0.3); | |
| } | |
| h1, h2, h3 { | |
| color: #ffffff; | |
| text-shadow: 0 2px 4px rgba(0, 0, 0, 0.2); | |
| } | |
| .stText { | |
| color: #e0e0e0; | |
| font-weight: bold; | |
| } | |
| .stAlert { | |
| background: rgba(255, 50, 50, 0.2); | |
| border: 1px solid rgba(255, 50, 50, 0.5); | |
| border-radius: 10px; | |
| color: #ffcccc; | |
| } | |
| .stAlert[role="alert"] > div { | |
| background: rgba(255, 200, 0, 0.2); | |
| border: 1px solid rgba(255, 200, 0, 0.5); | |
| color: #fff5cc; | |
| } | |
| .stSpinner > div { | |
| color: #00ddeb; | |
| } | |
| .footer { | |
| display: flex; | |
| align-items: center; | |
| justify-content: center; | |
| padding: 10px; | |
| background: rgba(255, 255, 255, 0.1); | |
| border-top: 1px solid rgba(255, 255, 255, 0.2); | |
| position: fixed; | |
| bottom: 0; | |
| width: 100%; | |
| color: #e0e0e0; | |
| font-size: 14px; | |
| } | |
| .footer img { | |
| margin-right: 10px; | |
| } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| # Display large logo at the top of the main page | |
| st.image("https://i.postimg.cc/2j0QWF3Z/Removal-575.png", width=390) | |
| # Set Streamlit app title | |
| st.title("WebChatter 💬") | |
| # Initialize session state | |
| if "url_content" not in st.session_state: | |
| st.session_state.url_content = None | |
| if "summary" not in st.session_state: | |
| st.session_state.summary = None | |
| if "vectorstore" not in st.session_state: | |
| st.session_state.vectorstore = None | |
| if "index_created" not in st.session_state: | |
| st.session_state.index_created = False | |
| if "content_type" not in st.session_state: | |
| st.session_state.content_type = None | |
| # Initialize LLM once at the start | |
| if "llm" not in st.session_state: | |
| st.session_state.llm = ChatGroq( | |
| api_key=GROQ_API_KEY, | |
| model="llama3-70b-8192", | |
| max_tokens=512 # Keep reduced to minimize resource usage | |
| ) | |
| # Sidebar for URL and YouTube input | |
| with st.sidebar: | |
| st.header("Enter Web URL") | |
| url = st.text_input("URL", placeholder="e.g., https://mahatirtusher.com/astronomy-mythology/") | |
| process_url_clicked = st.button("Process URL") | |
| st.header("Enter YouTube URL") | |
| youtube_url = st.text_input("YouTube URL", placeholder="e.g., https://www.youtube.com/watch?v=DJO_9auJhJQ") | |
| process_youtube_clicked = st.button("Process YouTube Video") | |
| # Main content container | |
| main_container = st.container() | |
| # Custom prompt for detailed answers (for web URLs only) | |
| qa_prompt = PromptTemplate( | |
| template="""You are an expert assistant tasked with providing detailed, extensive, and comprehensive answers. Use the provided context to answer the question thoroughly, including explanations, examples, and additional relevant information. If the context is limited, expand on the topic with your knowledge to ensure a complete response. In case of explaining anything, break the topic and explain step by step. Sometimes use your own reasoning and knowledge to explain anything to the users. If the users ask any question in Bengali, you too will answer it in fine and detailed Bengali. | |
| Context: {context} | |
| Question: {question} | |
| Answer with sources: """ | |
| ) | |
| # Function to summarize content | |
| def summarize_content(content, llm, is_youtube=False): | |
| if is_youtube: | |
| # Extensive summary for YouTube videos (15-20 sentences) | |
| summary_prompt = f"""You are an expert summarizer tasked with providing a very detailed and extensive summary of the following YouTube video transcript. Capture all key points, main ideas, and significant details in 15-20 sentences. Include specific examples, quotes, or moments from the transcript to make the summary comprehensive and vivid. Ensure the summary is well-organized, flowing naturally from one point to the next, and provides a thorough overview of the video's content. | |
| Transcript: {content} | |
| Extensive Summary: """ | |
| else: | |
| # Shorter summary for web URLs (5-10 sentences) | |
| summary_prompt = f"""Summarize the following content in 5-10 sentences, capturing the main points and key details in easy expression: | |
| {content} | |
| Summary: """ | |
| summary = llm.invoke(summary_prompt).content | |
| return summary | |
| # Function to extract YouTube video ID from URL | |
| def get_video_id(url): | |
| if "youtube.com/watch?v=" in url: | |
| return url.split("v=")[1].split("&")[0] | |
| elif "youtu.be/" in url: | |
| return url.split("youtu.be/")[1].split("?")[0] | |
| return None | |
| # Function to fetch YouTube transcript | |
| def fetch_youtube_transcript(video_id): | |
| try: | |
| transcript_list = YouTubeTranscriptApi.list_transcripts(video_id) | |
| # Try English variants first | |
| for lang in ['en', 'en-US', 'en-GB']: | |
| try: | |
| transcript = transcript_list.find_transcript([lang]).fetch() | |
| full_text = " ".join([item['text'] for item in transcript]) | |
| return full_text | |
| except NoTranscriptFound: | |
| continue | |
| # If no English transcript, try any available transcript and translate to English | |
| for transcript in transcript_list: | |
| if transcript.is_translatable: | |
| translated_transcript = transcript.translate('en').fetch() | |
| return " ".join([item['text'] for item in translated_transcript]) | |
| return None | |
| except TranscriptsDisabled: | |
| return None | |
| except Exception as e: | |
| st.error(f"Error fetching transcript with youtube-transcript-api: {str(e)}") | |
| return None | |
| # Function to get YouTube API credentials | |
| def get_youtube_credentials(): | |
| creds = None | |
| if os.path.exists(CREDENTIALS_FILE): | |
| creds = Credentials.from_authorized_user_file(CREDENTIALS_FILE, scopes=['https://www.googleapis.com/auth/youtube.force-ssl']) | |
| if not creds or not creds.valid: | |
| if creds and creds.expired and creds.refresh_token: | |
| creds.refresh(Request()) | |
| else: | |
| if os.path.exists(CLIENT_SECRETS_FILE): | |
| st.warning("Attempting to authenticate with YouTube Data API. This may not work in Hugging Face Spaces due to redirect URI limitations.") | |
| flow = InstalledAppFlow.from_client_secrets_file( | |
| CLIENT_SECRETS_FILE, | |
| scopes=['https://www.googleapis.com/auth/youtube.force-ssl'] | |
| ) | |
| # This will fail in Hugging Face Spaces because it can't open a browser | |
| creds = flow.run_local_server(port=0) | |
| with open(CREDENTIALS_FILE, 'w') as token_file: | |
| token_file.write(creds.to_json()) | |
| else: | |
| st.warning( | |
| f"{CLIENT_SECRETS_FILE} not found. To use OAuth 2.0 for YouTube Data API:\n" | |
| "1. Go to https://console.developers.google.com/.\n" | |
| "2. Create a project, enable YouTube Data API v3, and create OAuth 2.0 credentials.\n" | |
| "3. Download the credentials as 'client_secrets.json'.\n" | |
| "4. Run the app locally: pip install -r requirements.txt && streamlit run app.py\n" | |
| "5. Authenticate via the browser prompt to generate youtube_credentials.json.\n" | |
| "6. Upload youtube_credentials.json to your Hugging Face Space via the Files tab." | |
| ) | |
| return None | |
| return creds | |
| # Function to fetch captions using YouTube Data API (with OAuth 2.0 or API key fallback) | |
| def fetch_youtube_captions_api(video_id, api_key=None): | |
| # First, try OAuth 2.0 if credentials are available | |
| creds = get_youtube_credentials() | |
| if creds: | |
| try: | |
| youtube = build('youtube', 'v3', credentials=creds) | |
| captions = youtube.captions().list( | |
| part='snippet', | |
| videoId=video_id | |
| ).execute() | |
| caption_id = None | |
| for item in captions.get('items', []): | |
| if item['snippet']['language'] == 'en': | |
| caption_id = item['id'] | |
| break | |
| elif item['snippet']['language'] in ['en-US', 'en-GB']: | |
| caption_id = item['id'] | |
| break | |
| if not caption_id: | |
| st.warning("No English captions found via YouTube Data API.") | |
| return None | |
| # Download captions using OAuth 2.0 credentials | |
| caption_content = youtube.captions().download( | |
| id=caption_id, | |
| tfmt='srt' | |
| ).execute() | |
| # The response is a binary string, decode it | |
| caption_text = caption_content.decode('utf-8') | |
| # Parse SRT format to extract text | |
| lines = caption_text.split('\n') | |
| text_lines = [] | |
| for line in lines: | |
| if line.strip() and not line.isdigit() and not re.match(r'\d{2}:\d{2}:\d{2},\d{3} --> \d{2}:\d{2}:\d{2},\d{3}', line): | |
| text_lines.append(line.strip()) | |
| return " ".join(text_lines) | |
| except HttpError as e: | |
| st.error(f"Error fetching captions with YouTube Data API (OAuth 2.0): {str(e)}") | |
| return None | |
| # Fallback to API key if OAuth fails or credentials are not available | |
| if not api_key: | |
| st.warning("YOUTUBE_API_KEY not set and OAuth 2.0 credentials not available. Skipping YouTube Data API fallback.") | |
| return None | |
| try: | |
| youtube = build('youtube', 'v3', developerKey=api_key) | |
| captions = youtube.captions().list( | |
| part='snippet', | |
| videoId=video_id | |
| ).execute() | |
| caption_id = None | |
| for item in captions.get('items', []): | |
| if item['snippet']['language'] == 'en': | |
| caption_id = item['id'] | |
| break | |
| elif item['snippet']['language'] in ['en-US', 'en-GB']: | |
| caption_id = item['id'] | |
| break | |
| if not caption_id: | |
| st.warning("No English captions found via YouTube Data API.") | |
| return None | |
| # Note: Downloading captions requires OAuth 2.0 authentication | |
| st.warning( | |
| "English captions are available for this video but cannot be fetched with an API key alone. " | |
| "Downloading captions requires OAuth 2.0 authentication, which is not supported in Hugging Face Spaces without user interaction. " | |
| "To fetch captions:\n" | |
| "- Follow the instructions above to generate youtube_credentials.json locally and upload it.\n" | |
| "- Or try a video with transcripts available (e.g., https://www.youtube.com/watch?v=dQw4w9WgXcQ)." | |
| ) | |
| return None | |
| except HttpError as e: | |
| st.error(f"Error fetching captions with YouTube Data API (API Key): {str(e)}") | |
| return None | |
| # Function to extract subtitles using yt-dlp with cookies | |
| def extract_subtitles_with_ytdlp(video_url): | |
| ydl_opts = { | |
| 'writesubtitles': True, | |
| 'writeautomaticsub': True, | |
| 'subtitleslangs': ['all', '-live_chat'], | |
| 'skip_download': True, | |
| 'subtitlesformat': 'vtt', | |
| 'outtmpl': 'subtitle.%(ext)s', | |
| 'http_headers': { | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36', | |
| 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', | |
| 'Accept-Language': 'en-US,en;q=0.5', | |
| }, | |
| 'cookiefile': 'cookies.txt', | |
| } | |
| try: | |
| if not os.path.exists('cookies.txt'): | |
| st.error( | |
| "cookies.txt file not found. Please upload a valid cookies.txt file to the root directory of your Space. " | |
| "To generate it:\n" | |
| "1. Open Chrome and log in to YouTube.\n" | |
| "2. Install the 'Export Cookies' extension (or use a tool like 'cookies.txt' for Firefox).\n" | |
| "3. Export cookies for 'youtube.com' and save as 'cookies.txt'.\n" | |
| "4. Upload the file to your Space via the Files tab.\n" | |
| "Alternative: If this fails, test locally to rule out Spaces IP restrictions." | |
| ) | |
| return None | |
| with yt_dlp.YoutubeDL(ydl_opts) as ydl: | |
| info = ydl.extract_info(video_url, download=False) | |
| available_subs = info.get('subtitles', {}) | |
| auto_subs = info.get('automatic_captions', {}) | |
| subtitle_langs = list(available_subs.keys()) or list(auto_subs.keys()) | |
| if not subtitle_langs: | |
| st.warning("No subtitles or auto-captions available in any language.") | |
| return None | |
| ydl.params['subtitleslangs'] = subtitle_langs | |
| ydl.download([video_url]) | |
| subtitle_file = None | |
| for lang in subtitle_langs: | |
| possible_file = f"subtitle.{lang}.vtt" | |
| if os.path.exists(possible_file): | |
| subtitle_file = possible_file | |
| break | |
| if not subtitle_file: | |
| st.warning("No subtitle files were downloaded.") | |
| return None | |
| with open(subtitle_file, 'r', encoding='utf-8') as f: | |
| subtitle_text = f.read() | |
| os.remove(subtitle_file) | |
| lines = subtitle_text.split('\n') | |
| text_lines = [] | |
| for line in lines: | |
| if line.strip() and not line.startswith('WEBVTT') and not line.startswith('Kind:') and not line.startswith('Language:') and not re.match(r'\d{2}:\d{2}:\d{2}\.\d{3} --> \d{2}:\d{2}:\d{2}\.\d{3}', line): | |
| text_lines.append(line.strip()) | |
| return " ".join(text_lines) | |
| except Exception as e: | |
| st.error(f"Error fetching captions with yt-dlp: {str(e)}") | |
| return None | |
| # Function to process and chunk text (for web URLs only) | |
| def process_content(text, embeddings, source): | |
| text_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=1000, | |
| chunk_overlap=200, | |
| separators=["\n\n", "\n", ".", " ", ""] | |
| ) | |
| docs = text_splitter.create_documents([text], metadatas=[{"source": source}]) | |
| if not docs: | |
| st.error("No documents created from the content.") | |
| return None | |
| vectorstore = FAISS.from_documents(docs, embeddings) | |
| return vectorstore | |
| # Function to create QA chain (for web URLs only) | |
| def create_qa_chain(vectorstore, llm): | |
| if vectorstore is None: | |
| st.error("Vector store is not initialized. Cannot create QA chain.") | |
| return None | |
| retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) | |
| qa_chain = RetrievalQAWithSourcesChain.from_chain_type( | |
| llm=llm, | |
| retriever=retriever, | |
| chain_type="stuff", | |
| chain_type_kwargs={ | |
| "prompt": qa_prompt, | |
| "document_variable_name": "context" | |
| } | |
| ) | |
| return qa_chain | |
| # Process Web URL | |
| if process_url_clicked: | |
| with main_container: | |
| if not url.strip(): | |
| st.error("Please provide a valid URL.") | |
| else: | |
| with st.spinner("Processing URL..."): | |
| try: | |
| st.text("Data Loading...Started...✅✅✅") | |
| parse_only = SoupStrainer(['title', 'p', 'h1', 'h2', 'h3']) | |
| loader = WebBaseLoader( | |
| web_paths=[url.strip()], | |
| bs_kwargs={"parse_only": parse_only}, | |
| requests_kwargs={"headers": {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}}) | |
| data = loader.load() | |
| if not data or all(len(doc.page_content.strip()) == 0 for doc in data): | |
| st.error("No content loaded from URL. Try a different URL (e.g., https://www.bbc.com/news/science-environment-67299122).") | |
| st.stop() | |
| # Initialize embeddings only when needed | |
| if "embeddings" not in st.session_state: | |
| st.session_state.embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") | |
| st.session_state.url_content = "\n".join([doc.page_content for doc in data]) | |
| embeddings = st.session_state.embeddings | |
| st.session_state.vectorstore = process_content(st.session_state.url_content, embeddings, source=url.strip()) | |
| st.session_state.index_created = True | |
| st.session_state.content_type = "web" | |
| st.session_state.summary = None | |
| st.text("Content processed successfully! ✅✅✅") | |
| except Exception as e: | |
| st.error(f"Error processing URL: {str(e)}") | |
| st.stop() | |
| # Process YouTube Video | |
| if process_youtube_clicked: | |
| with main_container: | |
| if not youtube_url.strip(): | |
| st.error("Please provide a valid YouTube URL.") | |
| else: | |
| with st.spinner("Processing YouTube Video..."): | |
| try: | |
| video_id = get_video_id(youtube_url) | |
| if not video_id: | |
| st.error("Invalid YouTube URL. Please provide a URL like https://www.youtube.com/watch?v=VIDEO_ID.") | |
| st.stop() | |
| transcript_text = None | |
| st.text("Fetching Transcript...Started...✅✅✅") | |
| transcript_text = fetch_youtube_transcript(video_id) | |
| if not transcript_text: | |
| st.warning("Transcripts are disabled or unavailable. Attempting to fetch closed captions...") | |
| st.text("Fetching Closed Captions with yt-dlp...Started...✅✅✅") | |
| transcript_text = extract_subtitles_with_ytdlp(youtube_url) | |
| if not transcript_text: | |
| st.text("Fetching Captions via YouTube Data API...Started...✅✅✅") | |
| transcript_text = fetch_youtube_captions_api(video_id, YOUTUBE_API_KEY) | |
| if not transcript_text: | |
| st.error( | |
| "No transcripts or closed captions available. " | |
| "Possible reasons:\n" | |
| "1. Captions are not enabled for this video.\n" | |
| "2. YouTube detected this request as a bot (even with cookies.txt).\n" | |
| "Solutions:\n" | |
| "- Ensure captions are enabled for the video by checking the video settings on YouTube (gear icon > Subtitles/CC > Enable if available).\n" | |
| "- Regenerate and upload a fresh cookies.txt file (see instructions above).\n" | |
| "- Set up OAuth 2.0 credentials by following the instructions above to download captions directly.\n" | |
| "- Try a different video (e.g., https://www.youtube.com/watch?v=dQw4w9WgXcQ, which has transcripts available).\n" | |
| "- Test locally to rule out Hugging Face Spaces IP restrictions by running: pip install -r requirements.txt && streamlit run app.py" | |
| ) | |
| st.stop() | |
| if not transcript_text.strip(): | |
| st.error("Transcript or captions are empty. Try a different video.") | |
| st.stop() | |
| st.session_state.url_content = transcript_text | |
| # No vector store for YouTube videos since we're not doing QA | |
| st.session_state.vectorstore = None | |
| st.session_state.index_created = False | |
| st.session_state.content_type = "youtube" | |
| st.session_state.summary = None | |
| st.text("YouTube video processed successfully! ✅✅✅") | |
| except Exception as e: | |
| st.error(f"Error processing YouTube video: {str(e)}") | |
| st.stop() | |
| # Summary button | |
| with main_container: | |
| if st.session_state.url_content and st.button("Generate Summary"): | |
| with st.spinner("Generating summary..."): | |
| is_youtube = st.session_state.content_type == "youtube" | |
| st.session_state.summary = summarize_content(st.session_state.url_content, st.session_state.llm, is_youtube=is_youtube) | |
| # Display summary if generated | |
| if st.session_state.summary: | |
| with main_container: | |
| st.header("Summary of the Content") | |
| st.write(st.session_state.summary) | |
| # Query input with Ask button (only for web URLs) | |
| if st.session_state.url_content and st.session_state.content_type == "web": | |
| with main_container: | |
| st.header("Ask a Question") | |
| query = st.text_input("Question", placeholder="e.g., What is the article about?") | |
| ask_clicked = st.button("Ask") | |
| if ask_clicked and query: | |
| with st.spinner("Processing your question..."): | |
| try: | |
| if "qa_chain" not in st.session_state or st.session_state.qa_chain is None: | |
| st.session_state.qa_chain = create_qa_chain(st.session_state.vectorstore, st.session_state.llm) | |
| if st.session_state.qa_chain is None: | |
| st.error("Failed to create QA chain.") | |
| st.stop() | |
| result = st.session_state.qa_chain({"question": query}, return_only_outputs=True) | |
| if not result.get("answer"): | |
| st.warning("No answer generated. Try a different question or content.") | |
| st.stop() | |
| st.header("Answer") | |
| st.write(result["answer"]) | |
| sources = result.get("sources", "") | |
| if sources: | |
| st.subheader("Sources:") | |
| sources_list = sources.split("\n") | |
| for source in sources_list: | |
| st.write(source) | |
| else: | |
| st.write("No sources found.") | |
| except Exception as e: | |
| st.error(f"Error answering query: {str(e)}") | |
| st.stop() | |
| # Footer with tiny logo and text | |
| st.markdown( | |
| """ | |
| <div class="footer"> | |
| <img src="https://i.postimg.cc/2j0QWF3Z/Removal-575.png" width="80"> | |
| WebChatter © 2025 | Developed by Mahatir Ahmed Tusher | |
| </div> | |
| """, | |
| unsafe_allow_html=True | |
| ) |