update: code reformating

transcript.py

@@ -0,0 +1,467 @@
+import json
+from dataclasses import dataclass
+from typing import Dict, List, Union
+
+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 = ""
+
+
+@dataclass
+class AudioSegment:
+    id: int
+    transcript: str
+    start_time: float
+    end_time: float
+    speaker_label: str
+    original_file: str
+    items: List[int]
+
+
+class TranscriptProcessor:
+    def __init__(
+        self,
+        transcript_file: str = None,
+        transcript_data: Union[dict, list] = None,
+        call_type: str = "le",
+        person_names: list = None,
+    ):
+        self.transcript_file = transcript_file
+        self.transcript_data = transcript_data
+        self.formatted_transcript = None
+        self.segments = []
+        self.speaker_mapping = {}
+        self.person_names = person_names
+
+        if self.transcript_file:
+            self._load_transcript()
+        elif self.transcript_data:
+            if call_type == "rp":
+                self.merge_transcripts(transcript_data, person_names)
+        else:
+            raise ValueError(
+                "Either transcript_file or transcript_data must be provided."
+            )
+
+        self._process_transcript()
+        self._create_formatted_transcript()  # Create initial formatted transcript
+        if call_type != "si" and call_type != "rp":
+            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:
+        results = self.transcript_data["results"]
+        current_words = []
+        current_speaker = None
+        current_start = None
+        current_items = []
+
+        for item in results["items"]:
+            if item["type"] == "pronunciation":
+                if not self.person_names:
+                    speaker = (
+                        item.get("speaker_label", "")
+                        .replace("spk_", "")
+                        .replace("spk", "")
+                    )
+                else:
+                    speaker = item.get("speaker_label", "")
+                    print("ITEM", item)
+
+                # Initialize on first pronunciation item
+                if current_speaker is None:
+                    current_speaker = speaker
+                    current_start = float(item["start_time"])
+
+                # Check for speaker change
+                if speaker != current_speaker:
+                    if current_items:
+                        self._create_segment(
+                            current_speaker,
+                            current_start,
+                            float(item["start_time"]),
+                            current_items,
+                        )
+                    current_items = []
+                    current_words = []
+                    current_speaker = speaker
+                    current_start = float(item["start_time"])
+
+                current_items.append(item)
+                current_words.append(item["alternatives"][0]["content"])
+            elif item["type"] == "punctuation":
+                current_items.append(item)
+                # Only check for segment break if we're over 20 words
+                if len(current_words) >= 20:
+                    # Break on this punctuation
+                    next_item = next(
+                        (
+                            it
+                            for it in results["items"][
+                                results["items"].index(item) + 1 :
+                            ]
+                            if it["type"] == "pronunciation"
+                        ),
+                        None,
+                    )
+                    if next_item:
+                        self._create_segment(
+                            current_speaker,
+                            current_start,
+                            float(next_item["start_time"]),
+                            current_items,
+                        )
+                        current_items = []
+                        current_words = []
+                        current_start = float(next_item["start_time"])
+
+        # Don't forget the last segment
+        if current_items:
+            last_time = max(
+                float(item["end_time"])
+                for item in current_items
+                if item["type"] == "pronunciation"
+            )
+            self._create_segment(
+                current_speaker, current_start, last_time, current_items
+            )
+
+    def _create_segment(
+        self, speaker_id: str, start: float, end: float, items: list
+    ) -> None:
+        segment_content = []
+        for item in items:
+            if item["type"] == "pronunciation":
+                segment_content.append(item["alternatives"][0]["content"])
+            elif item["type"] == "punctuation":
+                # Append punctuation to the last word without a space
+                if segment_content:
+                    segment_content[-1] += item["alternatives"][0]["content"]
+
+        if segment_content:
+            self.segments.append(
+                TranscriptSegment(
+                    speaker_id=speaker_id,
+                    start_time=start,
+                    end_time=end,
+                    text=" ".join(segment_content),
+                )
+            )
+
+    def correct_speaker_mapping_with_agenda(self, url: str) -> None:
+        """Fetch agenda from a URL and correct the speaker mapping using OpenAI."""
+        try:
+            if not url.startswith("http"):
+                # add https to the url
+                url = "https://" + url
+
+            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 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. Who analyzes the given transcript, original speaker mapping and agenda. From the Agenda, you fix the spelling mistakes in the speaker names or update the names if they are similar to the agenda. 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.",
+                    },
+                    {"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
+
+            # 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 _create_formatted_transcript(self) -> None:
+        """Create formatted transcript with default speaker labels."""
+        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)
+            # Use default speaker label (spk_X) if no mapping exists
+            if not self.person_names:
+                speaker_label = f"spk_{seg.speaker_id}"
+            else:
+                speaker_label = f"{seg.speaker_id}"
+            formatted_segments.append(
+                f"time_stamp: {start_time_str}-{end_time_str}\n"
+                f"{speaker_label}: {seg.text}\n"
+            )
+        self.formatted_transcript = "\n".join(formatted_segments)
+
+    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",
+                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:
+                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 = {}
+
+            # Update segments with speaker names and recreate formatted transcript
+            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
+
+            self._create_formatted_transcript_with_names()
+
+        except Exception as e:
+            print(f"Error mapping speaker IDs to names: {str(e)}")
+            self.speaker_mapping = {}
+
+    def _create_formatted_transcript_with_names(self) -> None:
+        """Create formatted transcript with mapped 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)
+            speaker_name = getattr(seg, "speaker_name", f"spk_{seg.speaker_id}")
+            formatted_segments.append(
+                f"Start Time: {start_time_str} - End Time: {end_time_str}\n"
+                f"{speaker_name}: {seg.text}\n"
+            )
+        self.formatted_transcript = "\n".join(formatted_segments)
+
+    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 merge_transcripts(
+        self, transcript_files: List[Dict], person_names: List[str]
+    ) -> None:
+        """
+        Merge multiple AWS diarized transcripts while maintaining correct time ordering.
+        Each transcript is assumed to have one speaker (spk_0) and person_names list index
+        corresponds to transcript file index.
+        """
+        print(person_names)
+        if len(transcript_files) != len(person_names):
+            raise ValueError("Number of transcripts must match number of speaker names")
+
+        # Initialize merged structure
+        merged_transcript = {
+            "jobName": "merged_transcript",
+            "status": "COMPLETED",
+            "results": {
+                "audio_segments": [],
+                "items": [],
+                "speaker_labels": {"segments": [], "speakers": len(transcript_files)},
+            },
+        }
+
+        # First collect all items with their original data and file index
+        all_items = []
+        for file_idx, transcript in enumerate(transcript_files):
+            items = transcript["results"].get("items", [])
+            speaker_name = person_names[file_idx]
+
+            for item in items:
+                # Store original item data along with file index and original ID
+                item_data = dict(item)
+                # if "speaker_label" in item_data:
+                item_data["speaker_label"] = speaker_name
+                item_data["file_idx"] = file_idx
+                item_data["original_id"] = item["id"]
+                item_data["start_time"] = float(item.get("start_time", 0))
+                item_data["end_time"] = float(item.get("end_time", 0))
+                all_items.append(item_data)
+
+        # Sort items by start time
+        all_items.sort(key=lambda x: (x["start_time"], x["end_time"]))
+
+        # Create mapping from (file_idx, original_id) to new sequential ID
+        item_id_mapping = {}
+
+        # Assign new sequential IDs and add to merged transcript
+        for new_id, item in enumerate(all_items):
+            file_idx = item.pop("file_idx")
+            original_id = item.pop("original_id")
+            item_id_mapping[(file_idx, original_id)] = new_id
+
+            # Update item ID and convert times back to strings
+            item["id"] = new_id
+            item["start_time"] = str(item["start_time"])
+            item["end_time"] = str(item["end_time"])
+
+            merged_transcript["results"]["items"].append(item)
+
+        # Process audio segments
+        all_segments = []
+        for file_idx, transcript in enumerate(transcript_files):
+            file_segments = transcript["results"].get("audio_segments", [])
+            speaker_name = person_names[file_idx]
+
+            for segment in file_segments:
+                # Map original item IDs to new sequential IDs
+                new_items = [
+                    item_id_mapping[(file_idx, item_id)]
+                    for item_id in segment.get("items", [])
+                ]
+
+                all_segments.append(
+                    AudioSegment(
+                        id=len(all_segments),
+                        transcript=segment["transcript"],
+                        start_time=float(segment["start_time"]),
+                        end_time=float(segment["end_time"]),
+                        speaker_label=speaker_name,
+                        original_file=f"file_{file_idx}",
+                        items=new_items,
+                    )
+                )
+
+        # Sort segments by start time
+        sorted_segments = sorted(all_segments, key=lambda x: x.start_time)
+
+        # Convert segments back to dictionary format
+        for idx, segment in enumerate(sorted_segments):
+            merged_segment = {
+                "id": idx,
+                "transcript": segment.transcript,
+                "start_time": str(segment.start_time),
+                "end_time": str(segment.end_time),
+                "speaker_label": segment.speaker_label,
+                "source_file": segment.original_file,
+                "items": sorted(segment.items),
+            }
+            merged_transcript["results"]["audio_segments"].append(merged_segment)
+
+            # Add to speaker labels segments
+            speaker_segment = {
+                "start_time": str(segment.start_time),
+                "end_time": str(segment.end_time),
+                "speaker_label": segment.speaker_label,
+            }
+            merged_transcript["results"]["speaker_labels"]["segments"].append(
+                speaker_segment
+            )
+
+        # Update the instance transcript data with merged result
+        self.transcript_data = merged_transcript

utils.py

@@ -0,0 +1,237 @@
+import json
+import os
+
+import requests
+
+
+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 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")
+
+
+openai_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,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "correct_call_type",
+            "description": "If the user tells you the correct call type, you have to apologize and call this function with correct call type.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "call_type": {
+                        "type": "string",
+                        "description": "The correct call type. If street interview, call type is 'si'.",
+                    },
+                },
+                "required": ["call_type"],
+                "additionalProperties": False,
+            },
+        },
+    },
+]
+
+
+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;
+    border-width: 0px !important;
+}
+
+#link-frame {
+    position: absolute !important;
+    width: 1px !important;
+    height: 1px !important;
+    right: -100px !important;
+    bottom: -100px !important;
+    display: none !important;
+}
+.html-container {
+    display: none !important;
+}
+a {
+    text-decoration: none !important;
+}
+#topic {
+    color: #999 !important;
+}
+.bubble-wrap {
+    padding-top: 0px !important;
+}
+.message-content {
+    border: 0px !important;
+    margin: 5px !important;
+}
+.message-row {
+    border-style: none !important;
+    margin: 0px !important;
+    width: 100% !important;
+    max-width: 100% !important;
+}
+.flex-wrap {
+    border-style: none !important;
+}
+
+.panel-full-width {
+    border-style: none !important;
+    border-width: 0px !important;
+}
+ol {
+    list-style-position: outside; 
+    margin-left: 20px;
+}
+
+body.waiting * {
+cursor: progress;
+}
+"""
+head = f"""
+    <script defer src="https://www.gstatic.com/firebasejs/11.1.0/firebase-firestore.js"></script>
+    <script type="module">
+        // Import the functions you need from the SDKs you need
+        import {{ initializeApp }} from "https://www.gstatic.com/firebasejs/11.1.0/firebase-app.js";
+        import {{ getDatabase, ref, set }} from "https://www.gstatic.com/firebasejs/11.1.0/firebase-database.js";
+        // TODO: Add SDKs for Firebase products that you want to use
+        // https://firebase.google.com/docs/web/setup#available-libraries
+        console.log("Initializing Firebase");
+        // Your web app's Firebase configuration
+        // For Firebase JS SDK v7.20.0 and later, measurementId is optional
+        const firebaseConfig = {{
+            apiKey: "{os.getenv('FIREBASE_API_KEY')}",
+            authDomain: "{os.getenv('FIREBASE_AUTH_DOMAIN')}",
+            databaseURL: "{os.getenv('FIREBASE_DATABASE_URL')}",
+            projectId: "{os.getenv('FIREBASE_PROJECT_ID')}",
+            storageBucket: "{os.getenv('FIREBASE_STORAGE_BUCKET')}",
+            messagingSenderId: "{os.getenv('FIREBASE_MESSAGING_SENDER_ID')}",
+            appId: "{os.getenv('FIREBASE_APP_ID')}",
+            measurementId: "{os.getenv('FIREBASE_MEASUREMENT_ID')}"
+        }};
+
+        // Initialize Firebase
+        const app = initializeApp(firebaseConfig);
+        const realtimeDB = getDatabase(app);
+        const rollAccount = "{os.getenv('ROLL_ACCOUNT')}";
+        const COLLECTIONS = {{
+            COLLAB_EDIT_LINK: "collab_link_handler",
+        }};
+
+        // Event listener for click
+        document.addEventListener('click', function (event) {{
+            var link = event.target.closest('a');
+            event.preventDefault();
+            if (link && link.href) {{
+
+                // Parse the URL to extract 'st' and 'et'
+                const url = new URL(link.href);
+                const startTime = url.searchParams.get('st');
+                const endTime = url.searchParams.get('et');
+                const userId = url.searchParams.get('uid') || "";
+
+                if (startTime || endTime) {{
+                    let components = url.pathname.split("/");
+                    let callId = components[2];
+                    let recordingSessionId = components[3];
+
+                    let data = {{
+                        startTime: parseInt(startTime, 10),
+                        endTime: parseInt(endTime, 10),
+                    }};
+
+                    console.log("Data to save:", data);
+
+                    // Firebase reference
+                    let reference = ref(
+                        realtimeDB,
+                        `${{rollAccount}}/${{COLLECTIONS.COLLAB_EDIT_LINK}}/${{userId}}/${{callId}}/${{recordingSessionId}}`
+                    );
+
+                    set(reference, data)
+                        .then(() => {{
+                            console.log("Data saved successfully:", data);
+                        }})
+                        .catch((error) => {{
+                            console.error("Error saving data:", error);
+                        }});
+                }}
+            }}
+        }});
+    </script>
+"""