utils.py

import errno
import os
from bs4 import BeautifulSoup
import requests
from requests.exceptions import RequestException, Timeout, TooManyRedirects
import pandas as pd
import urllib
from yt_dlp import YoutubeDL
from yt_dlp.utils import DownloadError
import cv2
import numpy as np
import base64
from typing import Dict, List, Optional



def download_file(file_name: str) -> str:
    """
    Download a file from a given task ID and save it to the local filesystem.

    Args:
        file_name (str): The name of the file to be saved.

    Returns:
        str: A message indicating the result of the download operation.
    """
    task_id = file_name.split(".")[0].split("/")[-1]
    url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"

    try:
        # Check if file_name already exists in the folder
        if os.path.exists(file_name):
            return f"File {file_name} already exists. No need to download again."

        # Download the file
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()  # Raises an HTTPError for bad responses (4XX, 5XX)

            try:
                with open(file_name, 'wb') as file:
                    file.write(response.content)
                return f"Success downloading {file_name}"

            except IOError as e:
                if e.errno == errno.ENOSPC:
                    return f"Failed to save file {file_name}. No space left on the device."
                else:
                    return f"Failed to save file {file_name}. Error: {e}"

        except Timeout:
            return f"Request timed out while trying to download file {file_name}."
        except TooManyRedirects:
            return f"Too many redirects while trying to download file {file_name}."
        except RequestException as e:
            return f"Failed to download file {file_name}. Error: {e}"

    except Exception as e:
        return f"An unexpected error occurred: {e}"
    
def read_file(file_name: str) -> str:
    """
    Read the content of a file.

    Args:
        file_name (str): The name of the file to be read.

    Returns:
        str: The content of the file if successful, or an error message if failed.
    """
    try:
        # Check if the file exists
        if not os.path.exists(file_name):
            return f"File not found: {file_name}"
        
        # Check file extension
        extension = os.path.splitext(file_name)[1].lower()
        if extension not in ['.csv', '.xlsx', '.py', '.txt', '.md',]:
            return "Unsupported file format. Please provide a file with one of those format: csv, xlsx, py, txt, md."
        
        # Read the file using pandas
        try:
            if extension in ['.txt', '.py', '.md']:
                with open(file_name, 'r', encoding='utf-8') as file:
                    file_content = file.read()
            if extension == '.csv':
                file_content = pd.read_csv(file_name).to_string(index=False)
            elif extension == '.xlsx':
                file_content = pd.read_excel(file_name).to_string(index=False)

            return file_content

        except pd.errors.EmptyDataError:
            return "The file is empty."
        except pd.errors.ParserError:
            return "Error parsing the file. The file might be corrupted."
        except Exception as e:
            return f"Error reading file: {e}"

    except PermissionError:
        return f"Permission denied: Unable to access the file at {file_name}."
    except Exception as e:
        return f"An unexpected error occurred: {e}"
    
def sum_pandas_df_cols(df: pd.DataFrame, cols: list):
    return df[cols].sum().sum()
        
def download_yt_video(url: str) -> str:
    """
    Download a YouTube video based on its URL.

    Args:
        url (str): The YouTube URL.

    Returns:
        str: The name of the file downloaded or an error message.
    """
    # Options to download best video ≤ 480p + audio or fallback
    ydl_opts = {
        'format': 'bestvideo[height=480]/bestvideo[height<=480]+bestaudio/best[height<=480]',
        'outtmpl': '%(id)s.%(ext)s',
        'quiet': True,
        'no_warnings': True,
    }

    try:
        # Extract video ID safely from URL
        if 'v=' in url:
            video_id = url.split('v=')[-1].split('&')[0]
        else:
            # fallback to URL itself if no standard param found
            video_id = url.split('/')[-1]

        # Check for existing files with common extensions before downloading
        for ext in ('mp4', 'webm', 'mkv', 'flv'):
            file_path = f'{video_id}.{ext}'
            if os.path.exists(file_path):
                return file_path

        with YoutubeDL(ydl_opts) as ydl:
            info_dict = ydl.extract_info(url, download=True)
            file_name = ydl.prepare_filename(info_dict)
            print(f"Downloaded video: {file_name}")
            return file_name

    except DownloadError as de:
        return f"Download error: {de}"
    except Exception as e:
        return f"Unexpected error: {e}"
        
def extract_frames(video_path: str, interval: int=2, output_folder: str='extracted_frames') -> List[str]:
    """
    Extract frames from a .webm video and save them to a specified folder.
    If the frame is already existing, do not save it again but include it into the list of frames paths.

    Args:
        video_path (str): The path where the video is stored.
        interval (int): The time interval (in seconds) between 2 frames extraction in the video.
        output_folder (str): The folder where the extracted frames will be saved.

    Returns:
        List[str]: A list of paths to the saved frames (including the ones already existing).
    """
    # Create the output folder if it doesn't exist
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    # Open the video file
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise ValueError("Could not open video file")

    # Get video properties
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = frame_count / fps

    frame_paths = []

    # Calculate the frame interval based on the time interval
    frame_interval = int(interval * fps)

    current_frame = 0
    while current_frame < frame_count:
        # Set the current frame position
        cap.set(cv2.CAP_PROP_POS_FRAMES, current_frame)

        # Read the frame
        ret, frame = cap.read()
        if not ret:
            break

        # Generate the output filename
        frame_time = current_frame / fps
        frame_filename = os.path.join(output_folder, f"frame_{frame_time:.2f}.jpg")

        # Check if the frame already exists
        if not os.path.exists(frame_filename):
            # Save the frame
            cv2.imwrite(frame_filename, frame)
            print(f"{frame_filename} saved")
        else:
            print(f"{frame_filename} already existing")

        # Add the frame path to the list
        frame_paths.append(frame_filename)

        # Move to the next frame interval
        current_frame += frame_interval

    # Release the video capture object
    cap.release()

    return frame_paths

def encode_image(image_path):
    """Encode the image to base64."""
    try:
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    except FileNotFoundError:
        print(f"Error: The file {image_path} was not found.")
        return None
    except Exception as e:  # Added general exception handling
        print(f"Error: {e}")
        return None
    
def generate_prompt_for_video_frame_analysis(client, video_question: str, model="mistral-large-latest") -> str:
    chat_response = client.chat.complete(
    model=model,                   
    messages=
        [ 
            {
                "role": "system",
                "content": [
                    {"type": "text", "text":
                        """
                        You are a prompt engineer AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a prompt that can be provided as a user text input to a Large Language Model. DO NOT use any formating like **bold** or "quotation marks" to enclose YOUR FINAL ANSWER.
                        """
                    }
                    
                ]

            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": 
                        f"""
                        To answer the question about the video, I'm going to extract each frame and send them one by one to an independent Large Language Model that has the ability to analyze an image. What prompt should be given to the model so that it can answer the original question? The prompt should be as faithful as possible to the original question, but adapted to an image rather than a video.\n
                        \n
                        <original_question>\n
                        {video_question}\n
                        </original_question>\n
                        """
                    }, 
                ],  
            },
        ]
    )
    return chat_response.choices[0].message.content.split('FINAL ANSWER: ')[-1]

def analyze_frame(client, question: str, base64_image: str, model="pixtral-12b-2409") -> str:
    chat_response = client.chat.complete(
        model=model,                   # Spécification du modèle à utiliser : pixtral-12b-2409 ou pixtral-large-latest
        messages=[                   
            {
                "role": "system", 
                "content": [
                    {
                        "type": "text",
                        "text": """
                            You are an image analyst AI assistant. I will ask you a question. Examine EVERY detail of the image, report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER].\n 
                            \n
                            If the question is about giving a number, YOUR FINAL ANSWER should only be the number asked.\n 
                            If the question is about giving a word or a group of words, YOUR FINAL ANSWER should only be the word or group of words.\n 
                            \n
                            Do not use any formating like **bold** or "quotation marks" to enclose YOUR FINAL ANSWER.
                        """
                    }
                ]
            },
            {
                "role": "user", 
                "content": [
                    {
                        "type": "text", 
                        "text": question
                    }, 
                    {
                        "type": "image_url", 
                        "image_url": f"data:image/jpeg;base64,{base64_image}"
                    }
                ],  
            },
        ]
    )
    return chat_response.choices[0].message.content.split('FINAL ANSWER: ')[-1]

def get_response_from_frames_analysis(client, video_question: str, frames_answers: list, model="mistral-small-latest") -> str:
    chat_response = client.chat.complete(
        model=model,                   
        messages=[ 
            {
                "role": "system",
                "content": [
                    {"type": "text", "text":
                        """
                        You are a general AI assistant. I will ask you a question about a video and provide you with a list of possible answers. Since I can't analyze the video directly, I've broken it down into frames. The list contains the answers to the question I asked for each frame of the video. Your aim is to find the most likely answer to the question for the whole video. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. Do not use any formating like **bold** or "quotation marks" to enclose YOUR FINAL ANSWER.
                        """
                    }
                    
                ]
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": 
                        f"""
                        {video_question}\n
                        \n
                        <List_of_answers_for_each_frame>\n
                        {frames_answers}\n
                        </List_of_answers_for_each_frame>\n
                        """
                    }, 
                ],  
            },
        ]
    )
    return chat_response.choices[0].message.content.split('FINAL ANSWER: ')[-1]


## Not possible to install PyTorch 2.6 on my config, which is requiered to use CLIP Model and Processor :
# MacBook Pro 2017 ; MacOS 13.7.6 ; 2,3 GHz Intel Core i5 double cœur ; Intel Iris Plus Graphics 640 1536 Mo ; 8 Go 2133 MHz LPDDR3

# import torch
# from transformers import CLIPProcessor, CLIPModel
# from sklearn.cluster import DBSCAN

# def embed_frame(img, model, processor, device='gpu'):
#     inputs = processor(images=img, return_tensors="pt").to(device)
#     with torch.no_grad():
#         feat = model.get_image_features(**inputs)
#     return feat.cpu().numpy().squeeze()

# def count_species(embeddings, eps=0.5, min_samples=2):
#     km = DBSCAN(eps=eps, min_samples=min_samples)
#     labels = km.fit_predict(embeddings)
#     return len(set(l for l in labels if l != -1))

# device = 'cpu'
# model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to(device)
# processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# embedded_image = embed_frame(img=test_image, model=model)

# nb_species = count_species(embedded_image)

def transcript_audio_file(client, file_path: str) -> str:
    with open(file_path, "rb") as file:

        # Create a transcription of the audio file

        transcription = client.audio.transcriptions.create(
            file=file, # Required audio file
            model="whisper-large-v3-turbo", # Required model to use for transcription
            language="en",  # Optional
        )
    return transcription

def search_wikipedia(query: str, lang_tag: str) -> Optional[str]:
    search_url = f"https://{lang_tag}.wikipedia.org/w/api.php?action=query&list=search&srsearch={query}&format=json"
    response = requests.get(search_url, timeout=10)
    response.raise_for_status()
    data = response.json()

    search_results = data.get('query', {}).get('search', [])
    title = search_results[0]['title'] if search_results else None

    if not title:
        return "No relevant Wikipedia page found."
    return title.replace(' ', '_')

def fetch_page_content(url: str) -> Optional[str]:
    page_resp = requests.get(url)
    if page_resp.status_code != 200:
        return None

    content_soup = BeautifulSoup(page_resp.text, 'html.parser')
    content_div = content_soup.find("div", id="mw-content-text")
    return content_div.get_text(separator="\n", strip=True) if content_div else None


def get_history_versions(page_title: str, lang_tag: str) -> List[Dict[str, str]]:
    history_url = f"https://{lang_tag}.wikipedia.org/w/index.php?title={page_title}&action=history&limit=100"
    history_resp = requests.get(history_url)
    if history_resp.status_code != 200:
        return []

    history_soup = BeautifulSoup(history_resp.text, 'html.parser')

    history_items = history_soup.find_all("a", class_="mw-changeslist-date")
    if not history_items:
        return []

    versions = []
    for item in history_items:
        if item and 'oldid=' in item['href']:
            versions.append({
                "id": item['href'].split('oldid=')[-1],
                "date": item.get_text()
            })
    return versions


def select_historical_version(client, versions: List[Dict[str, str]], date: str) -> Optional[str]:
    formatted_versions = "\n".join([f"{v['date']} -> {v['id']}" for v in versions])
    prompt = f"""
        You are an AI assistant. I am trying to retrieve the most relevant version of a Wikipedia page for the date described as: "{date}". 

        Here is a list of available version timestamps and their IDs:
        {formatted_versions}

        Which ID best matches the given date? Return ONLY the ID.
        """

    resp = client.chat.complete(
        model="mistral-small-latest",
        messages=[
            {"role": "user", "content": prompt}
        ]
    )

    selected_id = resp.choices[0].message.content.strip()
    return selected_id