Upload helper.py

helper.py

@@ -0,0 +1,298 @@
+# -*- coding: utf-8 -*-
+"""Helper.ipynb
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1LWss_gahHvpiSsp7PsZRKTEsdRttjuAq
+"""
+
+import asyncio
+import json
+import os
+import subprocess
+import urllib
+from datetime import datetime
+from typing import Dict, List, Any, Optional
+import requests
+import re
+from bs4 import BeautifulSoup
+from gtts import gTTS
+#from logger.app_logger import app_logger
+
+# logger/app_logger.py
+import logging
+
+# Create a logger instance
+app_logger = logging.getLogger(__name__)
+
+# Set the logging level (e.g., DEBUG, INFO, WARNING, ERROR, CRITICAL)
+app_logger.setLevel(logging.DEBUG)
+
+# Create a handler (e.g., to write logs to a file or the console)
+handler = logging.StreamHandler() # Outputs logs to the console
+
+# Create a formatter to specify the log message format
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+handler.setFormatter(formatter)
+
+# Add the handler to the logger
+app_logger.addHandler(handler)
+
+# Now you can use the logger in your other modules
+# Example:
+# app_logger.info("This is an informational message.")
+
+!pip install gTTS
+
+
+
+import os
+
+    # Create the 'logger' directory if it doesn't exist
+if not os.path.exists('logger'):
+  os.makedirs('logger')
+
+    # Create an empty 'app_logger.py' file if it doesn't exist
+if not os.path.exists('logger/app_logger.py'):
+  with open('logger/app_logger.py', 'w') as f:
+    pass  # Leave the file empty for now
+
+class ChatBot:
+    """
+    A chatbot class that interacts with a local Llama model using Ollama.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the ChatBot instance with a conversation history."""
+        self.history: List[Dict[str, str]] = [{"role": "system", "content": "You are a helpful assistant."}]
+        app_logger.log_info("ChatBot instance initialized", level="INFO")
+
+    def generate_response(self, prompt: str) -> str:
+        """
+        Generate a response from the chatbot based on the user's prompt.
+
+        Args:
+            prompt (str): The input message from the user.
+
+        Returns:
+            str: The chatbot's response to the provided prompt.
+        """
+        self.history.append({"role": "user", "content": prompt})
+        # app_logger.log_info(f"User prompt added to history: {prompt}", level="INFO")
+        app_logger.log_info("User prompt added to history", level="INFO")
+
+        # Convert chat history into a string for subprocess input
+        conversation = "\n".join(f"{msg['role']}: {msg['content']}" for msg in self.history)
+
+        try:
+            # Run the Llama model using Ollama
+            completion = subprocess.run(
+                ["ollama", "run", "llama3.2:latest"],
+                input=conversation,
+                capture_output=True,
+                text=True,
+            )
+
+            if completion.returncode != 0:
+                app_logger.log_error(f"Error running subprocess: {completion.stderr}")
+                return "I'm sorry, I encountered an issue processing your request."
+
+            response = completion.stdout.strip()
+            self.history.append({"role": "assistant", "content": response})
+            # app_logger.log_info(f"Assistant response generated: {response}", level="INFO")
+            app_logger.log_info("Assistant response generated", level="INFO")
+
+            return response
+
+        except Exception as e:
+            app_logger.log_error(f"Error sending query to the model: {e}")
+            return "I'm sorry, an error occurred while processing your request."
+
+    async def rate_body_of_article(self, article_title: str, article_content: str) -> str:
+        """
+        Rate the quality of an article's content based on its title.
+
+        Args:
+            article_title (str): The title of the article.
+            article_content (str): The full content of the article.
+
+        Returns:
+            str: A rating between 1 and 5 based on relevance and quality.
+        """
+        prompt = f"""
+        Given the following article title and content, provide a rating between 1 and 5
+        based on how well the content aligns with the title and its overall quality.
+
+        - **Article Title**: {article_title}
+        - **Article Content**: {article_content[:1000]}  # Limit to first 1000 chars
+
+        **Instructions:**
+        - The rating should be a whole number between 1 and 5.
+        - Base your score on accuracy, clarity, and relevance.
+        - Only return a single numeric value (1-5) with no extra text.
+
+        **Example Output:**
+        `4` or `2` or `3.5` or `1.5`
+        """
+
+        try:
+            # Run the Llama model using Ollama
+            completion = subprocess.run(
+                ["ollama", "run", "llama3.2:latest"],
+                input=prompt,
+                capture_output=True,
+                text=True,
+            )
+
+            if completion.returncode != 0:
+                app_logger.log_error(f"Error running subprocess: {completion.stderr}")
+                return "Error"
+
+            response = completion.stdout.strip()
+
+            # Validate the rating is within the expected range
+            if response.isdigit() and 1 <= int(response) <= 5:
+                self.history.append({"role": "assistant", "content": response})
+                app_logger.log_info(f"Article rated: {response}", level="INFO")
+                return response
+            else:
+                app_logger.log_warning(f"Invalid rating received: {response}")
+                return "Error"
+
+        except Exception as e:
+            app_logger.log_error(f"Error sending query to the model: {e}")
+            return "Error"
+
+
+# ============================ EXTRACT NEWS BODY ============================
+
+def extract_news_body(news_url: str) -> str:
+    """
+    Extract the full article body from a given news URL.
+
+    Args:
+        news_url (str): The URL of the news article.
+
+    Returns:
+        str: Extracted full article content.
+    """
+    try:
+        headers = {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
+        }
+
+        response = requests.get(news_url, headers=headers, timeout=5)
+        if response.status_code != 200:
+            app_logger.log_error(f"Failed to fetch article: {response.status_code}")
+            return "Failed to fetch article."
+
+        soup = BeautifulSoup(response.text, "html.parser")
+        paragraphs = soup.find_all("p")
+
+        # Extract and return cleaned text
+        article_content = "\n".join([p.text.strip() for p in paragraphs if p.text.strip()])
+        app_logger.log_info(f"Article content extracted from {news_url}", level="INFO")
+        return article_content
+
+    except Exception as e:
+        app_logger.log_error(f"Error extracting article content: {e}")
+        return f"Error extracting article content: {e}"
+
+
+# ============================ ASYNC NEWS SCRAPING ============================
+
+async def invoke_duckduckgo_news_search(query: str, num: int = 5, location: str = "us-en", time_filter: str = "w") -> \
+Dict[str, Any]:
+    """
+    Perform a DuckDuckGo News search, extract news headlines, fetch full content,
+    and rate articles using parallel asynchronous processing.
+
+    Args:
+        query (str): The search query string.
+        num (int): Number of search results to retrieve.
+        location (str): The region code for location-based results (e.g., 'us-en', 'in-en').
+        time_filter (str): Time filter for news ('d' = past day, 'w' = past week, 'm' = past month, 'y' = past year).
+
+    Returns:
+        Dict[str, Any]: A dictionary containing extracted news articles.
+    """
+    app_logger.log_info(f"Starting DuckDuckGo news search for query: {query}", level="INFO")
+
+    duckduckgo_news_url = f"https://duckduckgo.com/html/?q={query.replace(' ', '+')}&kl={location}&df={time_filter}&ia=news"
+    headers = {"User-Agent": "Mozilla/5.0"}
+
+    response = requests.get(duckduckgo_news_url, headers=headers)
+    if response.status_code != 200:
+        app_logger.log_error(f"Failed to fetch news search results: {response.status_code}")
+        return {"status": "error", "message": "Failed to fetch news search results"}
+
+    soup = BeautifulSoup(response.text, "html.parser")
+    search_results = soup.find_all("div", class_="result__body")
+
+    async def process_article(result, index: int) -> Optional[Dict[str, Any]]:
+        """Processes a single article: extracts details, fetches content, and rates it."""
+        try:
+            title_tag = result.find("a", class_="result__a")
+            if not title_tag:
+                app_logger.log_warning(f"Title tag not found for result index {index}")
+                return None
+
+            title = title_tag.text.strip()
+            raw_link = title_tag["href"]
+
+            match = re.search(r"uddg=(https?%3A%2F%2F[^&]+)", raw_link)
+            link = urllib.parse.unquote(match.group(1)) if match else "Unknown Link"
+
+            snippet_tag = result.find("a", class_="result__snippet")
+            summary = snippet_tag.text.strip() if snippet_tag else "No summary available."
+
+            article_content = extract_news_body(link)
+
+            bot = ChatBot()
+            rating = await bot.rate_body_of_article(title, article_content)
+
+            app_logger.log_info(f"Processed article: {title}", level="INFO")
+
+            return {
+                "num": index + 1,
+                "link": link,
+                "title": title,
+                "summary": summary,
+                "body": article_content,
+                "rating": rating
+            }
+
+        except Exception as e:
+            app_logger.log_error(f"Error processing article: {e}")
+            return None
+
+    tasks = [process_article(result, index) for index, result in enumerate(search_results[:num])]
+    extracted_results = await asyncio.gather(*tasks)
+
+    extracted_results = [res for res in extracted_results if res is not None]
+
+    if extracted_results:
+        app_logger.log_info(f"News search completed successfully with {len(extracted_results)} results", level="INFO")
+        return {"status": "success", "results": extracted_results}
+    else:
+        app_logger.log_error("No valid news search results found")
+        return {"status": "error", "message": "No valid news search results found"}
+
+
+# ============================ UTILITY FUNCTIONS ============================
+
+def current_year() -> int:
+    """Returns the current year as an integer."""
+    return datetime.now().year
+
+
+def save_to_audio(text: str) -> None:
+    """Converts text to an audio file using Google Text-to-Speech (gTTS)."""
+    try:
+        tts = gTTS(text=text, lang="en")
+        tts.save("output.mp3")
+        app_logger.log_info("Response converted to audio", level="INFO")
+    except Exception as e:
+        app_logger.log_error(f"Error converting response to audio: {e}")
+