Spaces:
Sleeping
Sleeping
| import os | |
| import gradio as gr | |
| import tempfile | |
| import shutil | |
| import re | |
| import traceback | |
| from yt_dlp import YoutubeDL | |
| # Optional: use OpenAI Whisper if available | |
| try: | |
| import whisper | |
| WHISPER_AVAILABLE = True | |
| except: | |
| WHISPER_AVAILABLE = False | |
| # ✅ Download audio from YouTube | |
| def download_audio(url, cookies_path=None): | |
| try: | |
| temp_dir = tempfile.mkdtemp() | |
| output_path = os.path.join(temp_dir, "audio") | |
| ydl_opts = { | |
| 'format': 'bestaudio[ext=m4a]/bestaudio/best', | |
| 'outtmpl': output_path + '.%(ext)s', | |
| 'quiet': True, | |
| 'noplaylist': True, | |
| 'cookiefile': cookies_path if cookies_path else None, | |
| 'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', | |
| 'referer': 'https://www.youtube.com/', | |
| 'force_ipv4': True, | |
| 'http_headers': { | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', | |
| 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', | |
| 'Accept-Language': 'en-US,en;q=0.5', | |
| 'Referer': 'https://www.youtube.com/' | |
| }, | |
| } | |
| with YoutubeDL(ydl_opts) as ydl: | |
| ydl.download([url]) | |
| for ext in [".m4a", ".webm", ".mp3"]: | |
| final_path = output_path + ext | |
| if os.path.exists(final_path): | |
| return final_path, "✅ Audio downloaded successfully" | |
| return None, "❌ Audio file not found" | |
| except Exception as e: | |
| traceback.print_exc() | |
| return None, f"❌ Download error: {str(e)}" | |
| # ✅ Transcribe audio using Whisper | |
| def transcribe_audio(path): | |
| if not WHISPER_AVAILABLE: | |
| return "❌ Whisper not available. Please install openai-whisper." | |
| try: | |
| model = whisper.load_model("tiny") | |
| result = model.transcribe(path) | |
| return result["text"] | |
| except Exception as e: | |
| traceback.print_exc() | |
| return f"❌ Transcription failed: {str(e)}" | |
| # ✅ Extract stock-related information | |
| def extract_stock_info(text): | |
| try: | |
| companies = re.findall(r'\b[A-Z][a-z]+(?: [A-Z][a-z]+)*\b', text) | |
| symbols = re.findall(r'\b[A-Z]{2,5}\b', text) | |
| prices = re.findall(r'\$\d+(?:\.\d{1,2})?', text) | |
| actions = re.findall(r'\b(buy|sell|hold|target|bullish|bearish|stop loss)\b', text, re.IGNORECASE) | |
| result = "=== STOCK RECOMMENDATION ANALYSIS ===\n\n" | |
| if companies: | |
| result += f"🏢 Companies Mentioned: {', '.join(set(companies[:10]))}\n" | |
| if symbols: | |
| result += f"🔠 Symbols: {', '.join(set(symbols[:10]))}\n" | |
| if prices: | |
| result += f"💲 Prices: {', '.join(set(prices[:10]))}\n" | |
| if actions: | |
| result += f"📊 Actions: {', '.join(set(actions[:10]))}\n" | |
| recommendations = [] | |
| for line in text.split("."): | |
| if any(word in line.lower() for word in ['buy', 'sell', 'target', 'hold']): | |
| recommendations.append(line.strip()) | |
| if recommendations: | |
| result += "\n🎯 Potential Recommendations:\n" | |
| for r in recommendations[:5]: | |
| result += f"• {r}\n" | |
| if not any([companies, symbols, prices, actions]): | |
| result += "\n⚠️ No stock-related insights detected." | |
| return result | |
| except Exception as e: | |
| return f"❌ Stock info extraction failed: {str(e)}" | |
| # ✅ Save uploaded cookies.txt | |
| def save_cookies(file): | |
| if file is None: | |
| return None | |
| temp_path = tempfile.mktemp(suffix=".txt") | |
| try: | |
| if hasattr(file, "read"): | |
| with open(temp_path, "wb") as f: | |
| f.write(file.read()) | |
| else: | |
| shutil.copy(file, temp_path) | |
| return temp_path | |
| except Exception as e: | |
| print(f"❌ Failed to handle cookies.txt: {e}") | |
| return None | |
| # ✅ YouTube pipeline | |
| def run_pipeline_audio(audio_file): | |
| try: | |
| if not WHISPER_AVAILABLE: | |
| return "❌ Whisper is not installed. Run: pip install openai-whisper", "" | |
| if audio_file is None: | |
| return "❌ No audio file uploaded", "" | |
| # Handle both file-like and NamedString (path string) | |
| temp_audio_path = tempfile.mktemp(suffix=os.path.splitext(str(audio_file))[-1]) | |
| if hasattr(audio_file, "read"): | |
| with open(temp_audio_path, "wb") as f: | |
| f.write(audio_file.read()) | |
| else: | |
| shutil.copy(str(audio_file), temp_audio_path) | |
| transcript = transcribe_audio(temp_audio_path) | |
| if transcript.startswith("❌"): | |
| return transcript, "" | |
| stock_info = extract_stock_info(transcript) | |
| return "✅ Complete", stock_info | |
| except Exception as e: | |
| tb = traceback.format_exc() | |
| print(tb) | |
| return f"❌ Unhandled Error:\n{tb}", "" | |
| # ✅ Audio file upload pipeline | |
| def run_pipeline_audio(audio_file): | |
| try: | |
| if not WHISPER_AVAILABLE: | |
| return "❌ Whisper is not installed. Run: pip install openai-whisper", "" | |
| if audio_file is None: | |
| return "❌ No audio file uploaded", "" | |
| temp_audio_path = tempfile.mktemp(suffix=os.path.splitext(audio_file.name)[-1]) | |
| with open(temp_audio_path, "wb") as f: | |
| f.write(audio_file.read()) | |
| transcript = transcribe_audio(temp_audio_path) | |
| if transcript.startswith("❌"): | |
| return transcript, "" | |
| stock_info = extract_stock_info(transcript) | |
| return "✅ Complete", stock_info | |
| except Exception as e: | |
| tb = traceback.format_exc() | |
| print(tb) | |
| return f"❌ Unhandled Error:\n{tb}", "" | |
| # ✅ Gradio UI | |
| with gr.Blocks(title="Stock Insights from YouTube or Audio") as demo: | |
| gr.Markdown(""" | |
| # 🎧 Extract Stock Recommendations from YouTube or Uploaded Audio | |
| Upload a YouTube URL or an audio file. We'll transcribe it and extract stock-related insights! | |
| """) | |
| with gr.Tab("📺 From YouTube Video"): | |
| with gr.Row(): | |
| url_input = gr.Textbox(label="🎥 YouTube Video URL") | |
| cookie_input = gr.File(label="cookies.txt (optional)", file_types=[".txt"]) | |
| yt_run_btn = gr.Button("🚀 Extract from YouTube") | |
| yt_status = gr.Textbox(label="Status") | |
| yt_result = gr.Textbox(label="Stock Info", lines=12) | |
| yt_run_btn.click(fn=run_pipeline, inputs=[url_input, cookie_input], outputs=[yt_status, yt_result]) | |
| with gr.Tab("🎵 From Uploaded Audio"): | |
| audio_input = gr.File(label="Upload Audio File", file_types=[".mp3", ".wav", ".m4a", ".webm"]) | |
| audio_run_btn = gr.Button("🚀 Extract from Audio") | |
| audio_status = gr.Textbox(label="Status") | |
| audio_result = gr.Textbox(label="Stock Info", lines=12) | |
| audio_run_btn.click(fn=run_pipeline_audio, inputs=[audio_input], outputs=[audio_status, audio_result]) | |
| if __name__ == "__main__": | |
| demo.launch(debug=True) | |