diff --git a/src/backend/core/analytics.py b/src/backend/core/analytics.py new file mode 100644 index 00000000..48f3b34c --- /dev/null +++ b/src/backend/core/analytics.py @@ -0,0 +1,52 @@ +"""Analytics module.""" + +import logging +from enum import StrEnum + +from django.conf import settings + +import posthog + +from core.models import User + +logger = logging.getLogger(__name__) + + +class EventName(StrEnum): + """Analytics event names.""" + + TRANSCRIPT_GENERATION_SUCCESS = "transcript_generation_success" + TRANSCRIPT_GENERATION_FAILURE = "transcript_generation_failure" + SUMMARY_GENERATION_SUCCESS = "summary_generation_success" + SUMMARY_GENERATION_FAILURE = "summary_generation_failure" + + +def capture_event(event_name: EventName, *, user: User, properties=None) -> None: + """ + Capture an analytics event with user properties. + """ + if not settings.POSTHOG_ENABLED: + return + + properties = properties or {} + properties["$set"] = { + "name": user.full_name, + "email": user.email, + "sub": user.sub, + } + posthog.capture(event_name, distinct_id=user.id, properties=properties) + + +def is_feature_enabled(feature_name: str, distinct_id: str) -> bool: + """Check if a feature flag is enabled for a user.""" + if not settings.POSTHOG_ENABLED: + return False + + try: + return posthog.feature_enabled(feature_name, distinct_id) + except Exception as e: # noqa: BLE001 + logger.error("Error checking feature flag %s: %s", feature_name, e) + return False + + +__all__ = ["EventName", "capture_event", "is_feature_enabled"] diff --git a/src/backend/core/api/permissions.py b/src/backend/core/api/permissions.py index c607d80f..df7db217 100644 --- a/src/backend/core/api/permissions.py +++ b/src/backend/core/api/permissions.py @@ -136,3 +136,15 @@ class FilePermission(IsAuthenticated): raise Http404 return obj.get_abilities(request.user).get(view.action, False) + + +class TranscribeWebhookPermission(permissions.BasePermission): + """ + Permissions applying to the summary webhook endpoint. + """ + + def has_permission(self, request, view): + return request.method == "POST" + + def has_object_permission(self, request, view, obj): + return False diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index e883e1ea..d6d43467 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -17,6 +17,7 @@ from django.utils.text import slugify from django.utils.translation import gettext_lazy as _ from django_filters import rest_framework as django_filters +from pydantic import ValidationError from rest_framework import ( decorators, filters, @@ -34,7 +35,7 @@ from rest_framework import ( status as drf_status, ) -from core import enums, models, utils +from core import analytics, enums, models, utils from core.api.filters import ListFileFilter from core.enums import MEDIA_STORAGE_URL_PATTERN from core.recording.enums import FileExtension @@ -80,6 +81,10 @@ from core.services.subtitle import SubtitleException, SubtitleService from core.tasks.file import process_file_deletion from ..authentication.livekit import LiveKitTokenAuthentication +from ..authentication.webhooks import AiWebhookAuthentication +from ..models import AiJobStatusChoices, AiRecordingJob +from ..tasks.ai_job import handle_summary_received, handle_transcript_received +from ..transcription import webhook_schemas from . import permissions, serializers, throttling from .feature_flag import FeatureFlag @@ -915,15 +920,9 @@ class RecordingViewSet( # Attempt to notify external services about the recording # This is a non-blocking operation - failures are logged but don't interrupt the flow - notification_succeeded = notification_service.notify_external_services( - recording - ) + notification_service.notify_external_services(recording) - recording.status = ( - models.RecordingStatusChoices.NOTIFICATION_SUCCEEDED - if notification_succeeded - else models.RecordingStatusChoices.SAVED - ) + recording.status = models.RecordingStatusChoices.SAVED recording.save() return drf_response.Response( @@ -1332,3 +1331,92 @@ class FileViewSet( request = utils.generate_s3_authorization_headers(f"{url_params.get('key'):s}") return drf_response.Response("authorized", headers=request.headers, status=200) + + +class AiJobViewSet( + viewsets.GenericViewSet, +): + """AI jobs API.""" + + permission_classes = [] + serializer_class = None + + def get_queryset(self): + """Restrict AI jobs to current user except webhook endpoint.""" + + raise NotImplementedError() + + @decorators.action( + detail=False, + methods=["post"], + url_path="webhook", + authentication_classes=[AiWebhookAuthentication], + permission_classes=[permissions.TranscribeWebhookPermission], + ) + def on_ai_event(self, request): + """Handle incoming hook events for recordings.""" + logger.debug("Received transcribe webhook event: %s", request.data) + + try: + payload = webhook_schemas.webhook_payload_adapter.validate_python( + request.data + ) + except ValidationError as exc: + logger.error("Invalid webhook payload: %s", exc) + raise drf_exceptions.ValidationError(detail=exc) from exc + + ai_recording_job = AiRecordingJob.objects.filter( + remote_job_id=payload.job_id + ).first() + + if not ai_recording_job: + logger.warning("No AI recording job found for job ID: %s", payload.job_id) + return drf_response.Response( + {"message": "No AI recording job found for job ID, ignoring."}, + ) + + if ai_recording_job.status == AiJobStatusChoices.SUCCESS: + logger.warning( + "AI recording job already in success state for job ID: %s", + payload.job_id, + ) + return drf_response.Response( + {"message": "AI recording job already in success state, ignoring."}, + ) + + if isinstance(payload, webhook_schemas.TranscribeWebhookSuccessPayload): + handle_transcript_received.apply_async( + args=[payload.job_id, payload.transcription_data_url] + ) + elif isinstance(payload, webhook_schemas.SummarizeWebhookSuccessPayload): + handle_summary_received.apply_async( + args=[payload.job_id, payload.summary_data_url] + ) + elif isinstance( + payload, + ( + webhook_schemas.SummarizeWebhookFailurePayload, + webhook_schemas.TranscribeWebhookFailurePayload, + ), + ): + ai_recording_job.status = AiJobStatusChoices.FAILED + ai_recording_job.save() + analytics.capture_event( + analytics.EventName.TRANSCRIPT_GENERATION_FAILURE + if isinstance(payload, webhook_schemas.TranscribeWebhookFailurePayload) + else analytics.EventName.SUMMARY_GENERATION_FAILURE, + user=ai_recording_job.user, + properties={ + "generation_time_seconds": ( + timezone.now() - ai_recording_job.created_at + ).total_seconds(), + "ai_recording_job_id": ai_recording_job.id, + "recording_id": ai_recording_job.recording.id, + }, + ) + else: + raise NotImplementedError() + + return drf_response.Response( + {"message": "Event processed."}, + ) diff --git a/src/backend/core/authentication/webhooks.py b/src/backend/core/authentication/webhooks.py new file mode 100644 index 00000000..354b6704 --- /dev/null +++ b/src/backend/core/authentication/webhooks.py @@ -0,0 +1,34 @@ +"""Webhooks authentication.""" + +import logging + +from django.conf import settings +from django.contrib.auth.models import AnonymousUser + +from rest_framework.authentication import BaseAuthentication +from rest_framework.exceptions import AuthenticationFailed + +logger = logging.getLogger(__name__) + + +class AiWebhookAuthentication(BaseAuthentication): + """ + Custom authentication class for AI webhook requests. + Validates the API key in the Authorization header. + """ + + def authenticate(self, request): + """ + Authenticate the request and return a two-tuple of (user, token). + """ + + authorization_header: str = request.headers.get("Authorization") or "" + if authorization_header.removeprefix("Bearer ") != settings.AI_WEBHOOK_API_KEY: + logger.warning( + "Authentication failed: Bad Authorization header (ip: %s)", + request.META.get("REMOTE_ADDR"), + ) + raise AuthenticationFailed() + + # No users are associated with the transcribe webhooks + return AnonymousUser(), None diff --git a/src/backend/core/migrations/0019_airecordingjob.py b/src/backend/core/migrations/0019_airecordingjob.py new file mode 100644 index 00000000..6a5c168e --- /dev/null +++ b/src/backend/core/migrations/0019_airecordingjob.py @@ -0,0 +1,36 @@ +# Generated by Django 5.2.14 on 2026-05-14 12:40 + +import django.db.models.deletion +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0018_rename_active_application_is_active'), + ] + + operations = [ + migrations.CreateModel( + name='AiRecordingJob', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')), + ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')), + ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')), + ('remote_job_id', models.CharField(blank=True, max_length=255, null=True, unique=True)), + ('type', models.CharField(choices=[('transcript', 'Transcript'), ('summary', 'Summary')], max_length=25)), + ('status', models.CharField(choices=[('pending', 'Pending'), ('success', 'Success'), ('failed', 'Failed')], max_length=25)), + ('language', models.CharField(choices=[('fr', 'fr'), ('en', 'en'), ('de', 'de'), ('nl', 'nl')], default='fr', max_length=2)), + ('docs_app_id', models.CharField(blank=True, max_length=255, null=True)), + ('recording', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ai_jobs', to='core.recording')), + ], + options={ + 'verbose_name': 'AiJob', + 'verbose_name_plural': 'AiJobs', + 'db_table': 'ai_job', + 'ordering': ('created_at',), + 'indexes': [models.Index(fields=['recording', 'type', '-created_at'], name='ai_job_recordi_ca452e_idx')], + }, + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 9e921ae5..ffe72207 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -590,6 +590,16 @@ class Recording(BaseModel): verbose_name=_("Recording options"), help_text=_("Recording options"), ) + started_at = models.DateTimeField( + null=True, + blank=True, + help_text=_("Recording start timestamp as recorded by livekit."), + ) + ended_at = models.DateTimeField( + null=True, + blank=True, + help_text=_("Recording end timestamp as recorded by livekit."), + ) class Meta: db_table = "meet_recording" @@ -736,6 +746,66 @@ class RecordingAccess(BaseAccess): return self._get_abilities(self.recording, user) +class AiJobStatusChoices(models.TextChoices): + """Possible states of a file.""" + + PENDING = "pending", _("Pending") + SUCCESS = "success", _("Success") + FAILED = "failed", _("Failed") + + +class AiJobTypeChoices(models.TextChoices): + """Possible types of Ai Jobs.""" + + TRANSCRIPT = "transcript", _("Transcript") + SUMMARIZE = "summary", _("Summary") + + +class AiRecordingJob(BaseModel): + """ + A job that is run to process an audio file. + """ + + remote_job_id = models.CharField(max_length=255, unique=True, null=True, blank=True) + type = models.CharField( + max_length=25, + choices=AiJobTypeChoices.choices, + ) + recording = models.ForeignKey( + Recording, on_delete=models.CASCADE, related_name="ai_jobs" + ) + status = models.CharField( + max_length=25, + choices=AiJobStatusChoices.choices, + ) + language = models.CharField( + max_length=2, + choices=(("fr", "fr"), ("en", "en"), ("de", "de"), ("nl", "nl")), + default="fr", + ) + docs_app_id = models.CharField(max_length=255, null=True, blank=True) + + class Meta: + db_table = "ai_job" + verbose_name = _("AiJob") + verbose_name_plural = _("AiJobs") + ordering = ("created_at",) + indexes = [ + models.Index(fields=["recording", "type", "-created_at"]), + ] + + def __str__(self): + return f"{self.recording.id} - {self.type} - {self.status}" + + @property + def user(self): + return ( + RecordingAccess.objects.select_related("user") + .filter(role=RoleChoices.OWNER, recording_id=self.recording.id) + .first() + ).user + + class ApplicationScope(models.TextChoices): """Available permission scopes for application operations.""" diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index 2d389c4c..a1c2208b 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -12,11 +12,11 @@ from django.utils.translation import get_language, override from django.utils.translation import gettext_lazy as _ import aiohttp -import requests from asgiref.sync import async_to_sync from livekit import api as livekit_api from core import models, utils +from core.tasks.ai_job import call_transcribe_service logger = logging.getLogger(__name__) @@ -45,22 +45,17 @@ class NotificationService: """Process a recording based on its mode.""" if recording.mode == models.RecordingModeChoices.TRANSCRIPT: - return self._notify_summary_service(recording) - - if recording.mode == models.RecordingModeChoices.SCREEN_RECORDING: - summary_success = True + self._notify_summary_service(recording) + elif recording.mode == models.RecordingModeChoices.SCREEN_RECORDING: if recording.options.get("transcribe", False): - summary_success = self._notify_summary_service(recording) - - email_success = self._notify_user_by_email(recording) - return email_success and summary_success - - logger.error( - "Unknown recording mode %s for recording %s", - recording.mode, - recording.id, - ) - return False + self._notify_summary_service(recording) + self._notify_user_by_email(recording) + else: + logger.error( + "Unknown recording mode %s for recording %s", + recording.mode, + recording.id, + ) @staticmethod def _notify_user_by_email(recording) -> bool: @@ -187,71 +182,17 @@ class NotificationService: or not settings.SUMMARY_SERVICE_API_TOKEN ): logger.error("Summary service not configured") - return False - - owner_access = ( - models.RecordingAccess.objects.select_related("user") - .filter( - role=models.RoleChoices.OWNER, - recording_id=recording.id, - ) - .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 + return started_at, ended_at = async_to_sync( NotificationService._get_recording_timestamps )(recording.worker_id) - payload = { - "owner_id": str(owner_access.user.id), - "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"), - "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), - } + recording.started_at = started_at + recording.ended_at = ended_at + recording.save() - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {settings.SUMMARY_SERVICE_API_TOKEN}", - } - - try: - response = requests.post( - settings.SUMMARY_SERVICE_ENDPOINT, - json=payload, - headers=headers, - timeout=30, - ) - response.raise_for_status() - except requests.RequestException as exc: - logger.exception( - "Summary service error for recording %s. URL: %s. Exception: %s", - recording.id, - settings.SUMMARY_SERVICE_ENDPOINT, - exc, - ) - return False - - return True + call_transcribe_service.apply_async(args=[recording.id]) notification_service = NotificationService() diff --git a/src/backend/core/tasks/__init__.py b/src/backend/core/tasks/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/backend/core/tasks/ai_job.py b/src/backend/core/tasks/ai_job.py new file mode 100644 index 00000000..5a025ce0 --- /dev/null +++ b/src/backend/core/tasks/ai_job.py @@ -0,0 +1,319 @@ +import logging +from datetime import datetime +from urllib.parse import urljoin + +from django.conf import settings +from django.utils import timezone + +import requests + +from core import analytics, models +from core.models import ( + AiJobStatusChoices, + AiJobTypeChoices, + AiRecordingJob, + Recording, +) +from core.tasks._task import task +from core.transcription.locales import get_locale +from core.transcription.transcript_formatter import TranscriptFormatter +from core.transcription.webhook_schemas import WhisperXResponse +from core.utils import generate_download_s3_file_url + +logger = logging.getLogger(__name__) + + +@task +def call_transcribe_service(recording_id): + """ + Call the transcribe service for a given recording. + """ + try: + recording = Recording.objects.get(id=recording_id) + except Recording.DoesNotExist: + logger.error("Recoding %s does not exist", recording_id) + return None + + owner_access = ( + models.RecordingAccess.objects.select_related("user") + .filter( + role=models.RoleChoices.OWNER, + recording_id=recording.id, + ) + .first() + ) + + if not owner_access: + logger.error("No owner found for recording %s", recording.id) + return False + + metadata = None + if ( + settings.METADATA_COLLECTOR_ENABLED + and recording.options.get("collect_metadata", False) + and recording.started_at + and recording.ended_at + ): + output_folder = settings.METADATA_COLLECTOR_OUTPUT_FOLDER + metadata_filename = f"{output_folder}/{recording.id}-metadata.json" + metadata = { + "cloud_storage_url": generate_download_s3_file_url(metadata_filename), + "start_at": recording.started_at, + "end_at": recording.ended_at, + } + + language = ( + recording.options.get("language") or settings.TRANSCRIPTION_DEFAULT_LANGUAGE + ) + ai_transcribe_job = AiRecordingJob.objects.create( + remote_job_id=None, + recording=recording, + type=AiJobTypeChoices.TRANSCRIPT, + status=AiJobStatusChoices.PENDING, + language=language, + ) + + try: + response = requests.post( + settings.AI_SERVICE_URL + "async-jobs/transcribe/", + json={ + "user_sub": owner_access.user.sub, + "language": language, + "cloud_storage_url": generate_download_s3_file_url( + recording.key, expires_in=60 * 60 * 24, override_domain=False + ), + "metadata": metadata, + }, + headers={ + "Authorization": f"Bearer {settings.AI_SERVICE_API_KEY}", + }, + timeout=10, + ) + response.raise_for_status() + except Exception as e: + logger.error( + "Creating transcription job failed for recording %s: %s", recording_id, e + ) + ai_transcribe_job.status = AiJobStatusChoices.FAILED + ai_transcribe_job.save() + raise e + + data = response.json() + + ai_transcribe_job.remote_job_id = data["job_id"] + ai_transcribe_job.save() + + recording.status = models.RecordingStatusChoices.NOTIFICATION_SUCCEEDED + recording.save() + + logger.info("Transcription job created for recording %s", recording_id) + return ai_transcribe_job.id + + +def format_transcript( # noqa: PLR0913 + transcription, + *, + context_language: str | None, + language: str, + room: str | None, + recording_datetime: datetime | None, + owner_timezone: str | None, + download_link: str | None, +) -> tuple[str, str]: + """Format a transcription into readable content with a title. + + Resolves the locale from context_language / language, then uses + TranscriptFormatter to produce markdown content and a title. + + Returns a (content, title) tuple. + """ + locale = get_locale(context_language, language) + formatter = TranscriptFormatter(locale) + + return formatter.format( + transcription, + room=room, + recording_datetime=recording_datetime, + owner_timezone=owner_timezone, + download_link=download_link, + ) + + +@task +def handle_transcript_received(remote_job_id, url): + """ + Store the transcript and call the summarize service for a given recording. + """ + ai_transcript_job = AiRecordingJob.objects.filter( + remote_job_id=remote_job_id, type=AiJobTypeChoices.TRANSCRIPT + ).first() + if not ai_transcript_job: + logger.warning("No AI recording job found for job ID: %s", remote_job_id) + return + + user = ai_transcript_job.user + recording = ai_transcript_job.recording + + response = requests.get(url, timeout=(10, 20)) + response.raise_for_status() + transcript = WhisperXResponse(**response.json()) + + # Format output + content, title = format_transcript( + transcript, + context_language=user.language, + language=ai_transcript_job.language, + room=recording.room.name, + recording_datetime=recording.started_at or recording.created_at, + owner_timezone=user.timezone, + download_link=urljoin(settings.RECORDING_DOWNLOAD_BASE_URL, recording.id), + ) + + create_document_in_docs( + title=title, content=content, email=user.email, sub=user.sub + ) + + ai_transcript_job.status = AiJobStatusChoices.SUCCESS + ai_transcript_job.save() + + analytics.capture_event( + analytics.EventName.TRANSCRIPT_GENERATION_SUCCESS, + user=ai_transcript_job.user, + properties={ + "generation_time_seconds": ( + timezone.now() - ai_transcript_job.created_at + ).total_seconds(), + "ai_recording_job_id": ai_transcript_job.id, + "language": ai_transcript_job.language, + "recording_id": ai_transcript_job.recording.id, + "transcript_size": len(response.content), + }, + ) + + # LLM Summarization + if analytics.is_feature_enabled("summary-enabled", distinct_id=user.sub): + ai_summary_job = AiRecordingJob.objects.create( + remote_job_id=None, + file=recording, + type=AiJobTypeChoices.SUMMARIZE, + status=AiJobStatusChoices.PENDING, + language=ai_transcript_job.language, + ) + + try: + summary_response = requests.post( + settings.AI_SERVICE_URL + "async-jobs/summarize/", + json={ + "user_sub": ai_summary_job.user.sub, + "language": ai_transcript_job.language, + "content": content, + }, + headers={ + "Authorization": f"Bearer {settings.AI_SERVICE_API_KEY}", + }, + timeout=10, + ) + summary_response.raise_for_status() + except Exception as e: + logger.error( + "Creating summary job failed for recording %s: %s", recording.id, e + ) + ai_summary_job.status = AiJobStatusChoices.FAILED + ai_summary_job.save() + raise e + + ai_summary_job.remote_job_id = summary_response.json()["job_id"] + ai_summary_job.save() + + logger.info("Summary job created for recording %s", recording.id) + + +@task +def handle_summary_received(remote_job_id, url): + """ + Store the summary of a given file. + """ + ai_summary_job = AiRecordingJob.objects.filter( + remote_job_id=remote_job_id, type=AiJobTypeChoices.SUMMARIZE + ).first() + if not ai_summary_job: + logger.warning("No AI file job found for job ID: %s", remote_job_id) + return + + recording = ai_summary_job.recording + + logger.info("Storing summary for recording %s & url %s", recording.id, url) + response = requests.get(url, timeout=(10, 20)) + response.raise_for_status() + + user = ai_summary_job.user + + # We dynamically recompute the title for the document since we don't have access to the transcript + _, title = format_transcript( + None, + context_language=user.language, + language=ai_summary_job.language, + room=recording.room.name, + recording_datetime=recording.started_at or recording.created_at, + owner_timezone=user.timezone, + download_link=urljoin(settings.RECORDING_DOWNLOAD_BASE_URL, recording.id), + ) + + create_document_in_docs( + title=get_locale(user.language).summary_title_template.format(title=title), + content=response.text, + email=user.email, + sub=user.sub, + ) + + logger.info("Summary created in docs for recording %s & url %s", recording.id, url) + ai_summary_job.status = AiJobStatusChoices.SUCCESS + ai_summary_job.save() + + analytics.capture_event( + analytics.EventName.TRANSCRIPT_GENERATION_SUCCESS, + user=ai_summary_job.user, + properties={ + "generation_time_seconds": ( + timezone.now() - ai_summary_job.created_at + ).total_seconds(), + "ai_recording_job_id": ai_summary_job.id, + "language": ai_summary_job.language, + "recording_id": ai_summary_job.recording.id, + "transcript_size": len(response.content), + }, + ) + + +def create_document_in_docs(*, title: str, content: str, email: str, sub: str) -> str: + """ + Create a document in Docs for a given file. + """ + + response = requests.post( + urljoin(settings.DOCS_BASE_URL, "/api/v1.0/documents/create-for-owner/"), + json={ + "title": title, + "content": content, + "email": email, + "sub": sub, + }, + headers={ + "Authorization": f"Bearer {settings.DOCS_SERVER_TO_SERVER_API_KEY}", + }, + timeout=20, + ) + + if response.status_code != 201: + logger.error( + "Failed to create document in Docs %s", + title, + ) + response.raise_for_status() + + docs_app_id = response.json()["id"] + logger.info( + "Document created in Docs => %s (in docs)", + docs_app_id, + ) + return docs_app_id diff --git a/src/backend/core/transcription/locales/de.py b/src/backend/core/transcription/locales/de.py index eb64cd62..911fbcac 100644 --- a/src/backend/core/transcription/locales/de.py +++ b/src/backend/core/transcription/locales/de.py @@ -30,4 +30,5 @@ Einige Punkte, die wir Ihnen empfehlen zu überprüfen: document_title_template=( 'Besprechung "{room}" am {room_recording_date} um {room_recording_time}' ), + summary_title_template="Zusammenfassung von {title}", ) diff --git a/src/backend/core/transcription/locales/en.py b/src/backend/core/transcription/locales/en.py index a9ea3f15..b7aec97c 100644 --- a/src/backend/core/transcription/locales/en.py +++ b/src/backend/core/transcription/locales/en.py @@ -30,4 +30,5 @@ A few things we recommend you check: document_title_template=( 'Meeting "{room}" on {room_recording_date} at {room_recording_time}' ), + summary_title_template="Summary of {title}", ) diff --git a/src/backend/core/transcription/locales/fr.py b/src/backend/core/transcription/locales/fr.py index c170c5b8..0db7e3bf 100644 --- a/src/backend/core/transcription/locales/fr.py +++ b/src/backend/core/transcription/locales/fr.py @@ -30,4 +30,5 @@ Quelques points que nous vous conseillons de vérifier : document_title_template=( 'Réunion "{room}" du {room_recording_date} à {room_recording_time}' ), + summary_title_template="Résumé de {title}", ) diff --git a/src/backend/core/transcription/locales/nl.py b/src/backend/core/transcription/locales/nl.py index 667f31ad..c599ac53 100644 --- a/src/backend/core/transcription/locales/nl.py +++ b/src/backend/core/transcription/locales/nl.py @@ -30,4 +30,5 @@ Een paar punten die wij u aanraden te controleren: document_title_template=( 'Vergadering "{room}" op {room_recording_date} om {room_recording_time}' ), + summary_title_template="Samenvatting van {title}", ) diff --git a/src/backend/core/transcription/locales/strings.py b/src/backend/core/transcription/locales/strings.py index c182cfdf..ee8b2c9b 100644 --- a/src/backend/core/transcription/locales/strings.py +++ b/src/backend/core/transcription/locales/strings.py @@ -13,3 +13,4 @@ class LocaleStrings: hallucination_replacement_text: str document_default_title: str document_title_template: str + summary_title_template: str diff --git a/src/backend/core/transcription/transcript_formatter.py b/src/backend/core/transcription/transcript_formatter.py index 2f88d9ea..3d46e7ef 100644 --- a/src/backend/core/transcription/transcript_formatter.py +++ b/src/backend/core/transcription/transcript_formatter.py @@ -41,7 +41,7 @@ class TranscriptFormatter: self, transcription, room: str | None = None, - recording_datetime: str | None = None, + recording_datetime: datetime | None = None, owner_timezone: str | None = None, download_link: str | None = None, ) -> Tuple[str, str]: @@ -99,14 +99,14 @@ class TranscriptFormatter: def _generate_title( self, room: str | None = None, - recording_datetime: str | None = None, + recording_datetime: datetime | None = None, owner_timezone: str | None = None, ) -> str: """Generate title from context or return default.""" if not room or not recording_datetime: return self._locale.document_default_title - dt = datetime.fromisoformat(recording_datetime) + dt = recording_datetime if owner_timezone: dt = dt.astimezone(ZoneInfo(owner_timezone)) diff --git a/src/backend/core/transcription/webhook_schemas.py b/src/backend/core/transcription/webhook_schemas.py new file mode 100644 index 00000000..2f3caf79 --- /dev/null +++ b/src/backend/core/transcription/webhook_schemas.py @@ -0,0 +1,163 @@ +"""Transcribe / summary Shared / Webhook models.""" + +from typing import Annotated, Literal, Union + +from pydantic import BaseModel, Field, TypeAdapter + + +class WordSegment(BaseModel): + """Word segment model for transcription tasks.""" + + word: str = Field(title="Word") + start: float | None = Field( + default=None, title="Start Time", description="Start time in seconds." + ) + end: float | None = Field( + default=None, title="End Time", description="End time in seconds." + ) + score: float | None = Field( + default=None, + title="Confidence Score", + description="Confidence score for the word segment.", + ) + speaker: str | None = Field( + default=None, + title="Speaker", + description="Speaker identifier for the word segment.", + ) + + +class Segment(BaseModel): + """Segment model for transcription tasks.""" + + start: float | None = Field( + default=None, title="Start Time", description="Start time in seconds." + ) + end: float | None = Field( + default=None, title="End Time", description="End time in seconds." + ) + text: str = Field( + title="Segment Text", description="Transcribed text for the segment." + ) + words: tuple[WordSegment, ...] | None = Field( + title="Word Segments", description="List of word segments within the segment." + ) + speaker: str | None = Field( + default=None, title="Speaker", description="Speaker identifier for the segment." + ) + + +class WhisperXResponse(BaseModel): + """Model for WhisperX response.""" + + segments: tuple[Segment, ...] = Field( + title="Segments", description="List of transcribed segments." + ) + word_segments: tuple[WordSegment, ...] = Field( + title="Word Segments", description="List of word segments." + ) + + +class BaseWebhook(BaseModel): + """Base webhook payload.""" + + job_id: str = Field( + title="Job ID", + description="The ID of the job document in the receiver system.", + ) + + +class TranscribeWebhookSuccessPayload(BaseWebhook): + """Payload for a successful transcription webhook.""" + + type: Literal["transcript"] = Field(default="transcript") + status: Literal["success"] = Field(default="success") + transcription_data_url: str = Field( + title="Transcript", description="URL to the raw transcription data." + ) + + +class TranscribeWebhookPendingPayload(BaseWebhook): + """Payload for a pending transcription webhook-like response.""" + + type: Literal["transcript"] = Field(default="transcript") + status: Literal["pending"] = Field(default="pending") + + +class TranscribeWebhookFailurePayload(BaseWebhook): + """Payload for a failed transcription webhook.""" + + type: Literal["transcript"] = Field(default="transcript") + status: Literal["failure"] = Field(default="failure") + error_code: Literal["unknown_error"] = Field( + title="Error code", description="The error code." + ) + + +TranscribeWebhookPayloads = Annotated[ + Union[ + TranscribeWebhookSuccessPayload, + TranscribeWebhookPendingPayload, + TranscribeWebhookFailurePayload, + ], + Field(discriminator="status"), +] + + +class SummarizeWebhookSuccessPayload(BaseWebhook): + """Payload for a successful summarization webhook.""" + + type: Literal["summary"] = Field(default="summary") + status: Literal["success"] = Field(default="success") + summary_data_url: str = Field( + title="Summary", description="URL to the raw summary data." + ) + + +class SummarizeWebhookPendingPayload(BaseWebhook): + """Payload for a pending summarization webhook-like response.""" + + type: Literal["summary"] = Field(default="summary") + status: Literal["pending"] = Field(default="pending") + + +class SummarizeWebhookFailurePayload(BaseWebhook): + """Payload for a failed summarization webhook.""" + + type: Literal["summary"] = Field(default="summary") + status: Literal["failure"] = Field(default="failure") + error_code: Literal["unknown_error"] = Field( + title="Error code", description="The error code." + ) + + +SummarizeWebhookPayloads = Annotated[ + Union[ + SummarizeWebhookSuccessPayload, + SummarizeWebhookPendingPayload, + SummarizeWebhookFailurePayload, + ], + Field(discriminator="status"), +] + +WebhookPayloads = Annotated[ + Union[TranscribeWebhookPayloads, SummarizeWebhookPayloads], + Field(discriminator="type"), +] + + +webhook_payload_adapter = TypeAdapter(WebhookPayloads) + +__all__ = [ + "TranscribeWebhookSuccessPayload", + "TranscribeWebhookPendingPayload", + "TranscribeWebhookFailurePayload", + "SummarizeWebhookSuccessPayload", + "SummarizeWebhookPendingPayload", + "SummarizeWebhookFailurePayload", + "TranscribeWebhookPayloads", + "SummarizeWebhookPayloads", + "WebhookPayloads", + "WhisperXResponse", + "webhook_payload_adapter", +] diff --git a/src/backend/core/urls.py b/src/backend/core/urls.py index 3bc4f5e8..aebe5eed 100644 --- a/src/backend/core/urls.py +++ b/src/backend/core/urls.py @@ -16,6 +16,7 @@ router.register("users", viewsets.UserViewSet, basename="users") router.register("rooms", viewsets.RoomViewSet, basename="rooms") router.register("recordings", viewsets.RecordingViewSet, basename="recordings") router.register("files", viewsets.FileViewSet, basename="files") +router.register("ai-jobs", viewsets.AiJobViewSet, basename="ai-jobs") router.register( "resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses" ) diff --git a/src/backend/core/utils.py b/src/backend/core/utils.py index cbacbe1e..e992bf95 100644 --- a/src/backend/core/utils.py +++ b/src/backend/core/utils.py @@ -455,3 +455,38 @@ def generate_upload_policy(file): ) return policy + + +def generate_download_s3_file_url( + key, *, expires_in: int, override_domain: bool = True +): + """ + Generate a S3 signed download url for a given key. + """ + + # This settings should be used if the backend application and the frontend application + # can't connect to the object storage with the same domain. This is the case in the + # docker compose stack used in development. The frontend application will use localhost + # to connect to the object storage while the backend application will use the object storage + # service name declared in the docker compose stack. + # This is needed because the domain name is used to compute the signature. So it can't be + # changed dynamically by the frontend application. + if settings.AWS_S3_DOMAIN_REPLACE and override_domain: + s3_client = boto3.client( + "s3", + aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID, + aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY, + endpoint_url=settings.AWS_S3_DOMAIN_REPLACE, + config=botocore.client.Config( + region_name=settings.AWS_S3_REGION_NAME, + signature_version=settings.AWS_S3_SIGNATURE_VERSION, + ), + ) + else: + s3_client = default_storage.connection.meta.client + + return s3_client.generate_presigned_url( + ClientMethod="get_object", + Params={"Bucket": default_storage.bucket_name, "Key": key}, + ExpiresIn=expires_in, + ) diff --git a/src/backend/meet/settings.py b/src/backend/meet/settings.py index 9db07e76..7442e815 100755 --- a/src/backend/meet/settings.py +++ b/src/backend/meet/settings.py @@ -19,6 +19,7 @@ from socket import gethostbyname, gethostname from django.utils.translation import gettext_lazy as _ import dj_database_url +import posthog import sentry_sdk from configurations import Configuration, values from lasuite.configuration.values import SecretFileValue @@ -451,6 +452,11 @@ class Base(Configuration): CELERY_BROKER_URL = values.Value("redis://redis:6379/0", environ_prefix=None) CELERY_BROKER_TRANSPORT_OPTIONS = values.DictValue({}, environ_prefix=None) + # Analytics + POSTHOG_ENABLED = values.BooleanValue(False, environ_prefix=None) + POSTHOG_API_KEY = values.Value(None, environ_prefix=None) + POSTHOG_API_HOST = values.Value(None, environ_prefix=None) + # Session SESSION_ENGINE = values.Value( default="django.contrib.sessions.backends.cache", @@ -744,6 +750,16 @@ class Base(Configuration): SUMMARY_SERVICE_API_TOKEN = SecretFileValue( None, environ_name="SUMMARY_SERVICE_API_TOKEN", environ_prefix=None ) + DOCS_BASE_URL = values.Value( + "https://example.com", + environ_name="DOCS_BASE_URL", + environ_prefix=None, + ) + DOCS_SERVER_TO_SERVER_API_KEY = SecretFileValue( + None, + environ_name="DOCS_SERVER_TO_SERVER_API_KEY", + environ_prefix=None, + ) TRANSCRIPTION_DEFAULT_LANGUAGE = values.Value( default="fr", environ_name="TRANSCRIPTION_DEFAULT_LANGUAGE", environ_prefix=None ) diff --git a/src/backend/pyproject.toml b/src/backend/pyproject.toml index 01b4a3aa..caddf227 100644 --- a/src/backend/pyproject.toml +++ b/src/backend/pyproject.toml @@ -62,6 +62,7 @@ dependencies = [ "livekit-api==1.1.0", "aiohttp==3.13.4", "urllib3==2.7.0", + "posthog>=7.14.2", ] [project.urls] diff --git a/src/backend/uv.lock b/src/backend/uv.lock index 9ce7c36e..357d9163 100644 --- a/src/backend/uv.lock +++ b/src/backend/uv.lock @@ -148,6 +148,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" }, ] +[[package]] +name = "backoff" +version = "2.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/47/d7/5bbeb12c44d7c4f2fb5b56abce497eb5ed9f34d85701de869acedd602619/backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba", size = 17001, upload-time = "2022-10-05T19:19:32.061Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, +] + [[package]] name = "billiard" version = "4.2.4" @@ -559,6 +568,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1e/77/dc8c558f7593132cf8fefec57c4f60c83b16941c574ac5f619abb3ae7933/dill-0.4.1-py3-none-any.whl", hash = "sha256:1e1ce33e978ae97fcfcff5638477032b801c46c7c65cf717f95fbc2248f79a9d", size = 120019, upload-time = "2026-01-19T02:36:55.663Z" }, ] +[[package]] +name = "distro" +version = "1.9.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/fc/f8/98eea607f65de6527f8a2e8885fc8015d3e6f5775df186e443e0964a11c3/distro-1.9.0.tar.gz", hash = "sha256:2fa77c6fd8940f116ee1d6b94a2f90b13b5ea8d019b98bc8bafdcabcdd9bdbed", size = 60722, upload-time = "2023-12-24T09:54:32.31Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" }, +] + [[package]] name = "dj-database-url" version = "3.1.2" @@ -1204,6 +1222,7 @@ dependencies = [ { name = "markdown" }, { name = "mozilla-django-oidc" }, { name = "nested-multipart-parser" }, + { name = "posthog" }, { name = "psycopg", extra = ["binary"] }, { name = "pydantic" }, { name = "pyjwt" }, @@ -1266,6 +1285,7 @@ requires-dist = [ { name = "markdown", specifier = "==3.10.2" }, { name = "mozilla-django-oidc", specifier = "==5.0.2" }, { name = "nested-multipart-parser", specifier = "==1.6.0" }, + { name = "posthog", specifier = ">=7.14.2" }, { name = "psycopg", extras = ["binary"], specifier = "==3.3.3" }, { name = "pydantic", specifier = "==2.12.5" }, { name = "pyjwt", specifier = "==2.12.1" }, @@ -1506,6 +1526,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, ] +[[package]] +name = "posthog" +version = "7.14.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "backoff" }, + { name = "distro" }, + { name = "requests" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ca/e6/fe25f9eaeb5b4b66aa738554ba1ef9feece8b1d5b6a9ea431782b6c6c58f/posthog-7.14.2.tar.gz", hash = "sha256:b913dc23acc301a95ca9b851c193b261932d01a66a9af91eb6e9883cd05d5b6b", size = 205633, upload-time = "2026-05-13T16:36:27.153Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/b8/75f9eed446c1d48405871542b36fdf1dba3333756a477bcb0d9ef70d17a4/posthog-7.14.2-py3-none-any.whl", hash = "sha256:f78b45a5ad5c72e55bf1cadebe8cf46bae2a6094e7fe01cbc7b5ec0531aab4d8", size = 240750, upload-time = "2026-05-13T16:36:25.544Z" }, +] + [[package]] name = "pprintpp" version = "0.4.0"