Upload utils.py

utils.py

@@ -0,0 +1,291 @@
+import requests
+from bs4 import BeautifulSoup
+import pandas as pd
+from typing import List, Dict, Any
+import numpy as np
+from transformers import pipeline
+import urllib.parse
+from sklearn.feature_extraction.text import TfidfVectorizer
+import tldextract
+from deep_translator import GoogleTranslator
+from playsound import playsound
+import soundfile as sf
+from transformers import AutoModel, AutoTokenizer
+def search_news(company_name: str, num_articles: int = 2) -> List[str]:
+    search_url = f"https://www.google.com/search?q={company_name}+news&tbm=nws"
+    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"
+    }
+
+    try:
+        response = requests.get(search_url, headers=headers)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, "html.parser")
+
+        article_links = []
+        for article in soup.select('.SoaBEf'):
+            link_element = article.select_one('a')
+            if link_element and 'href' in link_element.attrs:
+                href = link_element['href']
+                if href.startswith('/url?q='):
+                    url = href.split('/url?q=')[1].split('&')[0]
+                    url = urllib.parse.unquote(url)
+                    article_links.append(url)
+                elif href.startswith('http'):
+                    article_links.append(href)
+
+                if len(article_links) >= num_articles:
+                    break
+
+        return article_links
+    except Exception as e:
+        print(f"Error fetching news articles: {e}")
+        return []
+
+def extract_article_content(url: str) -> Dict[str, Any]:
+    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
+
+    try:
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, "html.parser")
+
+        title = soup.find("h1").get_text().strip() if soup.find("h1") else "No title found"
+
+        content_element = soup.find("article") or soup.find("main") or soup.find("div", class_=["content", "article", "story"])
+        content = " ".join([p.get_text().strip() for p in content_element.find_all("p")]) if content_element else "No content found"
+
+        date_element = soup.find("time")
+        date = date_element["datetime"] if date_element and "datetime" in date_element.attrs else None
+
+        return {
+            'url': url,
+            'title': title,
+            'content': content,
+            'date': date
+        }
+    except Exception as e:
+        print(f"Error extracting content from {url}: {e}")
+        return {
+            'url': url,
+            'title': "Error extracting content",
+            'content': "Error extracting content",
+            'date': None
+        }
+
+def get_company_news(company_name: str) -> List[Dict[str, Any]]:
+    """
+    Fetch exactly 10 news articles for a given company.
+    If fewer than 10 articles are retrieved initially, retry fetching more.
+    """
+    max_articles = 10
+    articles = []
+    retries = 3  # Number of retries to fetch missing articles
+
+    for attempt in range(retries):
+        # Fetch article URLs
+        article_urls = search_news(company_name, num_articles=max_articles - len(articles))
+
+        # Process each URL to extract content
+        for url in article_urls:
+            try:
+                article_data = extract_article_content(url)
+                # Avoid duplicates by checking the URL
+                if article_data['url'] not in [a['url'] for a in articles]:
+                    articles.append(article_data)
+            except Exception as e:
+                print(f"Error extracting from {url}: {e}")
+
+        # Break if we have enough articles
+        if len(articles) >= max_articles:
+            break
+
+    # If still fewer than 10 articles, fill with placeholders
+    while len(articles) < max_articles:
+        articles.append({
+            'url': 'N/A',
+            'title': 'No Title Available',
+            'content': 'No Content Available',
+            'date': None
+        })
+
+    return articles
+def summarize_article(content: str, max_length: int = 50) -> str:
+    summarizer = pipeline("summarization")
+    max_input_length = summarizer.model.config.max_position_embeddings  # Get model's max input length
+
+    # Ensure content does not exceed max input length
+    truncated_content = content[:max_input_length]
+
+    summary = summarizer(truncated_content, max_length=max_length, min_length=0, do_sample=False)
+    return summary[0]['summary_text']
+
+def analyze_sentiment(text: str) -> Dict[str, Any]:
+    """
+    Analyze sentiment of the given text.
+
+    Args:
+        text: The text to analyze.
+
+    Returns:
+        Dictionary containing sentiment category and score.
+    """
+    try:
+        # Initialize sentiment analyzer
+        sentiment_analyzer = pipeline("sentiment-analysis", truncation=True)
+
+        # Truncate text manually to avoid exceeding token limits
+        max_token_limit = 512  # Most transformer models have a 512-token limit
+        words = text.split()
+        if len(words) > max_token_limit:
+            text = ' '.join(words[:max_token_limit])
+
+        # Perform sentiment analysis
+        result = sentiment_analyzer(text)
+
+        # Determine sentiment category based on label and score
+        sentiment_category = "Positive" if result[0]['label'] == "POSITIVE" else "Negative"
+        score = result[0]['score']
+
+        # Add neutral category for borderline cases
+        if 0.4 <= score <= 0.6:
+            sentiment_category = "Neutral"
+
+        return {
+            'sentiment': sentiment_category,
+            'score': score
+        }
+    except Exception as e:
+        print(f"Error in sentiment analysis: {e}")
+        return {
+            'sentiment': "Unknown",
+            'score': 0.0
+        }
+
+def extract_key_topics(text: str, num_topics: int = 5) -> List[str]:
+    if len(text.split()) < 10:
+        return ["Not enough text to extract topics"]
+
+    vectorizer = TfidfVectorizer(stop_words='english', max_features=100)
+    tfidf_matrix = vectorizer.fit_transform([text])
+    feature_names = vectorizer.get_feature_names_out()
+    tfidf_scores = tfidf_matrix.toarray()[0]
+    sorted_indices = np.argsort(tfidf_scores)[::-1]
+    top_topics = [feature_names[idx] for idx in sorted_indices[:num_topics]]
+
+    return top_topics
+
+def perform_comparative_analysis(articles: List[Dict[str, Any]]) -> Dict[str, Any]:
+    sentiment_counts = {
+        'Positive': len([a for a in articles if a['sentiment']['sentiment'] == 'Positive']),
+        'Neutral': len([a for a in articles if a['sentiment']['sentiment'] == 'Neutral']),
+        'Negative': len([a for a in articles if a['sentiment']['sentiment'] == 'Negative'])
+    }
+
+    all_topics = [topic for article in articles for topic in article['topics']]
+    topic_frequency = {}
+    for topic in all_topics:
+        topic_frequency[topic] = topic_frequency.get(topic, 0) + 1
+
+    common_topics = sorted(topic_frequency.items(), key=lambda x: x[1], reverse=True)
+
+    sentiment_by_source = {}
+    for article in articles:
+        source = extract_source_from_url(article['url'])
+        if source not in sentiment_by_source:
+            sentiment_by_source[source] = []
+        sentiment_by_source[source].append(article['sentiment']['sentiment'])
+
+    return {
+        'sentiment_distribution': sentiment_counts,
+        'common_topics': common_topics[:10],
+        'sentiment_by_source': sentiment_by_source
+    }
+
+def extract_source_from_url(url: str) -> str:
+    extracted_info = tldextract.extract(url)
+    return extracted_info.domain
+
+from typing import List, Dict, Any
+from transformers import pipeline
+
+def get_combined_summary(articles, max_length: int = 100) -> str:
+    """
+    Generate a combined summary from multiple news articles.
+
+    Args:
+        articles: List of article dictionaries containing content
+        max_length: Maximum length of the final summary
+
+    Returns:
+        A comprehensive summary combining insights from all articles
+    """
+    # Combine all article contents with titles as context
+    combined_content = ""
+    for article in articles:
+        # Use .get() with default values to handle missing keys
+        title = article.get('title', 'No Title')
+        content = article.get('content', 'Content not available')
+        combined_content += f"Article: {title}\n{content}\n\n"
+
+    # Initialize the summarizer
+    summarizer = pipeline("summarization")
+
+    # Handle token limit constraints
+    max_input_length = summarizer.model.config.max_position_embeddings
+    truncated_content = combined_content[:max_input_length]
+
+    # Generate the combined summary
+    summary = summarizer(truncated_content, max_length=max_length, min_length=30, do_sample=False)
+
+    # Handle different return formats from the pipeline
+    if isinstance(summary, list):
+        return summary[0]['summary_text']
+    else:
+        return summary['summary_text']
+
+def generate_hindi_summary(combined_summary: str) -> str:
+    """
+    Translate the combined summary to Hindi using deep-translator.
+
+    Args:
+        combined_summary: The English combined summary
+
+    Returns:
+        The Hindi translation of the combined summary
+    """
+    try:
+        translator = GoogleTranslator(source='auto', target='hi')
+        hindi_summary = translator.translate(text=combined_summary)
+        return hindi_summary
+    except Exception as e:
+        print(f"Error in translation: {e}")
+        return "Translation failed"
+def generate_hindi_speech(hindi_summary: str):
+    """
+    Convert Hindi summary to speech using AI4Bharat's VITS-Rasa-13 model and play it
+
+    Args:
+        hindi_summary: Hindi text summary to synthesize (max 500 characters)
+    """
+    try:
+        # Load pre-trained model (requires CUDA-enabled GPU)
+        model = AutoModel.from_pretrained("ai4bharat/vits_rasa_13", trust_remote_code=True).to("cuda")
+        tokenizer = AutoTokenizer.from_pretrained("ai4bharat/vits_rasa_13", trust_remote_code=True)
+
+        # Process text and generate speech
+        inputs = tokenizer(text=hindi_summary, return_tensors="pt").to("cuda")
+
+        # Use default Indian voice profile (speaker_id=16 for male, 17 for female)
+        outputs = model(inputs['input_ids'], speaker_id=16, emotion_id=0)
+
+        # Convert to numpy array and save as temporary file
+        audio_data = outputs.waveform.squeeze().cpu().numpy()
+        sf.write("temp_hindi_speech.wav", audio_data, model.config.sampling_rate)
+
+        # Play the audio using playsound
+        playsound("temp_hindi_speech.wav")
+
+    except Exception as e:
+        print(f"Error in speech generation or playback: {e}")
+