plug in user_assign

Author: cameledev

src/backend/core/api/viewsets.py

@@ -346,6 +346,7 @@ def start_room_recording(self, request, pk=None):  # pylint: disable=unused-argu
         ):
             try:
                 MetadataCollectorService().start(recording)
+                logger.info("Started MetadataCollectorService")
             except MetadataCollectorException:
                 logger.warning("Failed to start MetadataCollectorService")
 

src/backend/core/models.py

@@ -590,7 +590,6 @@ class Recording(BaseModel):
         verbose_name=_("Recording options"),
         help_text=_("Recording options"),
     )
-
     class Meta:
         db_table = "meet_recording"
         ordering = ("-created_at",)

src/backend/core/recording/event/notification.py

@@ -1,17 +1,20 @@
 """Service to notify external services when a new recording is ready."""
 
 import logging
+import os
 import smtplib
+from datetime import datetime, timezone
 
+import requests
+from asgiref.sync import async_to_sync
 from django.conf import settings
 from django.core.mail import send_mail
 from django.template.loader import render_to_string
 from django.utils.translation import get_language, override
 from django.utils.translation import gettext_lazy as _
+from livekit import api as livekit_api
 
-import requests
-
-from core import models
+from core import models, utils
 
 logger = logging.getLogger(__name__)
 
@@ -130,6 +133,51 @@ def _notify_user_by_email(recording) -> bool:
 
         return not has_failures
 
+    @staticmethod
+    def _get_recording_timestamps(worker_id):
+        """Fetch FileInfo.started_at and ended_at from LiveKit's egress API.
+
+        started_at is more accurate than EgressInfo.started_at because it
+        reflects when file recording actually began, not when the egress
+        process was initialized.
+
+        Returns:
+            Tuple of (started_at, ended_at) datetimes, either may be None.
+        """
+        if not worker_id:
+            return None, None
+
+        @async_to_sync
+        async def _fetch():
+            lkapi = utils.create_livekit_client()
+            try:
+                egress_list = await lkapi.egress.list_egress(
+                    livekit_api.ListEgressRequest(egress_id=worker_id)
+                )
+                if egress_list.items:
+                    file_results = egress_list.items[0].file_results
+                    if file_results:
+                        started = None
+                        ended = None
+                        if file_results[0].started_at:
+                            started = datetime.fromtimestamp(
+                                file_results[0].started_at / 1e9, tz=timezone.utc
+                            )
+                        if file_results[0].ended_at:
+                            ended = datetime.fromtimestamp(
+                                file_results[0].ended_at / 1e9, tz=timezone.utc
+                            )
+                        return started, ended
+            except Exception:
+                logger.exception(
+                    "Could not fetch egress info for worker %s", worker_id
+                )
+            finally:
+                await lkapi.aclose()
+            return None, None
+
+        return _fetch()
+
     @staticmethod
     def _notify_summary_service(recording):
         """Notify summary service about a new recording."""
@@ -150,24 +198,36 @@ def _notify_summary_service(recording):
             .first()
         )
 
+        # TODO: change how we get metadata_filename
+        output_folder = os.getenv("AWS_S3_OUTPUT_FOLDER", "metadata")
+        metadata_filename = f"{output_folder}/{recording.id}-metadata.json"
+
         if not owner_access:
             logger.error("No owner found for recording %s", recording.id)
             return False
+
+        started_at, ended_at = NotificationService._get_recording_timestamps(
+            recording.worker_id
+        )
+
         payload = {
             "owner_id": str(owner_access.user.id),
-            "filename": recording.key,
+            "recording_filename": recording.key,
+            "metadata_filename": metadata_filename,
             "email": owner_access.user.email,
             "sub": owner_access.user.sub,
             "room": recording.room.name,
             "language": recording.options.get("language"),
-            "recording_date": recording.created_at.astimezone(
-                owner_access.user.timezone
-            ).strftime("%Y-%m-%d"),
-            "recording_time": recording.created_at.astimezone(
-                owner_access.user.timezone
-            ).strftime("%H:%M"),
+            "worker_id": recording.worker_id,
+            "owner_timezone": str(owner_access.user.timezone),
             "download_link": f"{get_recording_download_base_url()}/{recording.id}",
             "context_language": owner_access.user.language,
+            "recording_started_at": (
+                started_at.isoformat() if started_at else None
+            ),
+            "recording_ended_at": (
+                ended_at.isoformat() if ended_at else None
+            ),
         }
 
         headers = {

src/summary/summary/api/route/tasks.py

@@ -19,16 +19,19 @@ class TranscribeSummarizeTaskCreation(BaseModel):
     """Transcription and summarization parameters."""
 
     owner_id: str
-    filename: str
+    recording_filename: str
+    metadata_filename: str
     email: str
     sub: str
     version: Optional[int] = 2
     room: Optional[str]
-    recording_date: Optional[str]
-    recording_time: Optional[str]
+    worker_id: Optional[str]
+    owner_timezone: Optional[str]
     language: Optional[str]
     download_link: Optional[str]
     context_language: Optional[str] = None
+    recording_started_at: Optional[str] = None
+    recording_ended_at: Optional[str] = None
 
     @field_validator("language")
     @classmethod
@@ -51,16 +54,19 @@ async def create_transcribe_summarize_task(request: TranscribeSummarizeTaskCreat
     task = process_audio_transcribe_summarize_v2.apply_async(
         args=[
             request.owner_id,
-            request.filename,
+            request.recording_filename,
+            request.metadata_filename,
             request.email,
             request.sub,
             time.time(),
             request.room,
-            request.recording_date,
-            request.recording_time,
+            request.worker_id,
+            request.owner_timezone,
             request.language,
             request.download_link,
             request.context_language,
+            request.recording_started_at,
+            request.recording_ended_at,
         ],
         queue=settings.transcribe_queue,
     )

src/summary/summary/core/analytics.py

@@ -112,7 +112,9 @@ def create(self, task_id, task_args):
         if self._is_disabled or self.has_task_id(task_id):
             return
 
-        _, filename, email, _, received_at, *_ = task_args
+        # Positional args mirror process_audio_transcribe_summarize_v2 signature:
+        # owner_id, recording_filename, metadata_filename, email, sub, received_at, ...
+        _, filename, _, email, _, received_at, *_ = task_args
 
         start_time = time.time()
         initial_metadata = {

src/summary/summary/core/celery_worker.py

@@ -4,6 +4,7 @@
 
 import json
 import time
+from datetime import datetime
 from typing import Optional
 
 import openai
@@ -28,6 +29,7 @@
     PROMPT_USER_PART,
 )
 from summary.core.transcript_formatter import TranscriptFormatter
+from summary.core.user_assign import assign_speakers
 from summary.core.webhook_service import submit_content
 
 settings = get_settings()
@@ -58,7 +60,7 @@ def init_sentry(**_kwargs):
 file_service = FileService()
 
 
-def transcribe_audio(task_id, filename, language):
+def transcribe_audio(task_id, recording_filename, language):
     """Transcribe an audio file using WhisperX.
 
     Downloads the audio from MinIO, sends it to WhisperX for transcription,
@@ -75,9 +77,13 @@ def transcribe_audio(task_id, filename, language):
 
     # Transcription
     try:
-        with file_service.prepare_audio_file(filename) as (audio_file, metadata):
+        with file_service.prepare_audio_file(recording_filename) as (
+            audio_file,
+            metadata,
+        ):
             metadata_manager.track(task_id, {"audio_length": metadata["duration"]})
 
+            # Compute language parameter
             if language is None:
                 language = settings.whisperx_default_language
                 logger.info(
@@ -90,22 +96,25 @@ def transcribe_audio(task_id, filename, language):
                     language,
                 )
 
+            # Call remote service for transcription
             transcription_start_time = time.time()
-
             transcription = whisperx_client.audio.transcriptions.create(
                 model=settings.whisperx_asr_model, file=audio_file, language=language
             )
 
-            transcription_time = round(time.time() - transcription_start_time, 2)
+            # Logging
+            transcription_duration = round(time.time() - transcription_start_time, 2)
             metadata_manager.track(
                 task_id,
-                {"transcription_time": transcription_time},
+                {"transcription_time": transcription_duration},
+            )
+            logger.info(
+                "Transcription received in %.2f seconds.", transcription_duration
             )
-            logger.info("Transcription received in %.2f seconds.", transcription_time)
             logger.debug("Transcription: \n %s", transcription)
 
     except FileServiceException:
-        logger.exception("Unexpected error for filename: %s", filename)
+        logger.exception("Unexpected error for recording: %s", recording_filename)
         return None
 
     metadata_manager.track_transcription_metadata(task_id, transcription)
@@ -117,8 +126,8 @@ def format_transcript(
     context_language,
     language,
     room,
-    recording_date,
-    recording_time,
+    recording_datetime,
+    owner_timezone,
     download_link,
 ):
     """Format a transcription into readable content with a title.
@@ -134,8 +143,8 @@ def format_transcript(
     return formatter.format(
         transcription,
         room=room,
-        recording_date=recording_date,
-        recording_time=recording_time,
+        recording_datetime=recording_datetime,
+        owner_timezone=owner_timezone,
         download_link=download_link,
     )
 
@@ -167,16 +176,19 @@ def format_actions(llm_output: dict) -> str:
 def process_audio_transcribe_summarize_v2(
     self,
     owner_id: str,
-    filename: str,
+    recording_filename: str,
+    metadata_filename: str,
     email: str,
     sub: str,
     received_at: float,
     room: Optional[str],
-    recording_date: Optional[str],
-    recording_time: Optional[str],
+    worker_id: Optional[str],
+    owner_timezone: Optional[str],
     language: Optional[str],
     download_link: Optional[str],
     context_language: Optional[str] = None,
+    recording_started_at: Optional[str] = None,
+    recording_ended_at: Optional[str] = None,
 ):
     """Process an audio file by transcribing it and generating a summary.
 
@@ -189,16 +201,21 @@ def process_audio_transcribe_summarize_v2(
     Args:
         self: Celery task instance (passed on with bind=True)
         owner_id: Unique identifier of the recording owner.
-        filename: Name of the audio file in MinIO storage.
+        recording_filename: Name of the audio file in MinIO storage.
+        metadata_filename: Name of the audio file in MinIO storage.
         email: Email address of the recording owner.
         sub: OIDC subject identifier of the recording owner.
         received_at: Unix timestamp when the recording was received.
         room: room name where the recording took place.
-        recording_date: Date of the recording (localized display string).
-        recording_time: Time of the recording (localized display string).
+        worker_id: LiveKit egress ID used to fetch the egress JSON from S3.
+        owner_timezone: IANA timezone of the recording owner (e.g. "Europe/Paris").
         language: ISO 639-1 language code for transcription.
         download_link: URL to download the original recording.
         context_language: ISO 639-1 language code of the meeting summary context text.
+        recording_started_at: ISO 8601 timestamp of when file recording actually started
+            (from LiveKit FileInfo.started_at via the egress_ended webhook).
+        recording_ended_at: ISO 8601 timestamp of when file recording ended
+            (from LiveKit FileInfo.ended_at via the egress_ended webhook).
     """
     logger.info(
         "Notification received | Owner: %s | Room: %s",
@@ -208,17 +225,44 @@ def process_audio_transcribe_summarize_v2(
 
     task_id = self.request.id
 
-    transcription = transcribe_audio(task_id, filename, language)
+    # Transcribe the audio
+    transcription = transcribe_audio(task_id, recording_filename, language)
     if transcription is None:
         return
 
+    # Use recording_started_at / recording_ended_at from the backend
+    # (sourced from LiveKit FileInfo via the egress_ended webhook).
+    recording_start_dt = None
+    recording_end_dt = None
+    if recording_start_dt:
+        recording_start_dt = datetime.fromisoformat(recording_started_at)
+    if recording_ended_at:
+        recording_end_dt = datetime.fromisoformat(recording_ended_at)
+
+    # Assign user names from metadata and transcription
+    try:
+        metadata = file_service.read_json(metadata_filename)
+    except Exception:
+        logger.exception("Failed to read metadata file: %s", metadata_filename)
+        metadata = None
+    if metadata and recording_start_dt is not None and recording_end_dt is not None:
+        assignment_result = assign_speakers(
+            metadata,
+            transcription,
+            recording_start_dt,
+            recording_end_dt,
+            overlap_threshold=0.5,
+        )
+        transcription = assignment_result.apply({"segments": transcription.segments})
+
+    # Format output
     content, title = format_transcript(
         transcription,
         context_language,
         language,
         room,
-        recording_date,
-        recording_time,
+        recording_started_at,
+        owner_timezone,
         download_link,
     )