diff --git a/CHANGELOG.md b/CHANGELOG.md index 54a35e56..3fbec3ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,10 @@ and this project adheres to - ✨(helm) add support multiple transcribe worker / endpoint #1247 - ✨(backend) make LiveKit Egress recording encoding configurable #1288 +### Changed + +- ♻️(summary) change tasks endpoint signature + ### Fixed - ♻(frontend) standardize role terminology across localizations diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index 916c51d0..fee4922a 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -1,7 +1,9 @@ """Service to notify external services when a new recording is ready.""" +import asyncio import logging 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 get_language, override 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__) @@ -131,7 +135,47 @@ class NotificationService: return not has_failures @staticmethod - def _notify_summary_service(recording): + async 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. The started_at value exposed + in the manifest file, as well as in the EgressInfo returned by the API, + corresponds to when the egress service received the request, not the moment + the egress worker effectively joined the room. + + Returns: + Tuple of (started_at, ended_at) datetimes, either may be None. + """ + + if not worker_id: + return None, None + + custom_configuration = {**settings.LIVEKIT_CONFIGURATION, "timeout": 10} + lkapi = utils.create_livekit_client(custom_configuration=custom_configuration) + try: + egress_list = await lkapi.egress.list_egress( + livekit_api.ListEgressRequest(egress_id=worker_id) # pylint: disable=no-member + ) + except (livekit_api.TwirpError, OSError, asyncio.TimeoutError): + logger.exception("Could not fetch egress info for worker %s", worker_id) + return None, None + finally: + await lkapi.aclose() + + if not egress_list.items or not egress_list.items[0].file_results: + logger.debug("No file_results for worker %s", worker_id) + return None, None + + file_result = egress_list.items[0].file_results[0] + + def _ns_to_utc(ns): + return datetime.fromtimestamp(ns / 1e9, tz=timezone.utc) if ns else None + + return _ns_to_utc(file_result.started_at), _ns_to_utc(file_result.ended_at) + + @staticmethod + def _notify_summary_service(recording: models.Recording): """Notify summary service about a new recording.""" if ( @@ -150,24 +194,37 @@ class NotificationService: .first() ) + if settings.METADATA_COLLECTOR_ENABLED and recording.options.get( + "collect_metadata", False + ): + output_folder = settings.METADATA_COLLECTOR_OUTPUT_FOLDER + metadata_filename = f"{output_folder}/{recording.id}-metadata.json" + else: + metadata_filename = None + if not owner_access: logger.error("No owner found for recording %s", recording.id) return False + + started_at, ended_at = async_to_sync( + 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, # For future use "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"), + "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 + ), # For future use } headers = { diff --git a/src/backend/meet/settings.py b/src/backend/meet/settings.py index 5a594347..a855a20d 100755 --- a/src/backend/meet/settings.py +++ b/src/backend/meet/settings.py @@ -855,6 +855,11 @@ class Base(Configuration): environ_name="METADATA_COLLECTOR_AGENT_NAME", environ_prefix=None, ) + METADATA_COLLECTOR_OUTPUT_FOLDER = values.Value( + "metadata", + environ_name="METADATA_COLLECTOR_OUTPUT_FOLDER", + environ_prefix=None, + ) # External Applications APPLICATION_ENABLED = values.BooleanValue( diff --git a/src/summary/summary/api/route/tasks.py b/src/summary/summary/api/route/tasks.py index 7645b4b5..91705f09 100644 --- a/src/summary/summary/api/route/tasks.py +++ b/src/summary/summary/api/route/tasks.py @@ -19,16 +19,18 @@ class TranscribeSummarizeTaskCreation(BaseModel): """Transcription and summarization parameters.""" owner_id: str - filename: str + recording_filename: str + metadata_filename: Optional[str] = None email: str sub: str version: Optional[int] = 2 room: Optional[str] - recording_date: Optional[str] - recording_time: 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 +53,18 @@ 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.owner_timezone, request.language, request.download_link, request.context_language, + request.recording_start_at, + request.recording_end_at, ], queue=settings.transcribe_queue, ) diff --git a/src/summary/summary/core/analytics.py b/src/summary/summary/core/analytics.py index d88b8e98..cffe1a37 100644 --- a/src/summary/summary/core/analytics.py +++ b/src/summary/summary/core/analytics.py @@ -112,7 +112,9 @@ class MetadataManager: 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 = { diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index fdd2c499..8b23fd1a 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -4,7 +4,6 @@ import json import time -from typing import Optional import openai import sentry_sdk @@ -78,7 +77,7 @@ file_service = FileService() def transcribe_audio( *, task_id: str, - filename: str | None = None, + recording_filename: str | None = None, language: str, cloud_storage_url=None, raises: bool = False, @@ -90,7 +89,7 @@ def transcribe_audio( Returns the transcription object, or None if the file could not be retrieved. """ - if bool(filename) == bool(cloud_storage_url): + if bool(recording_filename) == bool(cloud_storage_url): raise ValueError( "Either filename or cloud_storage_url must be provided, but not both." ) @@ -105,11 +104,12 @@ def transcribe_audio( # Transcription try: with file_service.prepare_audio_file( - remote_object_key=filename, + remote_object_key=recording_filename, cloud_storage_url=cloud_storage_url, ) 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( @@ -122,18 +122,21 @@ def transcribe_audio( 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 as e: @@ -148,7 +151,7 @@ def transcribe_audio( "Unexpected error while preparing file | filename: %s " "| cloud_storage_url: %s" ), - filename, + recording_filename, redacted_cloud_storage_url, ) return None @@ -162,8 +165,8 @@ def format_transcript( context_language: str | None, language: str, room: str | None, - recording_date: str | None, - recording_time: str | None, + recording_datetime: str | None, + owner_timezone: str | None, download_link: str | None, ) -> tuple[str, str]: """Format a transcription into readable content with a title. @@ -179,8 +182,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, ) @@ -188,7 +191,7 @@ def format_transcript( def format_actions(llm_output: dict) -> str: """Format the actions from the LLM output into a markdown list. - fomat: + format: - [ ] Action title Assignée à : assignee1, assignee2, Échéance : due_date """ lines = [] @@ -212,16 +215,18 @@ 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 | None, email: str, sub: str, received_at: float, - room: Optional[str], - recording_date: Optional[str], - recording_time: Optional[str], - language: Optional[str], - download_link: Optional[str], - context_language: Optional[str] = None, + room: str | None, + owner_timezone: str | None, + language: str | None, + download_link: str | None, + context_language: str | None = None, + recording_start_at: str | None = None, + recording_end_at: str | None = None, ): """Process an audio file by transcribing it and generating a summary. @@ -234,16 +239,20 @@ 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). + 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_start_at: ISO 8601 timestamp of when file recording actually started + (from LiveKit FileInfo.started_at via the egress_ended webhook). + recording_end_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", @@ -253,19 +262,21 @@ def process_audio_transcribe_summarize_v2( task_id = self.request.id + # Transcribe the audio transcription = transcribe_audio( - task_id=task_id, filename=filename, language=language + task_id=task_id, recording_filename=recording_filename, language=language ) if transcription is None: return + # Format output content, title = format_transcript( transcription, context_language, language, room, - recording_date, - recording_time, + recording_start_at, + owner_timezone, download_link, ) diff --git a/src/summary/summary/core/transcript_formatter.py b/src/summary/summary/core/transcript_formatter.py index 611806d2..4dec9e33 100644 --- a/src/summary/summary/core/transcript_formatter.py +++ b/src/summary/summary/core/transcript_formatter.py @@ -1,7 +1,9 @@ """Transcript formatting into readable conversation format with speaker labels.""" import logging -from typing import Optional, Tuple +from datetime import datetime +from typing import Tuple +from zoneinfo import ZoneInfo from summary.core.config import get_settings from summary.core.locales import LocaleStrings @@ -39,10 +41,10 @@ class TranscriptFormatter: def format( self, transcription, - room: Optional[str] = None, - recording_date: Optional[str] = None, - recording_time: Optional[str] = None, - download_link: Optional[str] = None, + room: str | None = None, + recording_datetime: str | None = None, + owner_timezone: str | None = None, + download_link: str | None = None, ) -> Tuple[str, str]: """Format transcription into the final document and its title.""" segments = self._get_segments(transcription) @@ -54,7 +56,7 @@ class TranscriptFormatter: content = self._remove_hallucinations(content) content = self._add_header(content, download_link) - title = self._generate_title(room, recording_date, recording_time) + title = self._generate_title(room, recording_datetime, owner_timezone) return content, title @@ -83,7 +85,7 @@ class TranscriptFormatter: return formatted_output - def _add_header(self, content, download_link: Optional[str]) -> str: + def _add_header(self, content, download_link: str | None) -> str: """Add download link header to the document content.""" if not download_link: return content @@ -97,16 +99,20 @@ class TranscriptFormatter: def _generate_title( self, - room: Optional[str] = None, - recording_date: Optional[str] = None, - recording_time: Optional[str] = None, + room: str | None = None, + recording_datetime: str | None = None, + owner_timezone: str | None = None, ) -> str: """Generate title from context or return default.""" - if not room or not recording_date or not recording_time: + if not room or not recording_datetime: return self._locale.document_default_title + dt = datetime.fromisoformat(recording_datetime) + if owner_timezone: + dt = dt.astimezone(ZoneInfo(owner_timezone)) + return self._locale.document_title_template.format( room=room, - room_recording_date=recording_date, - room_recording_time=recording_time, + room_recording_date=dt.strftime("%Y-%m-%d"), + room_recording_time=dt.strftime("%H:%M"), ) diff --git a/src/summary/tests/api/test_api_tasks.py b/src/summary/tests/api/test_api_tasks.py index c32a96cb..07bffd1b 100644 --- a/src/summary/tests/api/test_api_tasks.py +++ b/src/summary/tests/api/test_api_tasks.py @@ -19,14 +19,14 @@ class TestTasks: headers={"Authorization": "Bearer test-api-token"}, json={ "owner_id": "owner-123", - "filename": "recording.mp4", + "recording_filename": "recording.mp4", + "metadata_filename": "metadata.json", "email": "user@example.com", "sub": "sub-123", "room": "room-abc", - "recording_date": "2026-01-01", - "recording_time": "10:00:00", + "owner_timezone": "UTC", "language": None, - "download_link": "http://example.com/file.mp4", + "download_link": "https://example.com/file.mp4", }, ) @@ -36,16 +36,18 @@ class TestTasks: args = mock_apply_async.call_args.kwargs["args"] assert args == [ "owner-123", # owner_id - "recording.mp4", # filename + "recording.mp4", # recording_filename + "metadata.json", # metadata_filename "user@example.com", # email "sub-123", # sub - 1735725600.0, # frozen time + 1735725600.0, # received_at "room-abc", # room - "2026-01-01", # recording_date - "10:00:00", # recording_time + "UTC", # owner_timezone None, # language - "http://example.com/file.mp4", # download_link + "https://example.com/file.mp4", # download_link None, # context_language + None, # recording_start_at + None, # recording_end_at ] def test_create_task_invalid_language(self, client):