assign users

Author: cameledev

src/agents/metadata_collector.py

@@ -156,7 +156,9 @@ async def on_chat_message_received(
     ):
         """Read a complete chat message and record it as an event."""
         full_text = await reader.read_all()
-        logger.info("Received chat message from %s", participant_identity)
+        logger.info(
+            "Received chat message from %s: '%s'", participant_identity, full_text
+        )
 
         self.events.append(
             MetadataEvent(
@@ -177,7 +179,7 @@ def handle_chat_stream(self, reader, participant_identity):
 
     def save(self):
         """Serialize collected events and upload as JSON to S3."""
-        logger.info("Persisting metadata…")
+        logger.info("Persisting metadata...")
 
         participants = []
         for k, v in self.participants.items():
@@ -372,7 +374,7 @@ async def entrypoint(ctx: JobContext):
     await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
 
     async def cleanup():
-        logger.info("Shutting down metadata collector…")
+        logger.info("Shutting down metadata collector...")
         await metadata_collector.aclose()
 
     ctx.add_shutdown_callback(cleanup)

src/backend/core/api/viewsets.py

@@ -347,6 +347,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,7 +1,9 @@
 """Service to notify external services when a new recording is ready."""
 
 import logging
+import os
 import smtplib
+from datetime import datetime, timezone
 
 from django.conf import settings
 from django.core.mail import send_mail
@@ -10,8 +12,10 @@
 from django.utils.translation import gettext_lazy as _
 
 import requests
+from asgiref.sync import async_to_sync
+from livekit import api as livekit_api
 
-from core import models
+from core import models, utils
 
 logger = logging.getLogger(__name__)
 
@@ -130,6 +134,49 @@ 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.
+
+        FileInfo.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_at = None
+                        ended_at = None
+                        if file_results[0].started_at:
+                            started_at = datetime.fromtimestamp(
+                                file_results[0].started_at / 1e9, tz=timezone.utc
+                            )
+                        if file_results[0].ended_at:
+                            ended_at = datetime.fromtimestamp(
+                                file_results[0].ended_at / 1e9, tz=timezone.utc
+                            )
+                        return started_at, ended_at
+            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 +197,32 @@ 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_start_at": (started_at.isoformat() if started_at else None),
+            "recording_end_at": (ended_at.isoformat() if ended_at else None),
         }
 
         headers = {

src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl

@@ -308,6 +308,12 @@ agentSubtitles:
     {{- end }}
     {{- end }}
     ENABLE_SILERO_VAD: "false"
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_OUTPUT_FOLDER: recordings
 
   image:
     repository: localhost:5001/meet-agents
@@ -317,6 +323,7 @@ agentSubtitles:
   command:
     - "python"
     - "multi_user_transcriber.py"
+    # - "metadata_collector.py" # TODO: FIX THIS
     - "start"
 
   # Extra volume mounts to manage our local custom CA and avoid to disable ssl

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_start_at: Optional[str] = None
+    recording_end_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_start_at,
+            request.recording_end_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 = {