Add project files for HF Space

.env.example

@@ -0,0 +1 @@
+OPENAI_API_KEY="YOUR_API_KEY_HERE"

.gitignore

@@ -0,0 +1,210 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+#poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+#pdm.lock
+#pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+#pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+# Abstra
+# Abstra is an AI-powered process automation framework.
+# Ignore directories containing user credentials, local state, and settings.
+# Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#  and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#  you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Cursor
+#  Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to
+#  exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
+#  refer to https://docs.cursor.com/context/ignore-files
+.cursorignore
+.cursorindexingignore
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/
+
+# Logs
+logs

Dockerfile

@@ -0,0 +1,33 @@
+# Use the official Python 3.10 slim-buster image as a lightweight base
+FROM python:3.10-slim
+
+# Set the application's working directory inside the container
+WORKDIR /app
+
+# Install essential system-level dependencies
+# - ffmpeg: Required by the pydub library for audio processing
+# - git: Required by some pip packages for installation from version control
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy the dependency file first to leverage Docker's layer caching
+COPY requirements.txt .
+
+# Install Python packages, disabling the cache to reduce image size
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the application source code into the container
+# force a re-installation of all packages
+COPY src/ src/
+
+# Expose the port Gradio will run on, making it accessible to the host
+EXPOSE 7860
+
+# Set the environment variable for the Ollama host when running inside Docker.
+ENV OLLAMA_HOST=host.docker.internal
+
+# Define the default command to run when the container starts
+# Uses Python's module flag '-m' for correct package path resolution
+CMD ["python", "-m", "src.app"]

app.py

@@ -0,0 +1,148 @@
+import gradio as gr
+import os
+from src.controllers.processing_controller import ProcessingController
+from src.utils.exceptions import AppError
+from src.logging_config import logger
+from src import config
+
+
+def create_controller():
+    """Factory function to create a new controller instance."""
+    return ProcessingController()
+
+def process_audio_file(audio_file_path, controller):
+    """
+    Handles the primary audio processing workflow when a user uploads a file.
+    """
+    # Define the updates for a failed state (all components disabled)
+    fail_updates = (
+        gr.update(), gr.update(interactive=False), gr.update(interactive=False),
+        gr.update(interactive=False), gr.update(interactive=False),
+        gr.update(interactive=False), "<p>Please upload a file to begin.</p>", []
+    )
+    # Define the updates for a successful state (all components enabled)
+    success_updates = (
+        gr.update(), gr.update(interactive=True), gr.update(interactive=True),
+        gr.update(interactive=True), gr.update(interactive=True),
+        gr.update(interactive=True), "<p style='color:green;'>File processed successfully. Ready for analysis.</p>", []
+    )
+
+    if audio_file_path is None:
+        return fail_updates
+        
+    try:
+        logger.info(f"UI received file: {audio_file_path}")
+        controller.process_audio_file(audio_file_path)
+        return success_updates
+    except AppError as e:
+        logger.error(f"UI caught an application error: {e}")
+        # Return a failure state but with a specific error message for the user.
+        error_return = list(fail_updates)
+        error_return[6] = f"<p style='color:red;'>Error: {e}</p>"
+        return tuple(error_return)
+    finally:
+        # Ensure temporary files created by Gradio are always cleaned up.
+        if audio_file_path and os.path.exists(audio_file_path):
+            try:
+                os.remove(audio_file_path)
+                logger.info(f"Cleaned up temporary file: {audio_file_path}")
+            except OSError as e:
+                logger.error(f"Error removing temporary file {audio_file_path}: {e}")
+
+def handle_question(question, controller):
+    """
+    Manages the conversational Q&A flow. It takes the user's question, gets the
+    model's response via the controller, and returns the updated chat history.
+    """
+    if not question.strip():
+        # Do not process empty questions, just return the current state.
+        return controller.chat_history, ""
+    try:
+        # The controller manages appending the new Q&A turn to its internal history.
+        controller.answer_question(question)
+        # Return the full, updated history and a blank string to clear the input box.
+        return controller.chat_history, ""
+    except AppError as e:
+        logger.error(f"UI caught an application error during Q&A: {e}")
+        # On error, create a temporary history to show the error without saving it to the permanent chat log.
+        temp_history = controller.chat_history + [[question, f"Error: {e}"]]
+        return temp_history, question
+
+def handle_transcript(controller, chat_history):
+    """Appends the full transcript to the current chat display."""
+    chat_history.append([None, controller.get_transcript()])
+    return chat_history
+
+def handle_summary(controller, chat_history):
+    """Appends a generated summary to the current chat display."""
+    chat_history.append([None, controller.get_summary()])
+    return chat_history
+
+def handle_sentiment(controller, chat_history):
+    """Appends a sentiment analysis to the current chat display."""
+    chat_history.append([None, controller.get_sentiment()])
+    return chat_history
+
+#  Gradio UI Definition 
+with gr.Blocks(theme=gr.themes.Default(), css="footer {visibility: hidden}", title=config.APP_TITLE) as demo:
+    # A session-specific state object to hold a unique controller instance for each user.
+    controller_state = gr.State(value=create_controller)
+
+    # Main layout starts here.
+    gr.Markdown(f"# 🗣️ {config.APP_TITLE}")
+    gr.Markdown(config.APP_DESCRIPTION)
+
+    with gr.Row(equal_height=True):
+        # Left column for inputs and actions
+        with gr.Column(scale=1):
+            audio_input = gr.Audio(type="filepath", label="Upload Audio File")
+            status_output = gr.Markdown(value="<p>Please upload a file to begin.</p>")
+            
+            with gr.Accordion("Analysis Actions", open=True):
+                transcript_btn = gr.Button("Show Full Transcript", interactive=False)
+                summarize_btn = gr.Button("Generate Summary", interactive=False)
+                sentiment_btn = gr.Button("Analyze Sentiment", interactive=False)
+        
+        # Right column for chatbot interaction
+        with gr.Column(scale=2):
+            chatbot_ui = gr.Chatbot(label="Chatbot", height=500, show_copy_button=True)
+            with gr.Row():
+                question_input = gr.Textbox(
+                    show_label=False,
+                    placeholder="Type your question here...",
+                    interactive=False,
+                    scale=4
+                )
+                submit_btn = gr.Button(
+                    value="Submit", 
+                    interactive=False, 
+                    variant="primary",
+                    scale=1
+                )
+
+    # Defines how UI components react to user actions.
+    # Handle the audio file upload and processing.
+    audio_input.upload(
+        fn=process_audio_file,
+        inputs=[audio_input, controller_state],
+        outputs=[
+            audio_input, transcript_btn, summarize_btn, sentiment_btn, 
+            question_input, submit_btn, status_output, chatbot_ui
+        ]
+    )
+
+    #  Handle the analysis actions.
+    transcript_btn.click(fn=handle_transcript, inputs=[controller_state, chatbot_ui], outputs=[chatbot_ui])
+    summarize_btn.click(fn=handle_summary, inputs=[controller_state, chatbot_ui], outputs=[chatbot_ui])
+    sentiment_btn.click(fn=handle_sentiment, inputs=[controller_state, chatbot_ui], outputs=[chatbot_ui])
+
+    # Handle the question submission.
+    question_input.submit(fn=handle_question, inputs=[question_input, controller_state], outputs=[chatbot_ui, question_input])
+    submit_btn.click(fn=handle_question, inputs=[question_input, controller_state], outputs=[chatbot_ui, question_input])
+
+if __name__ == "__main__":
+    logger.info("Starting Gradio application...")
+    demo.launch(
+        #share=True,
+        server_name="0.0.0.0",
+    )

docker-compose.yml

@@ -0,0 +1,42 @@
+version: '3.8'
+
+services:
+  # Service 1: Our Voice Analysis Application
+  app:
+    # Build the image using the Dockerfile in the current directory
+    build: .
+    # Expose the Gradio port
+    ports:
+      - "7860:7860"
+    # Set the environment variables for the application
+    environment:
+      # Define the Ollama host address
+      # This allows the application to connect to the Ollama server
+      - OLLAMA_HOST=ollama
+    # Make this service depend on the 'ollama' service
+    # This ensures that Ollama starts up before our application tries to connect to it
+    depends_on:
+      - ollama
+
+  # Service 2: The Ollama Server
+  ollama:
+    # Use the official Ollama Docker image
+    image: ollama/ollama
+    # Expose the Ollama API port so our 'app' service can reach it.
+    ports:
+      - "11434:11434"
+    # Set the environment variable to enable GPU support
+    deploy:
+      resources:
+        reservations:
+          devices:
+            - driver: nvidia
+              count: all
+              capabilities: [gpu]
+    # Mount a volume to persist Ollama data
+    # This allows Ollama to retain its state and models across container restarts
+    volumes:
+      - ollama_data:/root/.ollama
+
+volumes:
+  ollama_data:

requirements.txt

@@ -0,0 +1,12 @@
+gradio
+transformers==4.41.2
+accelerate==0.29.3
+torch==2.3.0
+openai-whisper==20231117
+openai
+pydub==0.25.1
+python-dotenv==1.0.1
+requests==2.32.3
+pytest==8.2.2
+pytest-mock==3.14.0
+ollama==0.2.1

src/config.py

@@ -0,0 +1,39 @@
+import os
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Application Configuration
+APP_TITLE = "Voice Analysis Toolkit"
+APP_DESCRIPTION = (
+    "Upload an audio file to transcribe, summarize, analyze sentiment, "
+    "and ask questions about its content. All processing is done locally "
+    "and your data remains private."
+)
+
+# Model Configuration
+MODEL_PROVIDER = "local"  # Set to 'local' or 'openai'
+OLLAMA_HOST = os.getenv("OLLAMA_HOST", "localhost") # Default to localhost if not set
+OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3") # Default model for Ollama
+
+# Local model settings (if MODEL_PROVIDER is 'local')
+LOCAL_TRANSCRIPTION_MODEL = "openai/whisper-base.en" 
+LOCAL_ANALYSIS_MODEL = "microsoft/Phi-3-mini-4k-instruct"
+
+# OpenAI API settings (if MODEL_PROVIDER is 'openai')
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+OPENAI_TRANSCRIPTION_MODEL = "whisper-1"
+OPENAI_ANALYSIS_MODEL = "gpt-3.5-turbo"
+
+# File Validation Configuration
+# Maximum file size in megabytes (MB)
+MAX_FILE_SIZE_MB = 25 
+# Maximum audio duration in minutes
+MAX_FILE_LENGTH_MINS = 15
+# List of allowed audio file extensions (add more as needed)
+ALLOWED_FILE_EXTENSIONS = [".mp3", ".wav", ".m4a", ".flac", ".ogg"]
+
+# Logging Configuration
+LOG_FILE_PATH = "logs/app.log"
+LOG_LEVEL = "INFO" # Can be "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"

src/controllers/processing_controller.py

@@ -0,0 +1,104 @@
+from src.services.transcription_service import TranscriptionService
+from src.services.analysis_service import AnalysisService
+from src.utils.validator import Validator
+from src.utils.exceptions import AppError
+from src.logging_config import logger
+
+
+class ProcessingController:
+    """
+    The central controller that orchestrates the entire analysis process.
+    It manages the application's state and coordinates the services.
+    """
+    def __init__(self):
+        """
+        Initializes the controller and its required services.
+        """
+        self.transcription_service = TranscriptionService()
+        self.analysis_service = AnalysisService()
+        self.validator = Validator()
+        self.transcript: str | None = None
+        self.chat_history: list = []
+        logger.info("ProcessingController initialized.")
+
+    def process_audio_file(self, file_path: str):
+        """
+        The main workflow method. It validates and transcribes the audio file.
+        This method prepares the controller for on-demand analysis.
+
+        Args:
+            file_path: The path to the temporary audio file uploaded by the user.
+
+        Raises:
+            AppError: If any step in the validation or transcription fails.
+        """
+        try:
+            # Reset state for a new file
+            self.transcript = None
+            self.chat_history = []
+            logger.info(f"Starting processing for audio file: {file_path}")
+
+            # 1. Validate the file
+            self.validator.validate_audio_file(file_path)
+
+            # 2. Transcribe the file
+            self.transcript = self.transcription_service.transcribe(file_path)
+
+            logger.info(f"Successfully processed and transcribed file: {file_path}")
+
+        except AppError as e:
+            # Catch our known application errors, log them, and re-raise
+            logger.error(f"An application error occurred during processing: {e}", exc_info=True)
+            raise e
+        except Exception as e:
+            # Catch any other unexpected errors
+            logger.critical(f"An unexpected critical error occurred: {e}", exc_info=True)
+            raise AppError("An unexpected error occurred. Please check the logs.")
+
+    def _ensure_transcript_exists(self):
+        """
+        A private helper to check if a transcript is ready for analysis.
+        """
+        if not self.transcript:
+            logger.warning("Attempted to perform analysis before processing a file.")
+            raise AppError("Please process an audio file before requesting analysis.")
+
+    def get_transcript(self) -> str:
+        """
+        Returns the stored transcript.
+        """
+        self._ensure_transcript_exists()
+        logger.info("Transcript requested by user.")
+        return self.transcript
+
+    def get_summary(self) -> str:
+        """
+        Generates a summary for the currently loaded transcript.
+        """
+        self._ensure_transcript_exists()
+        logger.info("Summary requested by user.")
+        return self.analysis_service.summarize(self.transcript)
+
+    def get_sentiment(self) -> str:
+        """
+        Performs sentiment analysis on the currently loaded transcript.
+        """
+        self._ensure_transcript_exists()
+        logger.info("Sentiment analysis requested by user.")
+        return self.analysis_service.get_sentiment(self.transcript)
+
+    def answer_question(self, question: str) -> str:
+        """
+        Answers a question about the currently loaded transcript.
+        """
+        self._ensure_transcript_exists()
+        if not question or not question.strip():
+            raise AppError("Question cannot be empty.")
+            
+        logger.info(f"Question received from user: '{question}'")
+        # The entire history to the analysis service
+        response = self.analysis_service.answer_question(self.transcript, question, self.chat_history)
+        # Update history with the new turn
+        self.chat_history.append([question, response])
+        return response
+    

src/logging_config.py

@@ -0,0 +1,44 @@
+import logging
+import os
+from logging.handlers import RotatingFileHandler
+from src import config
+
+def setup_logging():
+    """
+    Configures the application's logger.
+    """
+    # Create logs directory if it doesn't exist
+    os.makedirs(os.path.dirname(config.LOG_FILE_PATH), exist_ok=True)
+
+    # Configure the root logger
+    logger = logging.getLogger()
+    logger.setLevel(config.LOG_LEVEL)
+
+    # Prevent logging from propagating to the root logger if handlers are already configured
+    if logger.hasHandlers():
+        logger.handlers.clear()
+
+    # Formatter
+    formatter = logging.Formatter(
+        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+
+    # Console Handler (for printing logs to the terminal)
+    ch = logging.StreamHandler()
+    ch.setFormatter(formatter)
+    logger.addHandler(ch)
+
+    # File Handler (for writing logs to a file)
+    # RotatingFileHandler ensures log files don't grow indefinitely
+    fh = RotatingFileHandler(
+        config.LOG_FILE_PATH,
+        maxBytes=10*1024*1024, # 10 MB
+        backupCount=5
+    )
+    fh.setFormatter(formatter)
+    logger.addHandler(fh)
+
+    return logger
+
+# Initialize the logger for the application
+logger = setup_logging()

src/services/analysis_service.py

@@ -0,0 +1,158 @@
+import ollama
+from openai import OpenAI, OpenAIError
+from src import config
+from src.utils.exceptions import AnalysisError, IrrelevantQuestionError
+from src.logging_config import logger
+
+
+class AnalysisService:
+    """
+    A service class for performing text analysis tasks.
+    It uses Ollama for local analysis and the OpenAI API for remote analysis.
+    """
+
+    def _analyze_local(self, prompt: str) -> str:
+        """
+        Generates a response by calling the local Ollama server using the host
+        address defined in the application's configuration.
+        """
+
+        # Read the configured host from config.py
+        ollama_host = config.OLLAMA_HOST
+        
+        try:
+            # Initialize the Ollama client with the determined host
+            client = ollama.Client(host=f"http://{ollama_host}:11434")
+            logger.info(f"Sending analysis request to Ollama server at {ollama_host}.")
+            
+            response = client.generate(
+                model=config.OLLAMA_MODEL,
+                prompt=prompt
+            )
+            
+            logger.info("Ollama analysis successful.")
+            return response['response'].strip()
+
+        except ollama.ResponseError as e:
+            logger.error(f"Ollama API error: {e.error}", exc_info=True)
+            raise AnalysisError(f"An error occurred with the Ollama API: {e.error}")
+        except Exception as e:
+            # Catch other potential issues like connection problems
+            logger.error(f"Error during Ollama request: {e}", exc_info=True)
+            raise AnalysisError("An unexpected error occurred while communicating with the Ollama server.")
+
+    def _analyze_openai(self, prompt: str) -> str:
+        """
+        Generates a response using the OpenAI API.
+        """
+        if not config.OPENAI_API_KEY:
+            logger.error("OpenAI API key not found for analysis.")
+            raise AnalysisError("OpenAI API key is not configured.")
+        
+        try:
+            logger.info("Sending analysis request to OpenAI.")
+            client = OpenAI(api_key=config.OPENAI_API_KEY)
+            
+            response = client.chat.completions.create(
+                model=config.OPENAI_ANALYSIS_MODEL,
+                messages=[{"role": "user", "content": prompt}]
+            )
+            
+            content = response.choices[0].message.content
+            logger.info("OpenAI analysis successful.")
+            return content.strip()
+        except OpenAIError as e:
+            logger.error(f"OpenAI API error during analysis: {e.response.text}", exc_info=True)
+            raise AnalysisError(f"An OpenAI API error occurred: {e.response.status_code}")
+        except Exception as e:
+            logger.error(f"An unexpected error occurred during OpenAI analysis: {e}", exc_info=True)
+            raise AnalysisError("An unexpected error occurred while using the OpenAI API.")
+
+    def _analyze(self, prompt: str) -> str:
+        """
+        Private dispatcher method to route analysis to the correct provider.
+        """
+        provider = config.MODEL_PROVIDER.lower()
+        if provider == 'local':
+            return self._analyze_local(prompt)
+        elif provider == 'openai':
+            return self._analyze_openai(prompt)
+        else:
+            logger.error(f"Invalid MODEL_PROVIDER configured: {config.MODEL_PROVIDER}")
+            raise ValueError(f"Invalid model provider '{config.MODEL_PROVIDER}' specified in config.")
+
+    def summarize(self, text: str) -> str:
+        """
+        Generates a summary of the provided text.
+        """
+        logger.info("Summarization task requested.")
+        prompt = f"""
+        Provide a concise summary of the following text. 
+        Focus on the key points and main conclusions.
+
+        Text:
+        ---
+        {text}
+        ---
+        Summary:
+        """
+        return self._analyze(prompt)
+
+    def get_sentiment(self, text: str) -> str:
+        """
+        Performs sentiment analysis on the provided text.
+        """
+        logger.info("Sentiment analysis task requested.")
+        prompt = f"""
+        Analyze the sentiment of the following text.
+        Your response must have two parts:
+        1.  **Sentiment:** Classify the sentiment as Positive, Negative, or Neutral.
+        2.  **Justification:** Briefly explain why you chose that sentiment, referencing key words or phrases from the text.
+
+        Format your response clearly using Markdown.
+
+        Text:
+        ---
+        {text}
+        ---
+        Sentiment:
+        """
+        return self._analyze(prompt)
+
+    def answer_question(self, text: str, question: str, chat_history: list) -> str:
+        """
+        Answers a question based on the provided text.
+        """
+        logger.info(f"Q&A task requested for question: '{question}'")
+        # Format the chat history for the prompt
+        formatted_history = "\n".join([f"User: {q}\nAssistant: {a}" for q, a in chat_history])
+        
+        prompt = f"""
+        You are a machine. You are a Q&A engine that answers questions about a document.
+        You MUST follow these rules strictly:
+        1. Use the "Conversation History" to understand the user's question, especially for follow-ups.
+        2. Find the answer to the user's "New User Question" using ONLY the "Document Transcript".
+        3. If the answer is not in the transcript, you MUST ONLY respond with the exact phrase: 'That information is not available in the provided document.'
+        4. Do not apologize. Do not explain your reasoning. Do not add any other words.
+
+        ---
+        **DOCUMENT TRANSCRIPT:**
+        {text}
+        ---
+        **CONVERSATION HISTORY:**
+        {formatted_history}
+        ---
+        **NEW USER QUESTION:**
+        {question}
+        """
+        
+        response = self._analyze(prompt)
+        
+        # Check for our custom error signal from the LLM
+        if "ERROR: The answer to this question cannot be found" in response:
+            logger.warning(f"Model indicated question '{question}' is unanswerable from text.")
+            raise IrrelevantQuestionError(
+                "The question could not be answered based on the provided audio content."
+            )
+            
+        return response

src/services/transcription_service.py

@@ -0,0 +1,115 @@
+import torch
+from transformers import pipeline, Pipeline
+from openai import OpenAI, OpenAIError
+from src import config
+from src.utils.exceptions import TranscriptionError
+from src.logging_config import logger
+
+
+class TranscriptionService:
+    """
+    A service class for handling audio transcription.
+    It can use either a local model or the OpenAI API based on the configuration.
+    """
+    _local_pipeline: Pipeline = None
+
+    @classmethod
+    def _get_local_pipeline(cls) -> Pipeline:
+        """
+        Initializes and returns the local transcription pipeline.
+        This method uses a class-level variable to cache the pipeline,
+        ensuring the model is loaded only once.
+        """
+        if cls._local_pipeline is None:
+            try:
+                logger.info(
+                    f"Initializing local transcription model: {config.LOCAL_TRANSCRIPTION_MODEL}"
+                )
+                # Check for GPU availability
+                device = "cuda:0" if torch.cuda.is_available() else "cpu"
+                logger.info(f"Using device: {device} for transcription.")
+                
+                cls._local_pipeline = pipeline(
+                    "automatic-speech-recognition",
+                    model=config.LOCAL_TRANSCRIPTION_MODEL,
+                    device=device
+                )
+                logger.info("Local transcription model initialized successfully.")
+            except Exception as e:
+                logger.critical(f"Failed to load local transcription model: {e}", exc_info=True)
+                raise TranscriptionError(
+                    "Could not initialize the local transcription model. "
+                    "Please check model name and dependencies."
+                )
+        return cls._local_pipeline
+
+    def _transcribe_local(self, file_path: str) -> str:
+        """
+        Transcribes audio using a local Hugging Face model.
+        """
+        try:
+            logger.info(f"Starting local transcription for {file_path}")
+            pipeline = self._get_local_pipeline()
+            # The pipeline handles chunking for long audio files automatically
+            result = pipeline(file_path)
+            transcript_text = result["text"].strip()
+            logger.info(f"Local transcription successful for {file_path}")
+            return transcript_text
+        except Exception as e:
+            logger.error(f"Error during local transcription for {file_path}: {e}", exc_info=True)
+            raise TranscriptionError("An unexpected error occurred during local transcription.")
+
+    def _transcribe_openai(self, file_path: str) -> str:
+        """
+        Transcribes audio using the OpenAI API.
+        """
+        if not config.OPENAI_API_KEY:
+            logger.error("OpenAI API key not found for transcription.")
+            raise TranscriptionError("OpenAI API key is not configured.")
+        
+        try:
+            logger.info(f"Sending transcription request to OpenAI for {file_path}")
+            client = OpenAI(api_key=config.OPENAI_API_KEY)
+            
+            with open(file_path, "rb") as audio_file:
+                transcript = client.audio.transcriptions.create(
+                    model=config.OPENAI_TRANSCRIPTION_MODEL,
+                    file=audio_file
+                )
+            
+            transcript_text = transcript.text.strip()
+            logger.info(f"OpenAI transcription successful for {file_path}")
+            return transcript_text
+        except OpenAIError as e:
+            logger.error(f"OpenAI API error during transcription for {file_path}: {e.response.text}", exc_info=True)
+            raise TranscriptionError(f"An OpenAI API error occurred: {e.response.status_code}")
+        except Exception as e:
+            logger.error(f"An unexpected error occurred during OpenAI transcription for {file_path}: {e}", exc_info=True)
+            raise TranscriptionError("An unexpected error occurred while using the OpenAI API.")
+
+
+    def transcribe(self, file_path: str) -> str:
+        """
+        Public method to transcribe an audio file.
+        Delegates to the appropriate method based on the MODEL_PROVIDER config.
+
+        Args:
+            file_path: The path to the audio file to be transcribed.
+
+        Returns:
+            The transcribed text as a string.
+        
+        Raises:
+            TranscriptionError: If the transcription process fails.
+            ValueError: If the configured MODEL_PROVIDER is invalid.
+        """
+        provider = config.MODEL_PROVIDER.lower()
+        logger.info(f"Transcription requested with provider: {provider}")
+
+        if provider == 'local':
+            return self._transcribe_local(file_path)
+        elif provider == 'openai':
+            return self._transcribe_openai(file_path)
+        else:
+            logger.error(f"Invalid MODEL_PROVIDER configured: {config.MODEL_PROVIDER}")
+            raise ValueError(f"Invalid model provider '{config.MODEL_PROVIDER}' specified in config.")

src/utils/exceptions.py

@@ -0,0 +1,31 @@
+class AppError(Exception):
+    """Base exception class for the application."""
+    pass
+
+class ValidationError(AppError):
+    """Custom exception for data validation errors."""
+    pass
+
+class InvalidFileType(ValidationError):
+    """Raised when the uploaded file type is not allowed."""
+    pass
+
+class FileSizeExceeded(ValidationError):
+    """Raised when the uploaded file is larger than the allowed limit."""
+    pass
+
+class FileLengthExceeded(ValidationError):
+    """Raised when the uploaded audio file's duration is longer than the allowed limit."""
+    pass
+
+class TranscriptionError(AppError):
+    """Raised when there is an error during the transcription process."""
+    pass
+
+class AnalysisError(AppError):
+    """Raised when there is an error during the text analysis process."""
+    pass
+
+class IrrelevantQuestionError(AnalysisError):
+    """Raised when a user's question is not related to the provided text."""
+    pass

src/utils/validator.py

@@ -0,0 +1,87 @@
+import os
+from pydub import AudioSegment
+from pydub.exceptions import CouldntDecodeError
+from src import config
+from src.utils.exceptions import (
+    InvalidFileType,
+    FileSizeExceeded,
+    FileLengthExceeded,
+    ValidationError
+)
+from src.logging_config import logger
+
+
+class Validator:
+    """
+    A class to handle all input validation for the application.
+    """
+
+    @staticmethod
+    def validate_audio_file(file_path: str):
+        """
+        Validates an uploaded audio file against the rules in config.py.
+
+        This method checks for file existence, type, size, and duration.
+
+        Args:
+            file_path: The path to the uploaded audio file.
+
+        Raises:
+            ValidationError: If the file does not exist.
+            InvalidFileType: If the file extension is not in the allowed list.
+            FileSizeExceeded: If the file size is over the configured limit.
+            FileLengthExceeded: If the audio duration is over the configured limit.
+            ValidationError: If the audio file is corrupted or cannot be read.
+        """
+        logger.info(f"Initiating validation for file: {file_path}")
+
+        # 1. Check for file existence
+        if not os.path.exists(file_path):
+            logger.error(f"Validation failed: File not found at {file_path}")
+            raise ValidationError(f"File not found at path: {file_path}")
+
+        # 2. Validate file type (extension)
+        _, ext = os.path.splitext(file_path)
+        if ext.lower() not in config.ALLOWED_FILE_EXTENSIONS:
+            logger.warning(
+                f"Validation failed: Invalid file type '{ext}' for {file_path}"
+            )
+            raise InvalidFileType(
+                f"Invalid file type. Allowed types are: "
+                f"{', '.join(config.ALLOWED_FILE_EXTENSIONS)}"
+            )
+
+        # 3. Validate file size
+        file_size_mb = os.path.getsize(file_path) / (1024 * 1024)
+        if file_size_mb > config.MAX_FILE_SIZE_MB:
+            logger.warning(
+                f"Validation failed: File size {file_size_mb:.2f}MB exceeds "
+                f"limit of {config.MAX_FILE_SIZE_MB}MB for {file_path}"
+            )
+            raise FileSizeExceeded(
+                f"File size of {file_size_mb:.2f}MB exceeds the "
+                f"{config.MAX_FILE_SIZE_MB}MB limit."
+            )
+
+        # 4. Validate file duration
+        try:
+            audio = AudioSegment.from_file(file_path)
+            duration_mins = audio.duration_seconds / 60
+            if duration_mins > config.MAX_FILE_LENGTH_MINS:
+                logger.warning(
+                    f"Validation failed: Duration {duration_mins:.2f} mins exceeds "
+                    f"limit of {config.MAX_FILE_LENGTH_MINS} mins for {file_path}"
+                )
+                raise FileLengthExceeded(
+                    f"Audio duration of {duration_mins:.2f} minutes exceeds the "
+                    f"{config.MAX_FILE_LENGTH_MINS} minute limit."
+                )
+        except CouldntDecodeError:
+            logger.error(f"Validation failed: Could not decode audio file {file_path}. "
+                         "It may be corrupted or an unsupported format.")
+            raise ValidationError(
+                "Failed to read audio file. It may be corrupted or in an "
+                "unsupported format despite the file extension."
+            )
+
+        logger.info(f"Validation successful for file: {file_path}")