helper.py

# -*- 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}")