data_handling.py

# Import necessary libraries for data handling and preprocessing
# os: For file system operations like path normalization and existence checks
# partition: Unstructured library function that automatically detects file type and extracts content
# elements_to_json: Converts unstructured elements into JSON format for processing
# File: Mutagen library for reading audio/video metadata (duration, format, etc.)
# VideoFileClip: MoviePy class for loading and inspecting video files
import os
from unstructured.partition.auto import partition  # Detects the file type
from unstructured.staging.base import elements_to_json  # For converting/preparing the content into json
# import speech_recognition as sr  # Commented out - speech recognition library (not currently used)
from mutagen import File  # For audio/video metadata extraction
# import pytesseract as pt  # Commented out - OCR library for image text extraction (not currently used)
# from PIL import Image  # Commented out - image processing library (not currently used)
# import moviepy.editor as mpe  # Commented out - alternative MoviePy import (not currently used)
from moviepy.editor import VideoFileClip  # For video file handling and duration checking
# import numpy as np  # Commented out - numerical computing (not currently used)
# import tempfile  # Commented out - temporary file management (not currently used)
# import shutil  # Commented out - file operations (not currently used)



# Data Extraction
# def mp3_to_wav(mp3_file, wav_file):
    # mp3 = AudioSegment.from_mp3(mp3_file)
    # mp3.export(wav_file, format= "wav")

# def video_audio_text(video, interval_sec= 3):
    # temp_dir = tempfile.mkdtemp()

    # try:

    #     video_text_combined = [] # list for collecting the text from all(interval 3sec) frames of video

    #     video = mpe.VideoFileClip(video)
    #     # making the folder for the frames
    #     video_folder = os.path.splitext(os.path.basename(video.filename))[0] + "_frames"
    #     os.makedirs(video_folder, exist_ok=True)
    #     print(f"Extracting the frames every {interval_sec} seconds to folder: {video_folder}")

    #     timestamps = np.arange(0, video.duration, interval_sec)

    #     for i in timestamps:
    #         frame_path = os.path.join(video_folder, f"frame{int(i)}.jpg")
    #         video.save_frame(frame_path, i)
    #         print(f"Saved frame at {i}, seconds: {frame_path}")

    #         # Applying OCR on video frames
    #         frame = Image.open(frame_path)
    #         video_text = pt.image_to_string(frame)
    #         video_text_combined.append(video_text)
    #         # print(f"OCR Text at {i} seconds:\n {video_text}\n{"-"*30}")
    #         # return video_text
    #     video_text = "\n".join(video_text_combined)    


    #     # now processing the audio of video
    #     video_audio = f"{video.filename}_audio.wav"
    #     video.audio.write_audiofile(video_audio) #writing the audio file

    #     recognizer = sr.Recognizer()
        
    #     with sr.AudioFile(video_audio) as source:
    #         video_audio_record = recognizer.record(source)

    #     try:
    #         video_audio_text = recognizer.recognize_google(video_audio_record)
    #         # print("Extracted Text in audio of the video:\n", video_audio_text)
    #     except Exception as e:
    #         print(f"Speech Recognition Error: {e}")

    #     try:
    #         if not video_text:
    #             print("Text in video couldn't processed\n")

    #         if not video_audio_text:
    #             print("Text in Audio of video couldn't processed\n")
            
    #         return video_text, video_audio_text
        
    #     except Exception as e:
    #         print(f"Error in processing Video and Audio: {e}\n")
    # finally:
    #     shutil.rmtree(temp_dir)
        
# def audio_text(audio):
    # input_file = audio
    # try:
    #     if audio.lower().endswith('.mp3'): # renaming the wav
    #         wav_file = audio.rsplit('.', 1)[0] + '.wav'
    #         mp3_to_wav(audio, wav_file) # calling mp3_to_wav function to convert the mp3 into wav
    #         input_file = wav_file


    #     if audio.lower().endswith(('.wav', '.flac', '.aiff')):
    #         recognizer = sr.Recognizer()
    #         with sr.AudioFile(input_file) as source:
    #             audio_wav = recognizer.record(source)
    #         try:
    #             # Using the Google Speech Recognition
    #             audio_text = recognizer.recognize_google(audio_wav)
    #             # print("Audio Transcribbed\n", audio_text)
    #             return audio_text
    #         except sr.UnknownValueError: 
    #             print("Speech Recognition could not understand the audio!")
    #             return ""
    #         except sr.RequestError as e:
    #             print(f"Could not request results; {e}")
    #             return ""
    # except Exception as e:
    #     print(f"Error loading file: {e}")

# def image_text(image):
    # image = os.path.normpath(image)
    # # Check if the file exists
    # if not os.path.exists(image):
    #     print(f"File does not exist: {image}")
    #     return ""    
          
    # try:
    #     # if the result list (from partition), convert to json and show both
    #     if image.lower().endswith(('.jpg', '.jpeg', '.png', '.webp', '.heif')):
    #         img = Image.open(image)
    #         img_text = pt.image_to_string(img)
    #         # print("\n", img_text)
    #         return img_text
    # except Exception as e:
    #     print(f"Error loading {image}\n {e}")

# Preprocessing (Validation) Functions
# These functions validate and prepare media files before sending to AI models
# They check file existence, format support, and enforce size/duration limits

def audio(audio):
    """
    Validate and preprocess audio files before AI analysis.

    Checks:
    1. File existence
    2. Format support using Mutagen library
    3. Duration limit (1 minute max for free tier)

    Args:
        audio (str): Path to audio file

    Returns:
        str: Validated file path if successful, error message if failed
    """
    # Normalize path to handle different OS path separators
    audio = os.path.normpath(audio)

    # Check if file exists
    if not os.path.exists(audio):
        err = f"File doesn't exist, recheck the given file path: {audio}!\n"
        return err

    # Validate audio format and check duration limits
    try:
        # Use Mutagen to read audio metadata
        audio_file = File(audio)
        if audio_file is None:
            # File exists but Mutagen can't read it (unsupported format or corrupted)
            err = f"Audio: {audio} is not supported or corrupted"
            return err

        # Get audio duration in seconds
        length = audio_file.info.length

        # Enforce 1-minute limit for free tier
        if length > 60:  # 60 seconds = 1 minute
            in_minute = length / 60
            limit_message = f"Audio length is {in_minute:.2f} min which exceeds 1 minute.\nFor longer audio, think once to subscribe our monthly premium!"
            return limit_message
        else:
            # File is valid and within limits
            return audio

    except Exception as e:
        # Handle any errors during file processing
        err = f"Audio: {audio} is not supported\n {e}"
        return err

def image(image):
    """
    Validate and preprocess image files before AI analysis.

    Checks:
    1. File existence
    2. Basic format validation (handled by AI model)

    Args:
        image (str): Path to image file

    Returns:
        str: Validated file path if successful, error message if failed
    """
    # Normalize path for cross-platform compatibility
    image = os.path.normpath(image)

    # Check if file exists
    if not os.path.exists(image):
        err = f"File doesn't exist, recheck the given file path: {image}!\n"
        return err
    else:
        # File exists - further validation handled by AI model
        return image

def video(video):
    """
    Validate and preprocess video files before AI analysis.

    Checks:
    1. File existence (commented out - handled elsewhere)
    2. Duration limit (30 seconds max for free tier)
    3. Format support using MoviePy

    Args:
        video (str): Path to video file

    Returns:
        str: Validated file path if successful, error message if failed
    """
    # Normalize path for cross-platform compatibility
    video = os.path.normpath(video)

    # File existence check commented out - handled in calling function
    # if not os.path.exists(video):
    #     err = f"File doesn't exist, recheck the given file path: {video}!\n"
    #     return err

    # Validate video format and check duration limits
    try:
        # Use MoviePy to load video and check duration
        with VideoFileClip(video) as clip:
            # Enforce 30-second limit for free tier
            if clip.duration > 30:  # 30 seconds max
                duration = clip.duration
                duration_in_minutes = duration / 60
                limit_message = f"Video length is {duration_in_minutes:.2f} min which exceeds 30 seconds.\nFor longer video, think once to subscribe our monthly premium!"
                return limit_message
            else:
                # Video is valid and within limits
                return video

    except Exception as e:
        # Handle unsupported formats or corrupted files
        err = f"Video: {video} is not supported\n {e}"
        return err


def unstructured_doc_extraction(doc):
    """
    Extract text content from various document formats using Unstructured library.

    Supports multiple formats: PDF, Word, Excel, PowerPoint, HTML, etc.
    Uses automatic file type detection and content partitioning.

    Args:
        doc (str): Path to document file

    Returns:
        list/str: Extracted content as list of strings, or error message
    """
    # Normalize file path to handle special characters and different OS formats
    doc = os.path.normpath(doc)

    # Verify file exists before processing
    if not os.path.exists(doc):
        print(f"File does not exist: {doc}")
        return ""

    try:
        # Use Unstructured's partition function - automatically detects file type
        # and extracts content into structured elements (paragraphs, tables, etc.)
        unstructured_doc = partition(filename=doc)
        print("File loaded successfully")  # Basic success logging

        all_content = []  # List to collect extracted text content

        if isinstance(unstructured_doc, list):
            # Most documents return a list of elements (paragraphs, tables, etc.)
            # Convert to JSON format for structured processing
            json_output = elements_to_json(unstructured_doc)

            # Extract text content from each element
            for element in unstructured_doc:
                try:
                    # Convert element object to string representation
                    element_string = str(element)
                    if element_string.strip():  # Only add non-empty content
                        all_content.append(element_string)
                    else:
                        print("Warning: Empty element found")
                except Exception as conversion_error:
                    # Skip problematic elements but continue processing
                    print(f"Warning: Could not convert element: {conversion_error}")
                    continue

            # Return list of extracted text strings
            return all_content

        elif isinstance(unstructured_doc, str):
            # Some formats (like plain text) return direct string content
            return unstructured_doc
        else:
            # Unexpected return type
            err = "No extractable content found in document.\n"
            return err

    except Exception as e:
        # Handle file reading errors, unsupported formats, etc.
        err = f"Error loading file: {e}"
        return err