import json import os from dataclasses import dataclass from typing import Dict, List 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 print(self.speaker_mapping) prompt = ( f"Given the original 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." "Remember that the content in agenda is accurate and transcript can have errors so prioritize the spellings and names in the agenda content." "If the speaker name and introduction is similar to the agenda, update the speaker name in the mapping." "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." "You should always include all the speakers in the mapping from the original mapping, even if you don't update their names. i.e if there are 4 speakers in original mapping, new mapping should have 4 speakers always, ignore all the other spekaers in the agenda. I REPEAT DO NOT ADD OTHER NEW SPEAKERS IN THE MAPPING." ) 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 Exception: 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" ching 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 Title Number of takes: number [Best Answer: 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 &. Keep the answer less verbose and to the point. """ 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. {{Title}}: [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 &. Keep the answer less verbose and to the point. """ 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: tools = [ { "type": "function", "function": { "name": "correct_speaker_name_with_url", "description": "If a User provides a link to Agenda file, call the correct_speaker_name_with_url function to correct the speaker names based on the url, i.e if a user says 'Here is the Luma link for the event' and provides a link to the event, the function will correct the speaker names based on the event.", "parameters": { "type": "object", "properties": { "url": { "type": "string", "description": "The url to the agenda.", }, }, "required": ["url"], "additionalProperties": False, }, }, } ] 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: If the user provides a link to the agenda, use the correct_speaker_name_with_url function to correct the speaker names based on the agenda. 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, tools=tools, ) response = completion.choices[0].message messages.append(response) if response.tool_calls: args = eval(response.tool_calls[0].function.arguments) url = args.get("url", None) if url: transcript_processor.correct_speaker_mapping_with_agenda(url) corrected_speaker_mapping = transcript_processor.speaker_mapping function_call_result_message = { "role": "tool", "content": json.dumps( { "speaker_mapping": f"Corrected Speaker Mapping is: {corrected_speaker_mapping}\n, All speakers should be addressed via this mapping. The next message should be comparing old speaker names (do not use spk_0, spk_1, use the old names) and corrected speaker names.", } ), "name": response.tool_calls[0].function.name, "tool_call_id": response.tool_calls[0].id, } # messages.append(response.choices[0]["message"]) messages.append(function_call_result_message) completion_payload = {"model": "gpt-4o-mini", "messages": messages} # print("messages", messages[3]) response = client.chat.completions.create(**completion_payload) # print("no error here") return response.choices[0].message.content else: return "No URL provided for correcting speaker names." 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; } #link-frame { position: absolute !important; width: 1px !important; height: 1px !important; right: -100px !important; bottom: -100px !important; display: none !important; } """ # js = """ # function createIframeHandler() { # let iframe = document.getElementById('link-frame'); # if (!iframe) { # iframe = document.createElement('iframe'); # iframe.id = 'link-frame'; # iframe.style.position = 'absolute'; # iframe.style.width = '1px'; # iframe.style.height = '1px'; # iframe.style.right = '-100px'; # iframe.style.bottom = '-100px'; # iframe.style.display = 'none'; // Hidden initially # document.body.appendChild(iframe); # } # document.addEventListener('click', function (event) { # var link = event.target.closest('a'); # if (link && link.href) { # try { # iframe.src = link.href; # iframe.style.display = 'block'; // Show iframe on link click # event.preventDefault(); # console.log('Opening link in iframe:', link.href); # } catch (error) { # console.error('Failed to open link in iframe:', error); # } # } # }); # return 'Iframe handler initialized'; # } # """ 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() iframe_html = "" gr.HTML(value=iframe_html) # Add iframe to the UI 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) required_params = ["cid", "rsid", "origin", "ct", "turl"] missing_params = [ param for param in required_params if request.query_params.get(param) is None ] if missing_params: error_message = ( f"Missing required parameters: {', '.join(missing_params)}" ) chatbot_value = [(None, error_message)] return [ chatbot_value, None, None, None, None, None, 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()