import spaces import gradio as gr import logging import os import tempfile import pandas as pd import requests from bs4 import BeautifulSoup import torch import whisper import subprocess from pydub import AudioSegment import fitz import docx import yt_dlp from functools import lru_cache import gc import time from huggingface_hub import login from unsloth import FastLanguageModel from transformers import AutoTokenizer # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Login to Hugging Face Hub if token is available HUGGINGFACE_TOKEN = os.environ.get('HUGGINGFACE_TOKEN') if HUGGINGFACE_TOKEN: login(token=HUGGINGFACE_TOKEN) class ModelManager: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(ModelManager, cls).__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if not self._initialized: self.tokenizer = None self.model = None self.pipeline = None self.whisper_model = None self._initialized = True self.last_used = time.time() @spaces.GPU() def initialize_llm(self): """Initialize LLM model with Unsloth optimization""" try: MODEL_NAME = "TinyLlama/TinyLlama-1.1B-Chat-v1.0" logger.info("Loading Unsloth-optimized model...") self.model, self.tokenizer = FastLanguageModel.from_pretrained( model_name = MODEL_NAME, max_seq_length = 2048, dtype = torch.float16, load_in_4bit = True, token = HUGGINGFACE_TOKEN, ) # Enable LoRA for better ZeroGPU performance self.model = FastLanguageModel.get_peft_model( self.model, r = 16, target_modules = ["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"], lora_alpha = 16, lora_dropout = 0, bias = "none", use_gradient_checkpointing = True, random_state = 3407, max_seq_length = 2048, ) logger.info("LLM initialized successfully with Unsloth") self.last_used = time.time() return True except Exception as e: logger.error(f"Error initializing LLM: {str(e)}") raise @spaces.GPU() def initialize_whisper(self): """Initialize Whisper model with safety fix""" try: logger.info("Loading Whisper model...") # Load with weights_only=True for security self.whisper_model = whisper.load_model( "tiny", device="cuda" if torch.cuda.is_available() else "cpu", download_root="/tmp/whisper", weights_only=True # Security fix ) logger.info("Whisper model initialized successfully") self.last_used = time.time() return True except Exception as e: logger.error(f"Error initializing Whisper: {str(e)}") raise def check_llm_initialized(self): """Check if LLM is initialized and initialize if needed""" if self.tokenizer is None or self.model is None: logger.info("LLM not initialized, initializing...") self.initialize_llm() self.last_used = time.time() def check_whisper_initialized(self): """Check if Whisper model is initialized and initialize if needed""" if self.whisper_model is None: logger.info("Whisper model not initialized, initializing...") self.initialize_whisper() self.last_used = time.time() def reset_models(self, force=False): """Reset models to free memory if they haven't been used recently""" current_time = time.time() if force or (current_time - self.last_used > 600): try: logger.info("Resetting models to free memory...") if self.model is not None: del self.model if self.tokenizer is not None: del self.tokenizer if self.whisper_model is not None: del self.whisper_model self.tokenizer = None self.model = None self.whisper_model = None if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.synchronize() gc.collect() logger.info("Models reset successfully") except Exception as e: logger.error(f"Error resetting models: {str(e)}") model_manager = ModelManager() @lru_cache(maxsize=32) def download_social_media_video(url): """Download a video from social media.""" ydl_opts = { 'format': 'bestaudio/best', 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192', }], 'outtmpl': '%(id)s.%(ext)s', } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info_dict = ydl.extract_info(url, download=True) audio_file = f"{info_dict['id']}.mp3" logger.info(f"Video downloaded successfully: {audio_file}") return audio_file except Exception as e: logger.error(f"Error downloading video: {str(e)}") raise def convert_video_to_audio(video_file): """Convert a video file to audio using ffmpeg directly.""" try: with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file: output_file = temp_file.name command = [ "ffmpeg", "-i", video_file, "-q:a", "0", "-map", "a", "-vn", output_file, "-y" ] subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) logger.info(f"Video converted to audio: {output_file}") return output_file except Exception as e: logger.error(f"Error converting video: {str(e)}") raise def preprocess_audio(audio_file): """Preprocess the audio file to improve quality.""" try: audio = AudioSegment.from_file(audio_file) audio = audio.apply_gain(-audio.dBFS + (-20)) with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file: audio.export(temp_file.name, format="mp3") logger.info(f"Audio preprocessed: {temp_file.name}") return temp_file.name except Exception as e: logger.error(f"Error preprocessing audio: {str(e)}") raise @spaces.GPU() def transcribe_audio(file): """Transcribe an audio or video file.""" try: model_manager.check_whisper_initialized() if isinstance(file, str) and file.startswith('http'): file_path = download_social_media_video(file) elif isinstance(file, str) and file.lower().endswith(('.mp4', '.avi', '.mov', '.mkv')): file_path = convert_video_to_audio(file) elif file is not None: file_path = preprocess_audio(file.name) else: return "" logger.info(f"Transcribing audio: {file_path}") if not os.path.exists(file_path): raise FileNotFoundError(f"Audio file not found: {file_path}") with torch.inference_mode(): result = model_manager.whisper_model.transcribe(file_path) transcription = result.get("text", "Error in transcription") logger.info(f"Transcription completed: {transcription[:50]}...") try: if os.path.exists(file_path): os.remove(file_path) except Exception as e: logger.warning(f"Could not remove temp file {file_path}: {str(e)}") return transcription except Exception as e: logger.error(f"Error transcribing: {str(e)}") return f"Error processing the file: {str(e)}" @lru_cache(maxsize=32) def read_document(document_path): """Read the content of a document.""" try: if document_path.endswith(".pdf"): doc = fitz.open(document_path) return "\n".join([page.get_text() for page in doc]) elif document_path.endswith(".docx"): doc = docx.Document(document_path) return "\n".join([paragraph.text for paragraph in doc.paragraphs]) elif document_path.endswith((".xlsx", ".xls")): return pd.read_excel(document_path).to_string() elif document_path.endswith(".csv"): return pd.read_csv(document_path).to_string() else: return "Unsupported file type. Please upload a PDF, DOCX, XLSX or CSV document." except Exception as e: logger.error(f"Error reading document: {str(e)}") return f"Error reading document: {str(e)}" @lru_cache(maxsize=32) def read_url(url): """Read the content of a URL.""" if not url or url.strip() == "": return "" try: headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' } response = requests.get(url, headers=headers, timeout=15) response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') for element in soup(["script", "style", "meta", "noscript", "iframe", "header", "footer", "nav"]): element.extract() main_content = soup.find("main") or soup.find("article") or soup.find("div", class_=["content", "main", "article"]) if main_content: text = main_content.get_text(separator='\n', strip=True) else: text = soup.get_text(separator='\n', strip=True) lines = [line.strip() for line in text.split('\n') if line.strip()] text = '\n'.join(lines) return text[:10000] except Exception as e: logger.error(f"Error reading URL: {str(e)}") return f"Error reading URL: {str(e)}" def process_social_content(url): """Process social media content.""" if not url or url.strip() == "": return None try: text_content = read_url(url) try: video_content = transcribe_audio(url) except Exception as e: logger.error(f"Error processing video content: {str(e)}") video_content = None return { "text": text_content, "video": video_content } except Exception as e: logger.error(f"Error processing social content: {str(e)}") return None @spaces.GPU() def generate_news(instructions, facts, size, tone, *args): """Generate a news article based on provided data""" try: if isinstance(size, float): size = int(size) elif not isinstance(size, int): size = 250 model_manager.check_llm_initialized() knowledge_base = { "instructions": instructions or "", "facts": facts or "", "document_content": [], "audio_data": [], "url_content": [], "social_content": [] } num_audios = 5 * 3 num_social_urls = 3 * 3 num_urls = 5 args = list(args) while len(args) < (num_audios + num_social_urls + num_urls + 5): args.append("") audios = args[:num_audios] social_urls = args[num_audios:num_audios+num_social_urls] urls = args[num_audios+num_social_urls:num_audios+num_social_urls+num_urls] documents = args[num_audios+num_social_urls+num_urls:] logger.info("Processing URLs...") for url in urls: if url and isinstance(url, str) and url.strip(): content = read_url(url) if content and not content.startswith("Error"): knowledge_base["url_content"].append(content) logger.info("Processing documents...") for document in documents: if document and hasattr(document, 'name'): content = read_document(document.name) if content and not content.startswith("Error"): knowledge_base["document_content"].append(content) logger.info("Processing audio/video files...") for i in range(0, len(audios), 3): if i+2 < len(audios): audio_file, name, position = audios[i:i+3] if audio_file and hasattr(audio_file, 'name'): knowledge_base["audio_data"].append({ "audio": audio_file, "name": name or "Unknown", "position": position or "Not specified" }) logger.info("Processing social media content...") for i in range(0, len(social_urls), 3): if i+2 < len(social_urls): social_url, social_name, social_context = social_urls[i:i+3] if social_url and isinstance(social_url, str) and social_url.strip(): social_content = process_social_content(social_url) if social_content: knowledge_base["social_content"].append({ "url": social_url, "name": social_name or "Unknown", "context": social_context or "Not specified", "text": social_content.get("text", ""), "video": social_content.get("video", "") }) transcriptions_text = "" raw_transcriptions = "" logger.info("Transcribing audio...") for idx, data in enumerate(knowledge_base["audio_data"]): if data["audio"] is not None: transcription = transcribe_audio(data["audio"]) if transcription and not transcription.startswith("Error"): transcriptions_text += f'"{transcription}" - {data["name"]}, {data["position"]}\n\n' raw_transcriptions += f'[Audio/Video {idx + 1}]: "{transcription}" - {data["name"]}, {data["position"]}\n\n' for idx, data in enumerate(knowledge_base["social_content"]): if data["text"] and not str(data["text"]).startswith("Error"): text_excerpt = data["text"][:500] + "..." if len(data["text"]) > 500 else data["text"] social_text = f'[Social media {idx+1} - text]: "{text_excerpt}" - {data["name"]}, {data["context"]}\n\n' transcriptions_text += social_text raw_transcriptions += social_text if data["video"] and not str(data["video"]).startswith("Error"): video_transcription = f'[Social media {idx+1} - video]: "{data["video"]}" - {data["name"]}, {data["context"]}\n\n' transcriptions_text += video_transcription raw_transcriptions += video_transcription document_summaries = [] for idx, doc in enumerate(knowledge_base["document_content"]): if len(doc) > 1000: doc_excerpt = doc[:1000] + "... [document continues]" else: doc_excerpt = doc document_summaries.append(f"[Document {idx+1}]: {doc_excerpt}") document_content = "\n\n".join(document_summaries) url_summaries = [] for idx, url_content in enumerate(knowledge_base["url_content"]): if len(url_content) > 1000: url_excerpt = url_content[:1000] + "... [content continues]" else: url_excerpt = url_content url_summaries.append(f"[URL {idx+1}]: {url_excerpt}") url_content = "\n\n".join(url_summaries) prompt = f"""[INST] You are a professional news writer. Write a news article based on the following information: Instructions: {knowledge_base["instructions"]} Facts: {knowledge_base["facts"]} Additional content from documents: {document_content} Additional content from URLs: {url_content} Use these transcriptions as direct and indirect quotes: {transcriptions_text} Follow these requirements: - Write a title - Write a 15-word hook that complements the title - Write the body with approximately {size} words - Use a {tone} tone - Answer the 5 Ws (Who, What, When, Where, Why) in the first paragraph - Use at least 80% direct quotes (in quotation marks) - Use proper journalistic style - Do not invent information - Be rigorous with the provided facts [/INST]""" try: logger.info("Generating news article...") max_length = min(len(prompt.split()) + size * 2, 2048) inputs = model_manager.tokenizer( prompt, return_tensors = "pt", padding = True, truncation = True, max_length = 2048, ).to("cuda") outputs = model_manager.model.generate( **inputs, max_new_tokens = size + 100, temperature = 0.7, do_sample = True, pad_token_id = model_manager.tokenizer.eos_token_id, ) generated_text = model_manager.tokenizer.decode(outputs[0], skip_special_tokens = True) if "[/INST]" in generated_text: news_article = generated_text.split("[/INST]")[1].strip() else: prompt_fragment = " ".join(prompt.split()[:50]) if prompt_fragment in generated_text: news_article = generated_text[generated_text.find(prompt_fragment) + len(prompt_fragment):].strip() else: news_article = generated_text logger.info(f"News generation completed: {len(news_article)} chars") except Exception as gen_error: logger.error(f"Error in text generation: {str(gen_error)}") raise return news_article, raw_transcriptions except Exception as e: logger.error(f"Error generating news: {str(e)}") try: model_manager.reset_models(force=True) except Exception as reset_error: logger.error(f"Failed to reset models: {str(reset_error)}") return f"Error generating news: {str(e)}", "Error processing transcriptions." def create_demo(): with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown("# 📰 NewsIA - AI News Generator") gr.Markdown("Create professional news articles from multiple sources.") with gr.Row(): with gr.Column(scale=2): instructions = gr.Textbox( label="News Instructions", placeholder="Enter specific instructions for news generation", lines=2 ) facts = gr.Textbox( label="Key Facts", placeholder="Describe the most important facts to include", lines=4 ) with gr.Row(): size = gr.Slider( label="Approximate Length (words)", minimum=100, maximum=500, value=250, step=50 ) tone = gr.Dropdown( label="News Tone", choices=["serious", "neutral", "funny", "formal", "informal", "urgent"], value="neutral" ) with gr.Column(scale=3): inputs_list = [] inputs_list.extend([instructions, facts, size, tone]) with gr.Tabs(): with gr.TabItem("📝 Documents"): documents = [] for i in range(1, 6): doc = gr.File( label=f"Document {i}", file_types=["pdf", "docx", "xlsx", "csv"], file_count="single" ) documents.append(doc) inputs_list.append(doc) with gr.TabItem("🔊 Audio/Video"): for i in range(1, 6): with gr.Group(): gr.Markdown(f"**Source {i}**") file = gr.File( label=f"Audio/Video {i}", file_types=["audio", "video"] ) with gr.Row(): name = gr.Textbox( label="Name", placeholder="Interviewee name" ) position = gr.Textbox( label="Position/Role", placeholder="Position or role" ) inputs_list.extend([file, name, position]) with gr.TabItem("🌐 URLs"): for i in range(1, 6): url = gr.Textbox( label=f"URL {i}", placeholder="https://..." ) inputs_list.append(url) with gr.TabItem("📱 Social Media"): for i in range(1, 4): with gr.Group(): gr.Markdown(f"**Social Media {i}**") social_url = gr.Textbox( label="URL", placeholder="https://..." ) with gr.Row(): social_name = gr.Textbox( label="Account/Name", placeholder="Account or person name" ) social_context = gr.Textbox( label="Context", placeholder="Relevant context" ) inputs_list.extend([social_url, social_name, social_context]) with gr.Row(): generate_btn = gr.Button("✨ Generate News", variant="primary") reset_btn = gr.Button("🔄 Clear All") with gr.Tabs(): with gr.TabItem("📄 Generated News"): news_output = gr.Textbox( label="News Draft", lines=15, show_copy_button=True ) with gr.TabItem("🎙️ Transcriptions"): transcriptions_output = gr.Textbox( label="Source Transcriptions", lines=10, show_copy_button=True ) generate_btn.click( fn=generate_news, inputs=inputs_list, outputs=[news_output, transcriptions_output] ) def reset_all(): return [None]*len(inputs_list) + ["", ""] reset_btn.click( fn=reset_all, inputs=None, outputs=inputs_list + [news_output, transcriptions_output] ) return demo if __name__ == "__main__": try: model_manager.initialize_whisper() except Exception as e: logger.warning(f"Initial whisper model loading failed: {str(e)}") demo = create_demo() demo.queue(max_size=5) demo.launch( share=True, server_name="0.0.0.0", server_port=7860 )