app.py

import json
import os
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

import gradio as gr
import requests
from bs4 import BeautifulSoup
from openai import OpenAI


@dataclass
class TranscriptSegment:
    speaker_id: str
    start_time: float
    end_time: float
    text: str
    speaker_name: str = ""


class TranscriptProcessor:
    def __init__(self, transcript_file: str = None, transcript_data: dict = None):
        self.transcript_file = transcript_file
        self.transcript_data = transcript_data
        self.formatted_transcript = None
        self.segments = []
        self.text_windows = []
        self.window_size = 2
        self.speaker_mapping = {}
        if self.transcript_file:
            self._load_transcript()
        elif self.transcript_data:
            pass  # transcript_data is already set
        else:
            raise ValueError(
                "Either transcript_file or transcript_data must be provided."
            )

        self._process_transcript()
        self.map_speaker_ids_to_names()

    def _load_transcript(self) -> None:
        """Load the transcript JSON file."""
        with open(self.transcript_file, "r") as f:
            self.transcript_data = json.load(f)

    def _format_time(self, seconds: float) -> str:
        """Convert seconds to formatted time string (MM:SS)."""
        minutes = int(seconds // 60)
        seconds = int(seconds % 60)
        return f"{minutes:02d}:{seconds:02d}"

    def _process_transcript(self) -> None:
        """Process the transcript into segments with speaker information and create a formatted version with timestamps."""
        results = self.transcript_data["results"]

        # Process into segments
        for segment in results["speaker_labels"]["segments"]:
            speaker_id = segment.get("speaker_label", segment.get("speakerlabel", ""))
            speaker_id = (
                speaker_id.replace("spk_", "").replace("spk", "") if speaker_id else ""
            )

            start_time = float(segment.get("start_time", 0))
            end_time = float(segment.get("end_time", 0))

            items = [
                item
                for item in results["items"]
                if "start_time" in item
                and float(item["start_time"]) >= start_time
                and float(item["start_time"]) < end_time
                and item["type"] == "pronunciation"
            ]

            words = [item["alternatives"][0]["content"] for item in items]
            if words:
                self.segments.append(
                    TranscriptSegment(
                        speaker_id=speaker_id,
                        start_time=start_time,
                        end_time=end_time,
                        text=" ".join(words),
                    )
                )

        formatted_segments = []
        for seg in self.segments:
            start_time_str = self._format_time(seg.start_time)
            end_time_str = self._format_time(seg.end_time)
            formatted_segments.append(
                f"time_stamp: {start_time_str}-{end_time_str}\n"
                f"spk {seg.speaker_id}: {seg.text}\n"
            )

        self.formatted_transcript = "\n".join(formatted_segments)

        # Create sliding windows of text for better matching
        for i in range(len(self.segments)):
            # Combine current segment with next segments within window
            window_segments = self.segments[i : i + self.window_size]
            combined_text = " ".join(seg.text for seg in window_segments)
            if window_segments:
                self.text_windows.append(
                    {
                        "text": combined_text,
                        "start_time": window_segments[0].start_time,
                        "end_time": window_segments[-1].end_time,
                    }
                )

    def map_speaker_ids_to_names(self) -> None:
        """Map speaker IDs to names based on introductions in the transcript."""
        try:

            transcript = self.formatted_transcript

            prompt = (
                "Given the following transcript where speakers are identified as spk 0, spk 1, spk 2, etc., please map each spk ID to the speaker's name based on their introduction in the transcript. If no name is introduced for a speaker, keep it as spk_id. Return the mapping as a JSON object in the format {'spk_0': 'Speaker Name', 'spk_1': 'Speaker Name', ...}\n\n"
                f"Transcript:\n{transcript}"
            )

            client = OpenAI()

            completion = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": prompt},
                ],
                temperature=0,
            )

            response_text = completion.choices[0].message.content.strip()
            try:
                self.speaker_mapping = json.loads(response_text)
            except json.JSONDecodeError:
                # extract left most and right most {}
                response_text = response_text[
                    response_text.find("{") : response_text.rfind("}") + 1
                ]
                try:
                    self.speaker_mapping = json.loads(response_text)
                except json.JSONDecodeError:
                    print("Error parsing speaker mapping JSON.")
                    self.speaker_mapping = {}
            for segment in self.segments:
                spk_id = f"spk_{segment.speaker_id}"
                speaker_name = self.speaker_mapping.get(spk_id, spk_id)
                segment.speaker_name = speaker_name

            # Recreate the formatted transcript with speaker names
            formatted_segments = []
            for seg in self.segments:
                start_time_str = self._format_time(seg.start_time)
                end_time_str = self._format_time(seg.end_time)
                formatted_segments.append(
                    f"time_stamp: {start_time_str}-{end_time_str}\n"
                    f"{seg.speaker_name}: {seg.text}\n"
                )
            self.formatted_transcript = "\n".join(formatted_segments)

        except Exception as e:
            print(f"Error mapping speaker IDs to names: {str(e)}")
            self.speaker_mapping = {}

    def correct_speaker_mapping_with_agenda(self, url: str) -> None:
        """Fetch agenda from a URL and correct the speaker mapping using OpenAI."""
        try:

            response = requests.get(url)
            response.raise_for_status()
            html_content = response.text

            # Parse the HTML to find the desired description
            soup = BeautifulSoup(html_content, "html.parser")
            description_tag = soup.find(
                "script", {"type": "application/ld+json"}
            )  # Find the ld+json metadata block
            agenda = ""

            if description_tag:
                # Extract the JSON content
                json_data = json.loads(description_tag.string)
                if "description" in json_data:
                    agenda = json_data["description"]
                else:
                    print("Agenda description not found in the JSON metadata.")
            else:
                print("No structured data (ld+json) found.")

            if not agenda:
                print("No agenda found in the structured metadata. Trying meta tags.")

                # Fallback: Use meta description if ld+json doesn't have it
                meta_description = soup.find("meta", {"name": "description"})
                agenda = meta_description["content"] if meta_description else ""

            if not agenda:
                print("No agenda found in any description tags.")
                return

            prompt = (
                f"Given the speaker mapping {self.speaker_mapping}, agenda:\n{agenda}, and the transcript: {self.formatted_transcript}\n\n"
                "Some speaker names in the mapping might have spelling errors or be incomplete."
                "Please correct the names based on the agenda. Return the corrected mapping in JSON format as "
                "{'spk_0': 'Correct Name', 'spk_1': 'Correct Name', ...}."
                "You should only update the name if the name sounds very similar, or there is a good spelling overlap/ The Speaker Introduction matches the description of the Talk from Agends. If the name is totally unrelated, keep the original name."
            )

            client = OpenAI()

            completion = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": prompt},
                ],
                temperature=0,
            )

            response_text = completion.choices[0].message.content.strip()
            try:
                corrected_mapping = json.loads(response_text)
            except:
                response_text = response_text[
                    response_text.find("{") : response_text.rfind("}") + 1
                ]
                try:
                    corrected_mapping = json.loads(response_text)
                except json.JSONDecodeError:
                    print(
                        "Error parsing corrected speaker mapping JSON, keeping the original mapping."
                    )
                    corrected_mapping = self.speaker_mapping
            # Update the speaker mapping with corrected names
            self.speaker_mapping = corrected_mapping
            print("Corrected Speaker Mapping:", self.speaker_mapping)

            # Update the transcript segments with corrected names
            for segment in self.segments:
                spk_id = f"spk_{segment.speaker_id}"
                segment.speaker_name = self.speaker_mapping.get(spk_id, spk_id)

            # Recreate the formatted transcript with corrected names
            formatted_segments = []
            for seg in self.segments:
                start_time_str = self._format_time(seg.start_time)
                end_time_str = self._format_time(seg.end_time)
                formatted_segments.append(
                    f"time_stamp: {start_time_str}-{end_time_str}\n"
                    f"{seg.speaker_name}: {seg.text}\n"
                )
            self.formatted_transcript = "\n".join(formatted_segments)

        except requests.exceptions.RequestException as e:
            print(f"Error fetching agenda from URL: {str(e)}")
        except Exception as e:
            print(f"Error correcting speaker mapping: {str(e)}")

    def get_transcript(self) -> str:
        """Return the formatted transcript with speaker names."""
        return self.formatted_transcript

    def get_transcript_data(self) -> Dict:
        """Return the raw transcript data."""
        return self.transcript_data


def setup_openai_key() -> None:
    """Set up OpenAI API key from file."""
    try:
        with open("api.key", "r") as f:
            os.environ["OPENAI_API_KEY"] = f.read().strip()
    except FileNotFoundError:
        print("Using ENV variable")
        # raise FileNotFoundError(
        #     "api.key file not found. Please create it with your OpenAI API key."
        # )


def get_transcript_for_url(url: str) -> dict:
    """
    This function fetches the transcript data for a signed URL.
    If the URL results in a direct download, it processes the downloaded content.

    :param url: Signed URL for the JSON file
    :return: Parsed JSON data as a dictionary
    """
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }

    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        if "application/json" in response.headers.get("Content-Type", ""):
            return response.json()  # Parse and return JSON directly

        # If not JSON, assume it's a file download (e.g., content-disposition header)
        content_disposition = response.headers.get("Content-Disposition", "")
        if "attachment" in content_disposition:
            # Process the content as JSON
            return json.loads(response.content)

        return json.loads(response.content)

    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
    except requests.exceptions.RequestException as req_err:
        print(f"Request error occurred: {req_err}")
    except json.JSONDecodeError as json_err:
        print(f"JSON decoding error: {json_err}")

    return {}


def get_initial_analysis(
    transcript_processor: TranscriptProcessor,
    cid,
    rsid,
    origin,
    ct,
) -> str:
    """Perform initial analysis of the transcript using OpenAI."""
    try:
        transcript = transcript_processor.get_transcript()
        client = OpenAI()
        if "localhost" in origin:
            link_start = "http"
        else:
            link_start = "https"

        if ct == "si":  # street interview
            prompt = f"""This is a transcript for a street interview. Transcript: {transcript}
In this street interview, the host asks multiple questions to the interviewees.
The interviewee can repeat a single answer multiple time to get the best take.
Your job is to find out the timestamp of the best answer given by the interviewee (Do not include the Question timestamp by interviwer in this). If there are multiple attempts for a question, best part is the last part of the question. If no question was asked but something is repeated, please include that in the answer as well
The way to know if there are multiple takes to a question is to see in the transcript if the same text is repeated, If not then number of takes is 1.
Question 1 should always be the introduction if the speaker has introduced themselves to find the best introduction time (Last timestamp is the best timestamp), Rest of questions should be in the order they were asked.
Return format is:
1. Question: question
Number of takes: number
Best Answer timestamp: [Timestamp: start_time - end_time]({link_start}://{{origin}}/collab/{{cid}}/{{rsid}}?st={{start_time_in_sec}}&et={{end_time_in_sec}}"').
For Example:
If the start time is 10:13 and end time is 10:18, the url will be:
{link_start}://roll.ai/colab/1234aq_12314/51234151?st=613&et=618
In the URL, make sure that after RSID there is ? and then rest of the fields are added via &.
"""
        else:
            prompt = f"""Given the transcript {transcript}, For All the speakers, short list all people, news, events, trends, and source that are discussed by speakers along with the start time of that topic and end time of that topic from the transcript. Rank all topics based on what would make for the best social clips. I need atleast 3 topics per speaker.
You should mention the Speaker Name first, then atleast 3 posts with their timestamps, and so on.
Return format is:
Speaker Name
1.Topic: topic,
[Timestamp: start_time - end_time]({link_start}://{{origin}}/collab/{{cid}}/{{rsid}}?st={{start_time_in_sec}}&et={{end_time_in_sec}}"').
2....
For Example:
If the start time is 10:13 and end time is 10:18, the url will be:
{link_start}://roll.ai/colab/1234aq_12314/51234151?st=613&et=618
In the URL, make sure that after RSID there is ? and then rest of the fields are added via &.
"""

        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": f"You are a helpful assistant who is analyzing the transcript. The transcript is for Call ID: {cid}, Session ID: {rsid}, Origin: {origin}, Call Type: {ct}.",
                },
                {"role": "user", "content": prompt},
            ],
        )
        return completion.choices[0].message.content
    except Exception as e:
        print(f"Error in initial analysis: {str(e)}")
        return "An error occurred during initial analysis. Please check your API key and file path."


def chat(
    message: str,
    chat_history: List,
    transcript_processor: TranscriptProcessor,
    cid,
    rsid,
    origin,
    ct,
) -> str:
    try:
        client = OpenAI()

        if "localhost" in origin:
            link_start = "http"
        else:
            link_start = "https"
        prompt = f"""You are a helpful assistant analyzing transcripts and generating timestamps and URL. Call ID is {cid}, Session ID is {rsid}, origin is {origin}, Call Type is {ct}.
Transcript:\n{transcript_processor.get_transcript()}
If a user asks timestamps for a specific topic, find the start time and end time of that specific topic and return answer in the format: 
Answer format:
Topic: Heading [Timestamp: start_time - end_time]({link_start}://{{origin}}/collab/{{cid}}/{{rsid}}?st={{start_time_in_sec}}&et={{end_time_in_sec}}"').

For Example:
If the start time is 10:13 and end time is 10:18, the url will be:
{link_start}://roll.ai/colab/1234aq_12314/51234151?st=613&et=618
In the URL, make sure that after RSID there is ? and then rest of the fields are added via &.

"""
        messages = [{"role": "system", "content": prompt}]

        for user_msg, assistant_msg in chat_history:
            if user_msg is not None:  # Skip the initial message where user_msg is None
                messages.append({"role": "user", "content": user_msg})
            if assistant_msg is not None:
                messages.append({"role": "assistant", "content": assistant_msg})

        # Add the current message
        messages.append({"role": "user", "content": message})

        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
        )

        response = completion.choices[0].message

        return response.content

    except Exception as e:
        print(f"Unexpected error in chat: {str(e)}")
        import traceback

        print(f"Traceback: {traceback.format_exc()}")
        return "Sorry, there was an error processing your request."


def create_chat_interface():
    """Create and configure the chat interface."""
    css = """
    .gradio-container {

        padding-top: 0px !important;
        padding-left: 0px !important;
        padding-right: 0px !important;
        padding: 0px !important;
        margin: 0px !important;
    }
    #component-0 {
        gap: 0px !important;
    }

    .icon-button-wrapper{
        display: none !important;
    }


    footer {
        display: none !important;
    }
    #chatbot_box{
        flex-grow: 1 !important;
    }
    """
    with gr.Blocks(fill_height=True, fill_width=True, css=css) as demo:
        chatbot = gr.Chatbot(
            elem_id="chatbot_box",
            layout="bubble",
            show_label=False,
            show_share_button=False,
            show_copy_all_button=False,
            show_copy_button=False,
        )
        msg = gr.Textbox(elem_id="chatbot_textbox", show_label=False)
        transcript_processor_state = gr.State()  # maintain state of imp things
        call_id_state = gr.State()
        colab_id_state = gr.State()
        origin_state = gr.State()
        ct_state = gr.State()
        turl_state = gr.State()

        def on_app_load(request: gr.Request):
            cid = request.query_params.get("cid", None)
            rsid = request.query_params.get("rsid", None)
            origin = request.query_params.get("origin", None)
            ct = request.query_params.get("ct", None)
            turl = request.query_params.get("turl", None)

            # if any param is missing, return error
            if not cid or not rsid or not origin or not ct or not turl:
                error_message = "Error processing"
                chatbot_value = [(None, error_message)]
                return [
                    chatbot_value,
                    None,
                    None,
                    None,
                    None,
                    None,
                    None,
                ]

            try:
                transcript_data = get_transcript_for_url(turl)
                transcript_processor = TranscriptProcessor(
                    transcript_data=transcript_data
                )
                initial_analysis = get_initial_analysis(
                    transcript_processor, cid, rsid, origin, ct
                )

                chatbot_value = [
                    (None, initial_analysis)
                ]  # initialized with initial analysis and assistant is None
                return [
                    chatbot_value,
                    transcript_processor,
                    cid,
                    rsid,
                    origin,
                    ct,
                    turl,
                ]
            except Exception as e:
                error_message = f"Error processing call_id {cid}: {str(e)}"
                chatbot_value = [(None, error_message)]
                return [
                    chatbot_value,
                    None,
                    None,
                    None,
                    None,
                    None,
                    None,
                ]

        demo.load(
            on_app_load,
            inputs=None,
            outputs=[
                chatbot,
                transcript_processor_state,
                call_id_state,
                colab_id_state,
                origin_state,
                ct_state,
                turl_state,
            ],
        )

        def respond(
            message: str,
            chat_history: List,
            transcript_processor,
            cid,
            rsid,
            origin,
            ct,
        ):
            if not transcript_processor:
                bot_message = "Transcript processor not initialized."
            else:
                bot_message = chat(
                    message,
                    chat_history,
                    transcript_processor,
                    cid,
                    rsid,
                    origin,
                    ct,
                )
            chat_history.append((message, bot_message))
            return "", chat_history

        msg.submit(
            respond,
            [
                msg,
                chatbot,
                transcript_processor_state,
                call_id_state,
                colab_id_state,
                origin_state,
                ct_state,
            ],
            [msg, chatbot],
        )

    return demo


def main():
    """Main function to run the application."""
    try:
        setup_openai_key()
        demo = create_chat_interface()
        demo.launch(share=True)
    except Exception as e:
        print(f"Error starting application: {str(e)}")
        raise


if __name__ == "__main__":
    main()