diff --git a/src/summary/summary/api/route/tasks_v2.py b/src/summary/summary/api/route/tasks_v2.py index e8f8cc78..eebe88e8 100644 --- a/src/summary/summary/api/route/tasks_v2.py +++ b/src/summary/summary/api/route/tasks_v2.py @@ -1,15 +1,19 @@ """API routes related to application tasks (V2 / tenant friendly).""" -from celery.result import AsyncResult -from fastapi import APIRouter, Depends, HTTPException, Request +import logging +from datetime import datetime, timezone +from celery.result import AsyncResult +from fastapi import APIRouter, Depends, HTTPException, Request, status + +from summary.core.analytics import get_analytics from summary.core.celery_worker import ( celery, process_audio_transcribe_v2_task, summarize_v2_task, ) -from summary.core.config import AuthorizedTenant -from summary.core.models import SummarizeTaskV2Request, TranscribeTaskV2Request +from summary.core.config import AuthorizedTenant, get_settings +from summary.core.models import SummarizeTaskApiRequest, TranscribeTaskApiRequest from summary.core.security import verify_tenant_api_key_v2 from summary.core.shared_models import ( SummarizeWebhookFailurePayload, @@ -20,17 +24,52 @@ from summary.core.shared_models import ( TranscribeWebhookSuccessPayload, ) +logger = logging.getLogger(__name__) router_tasks_v2 = APIRouter() +analytics = get_analytics() +settings = get_settings() + @router_tasks_v2.post("/async-jobs/transcribe") async def create_transcribe_task_v2( - request: TranscribeTaskV2Request, + request: TranscribeTaskApiRequest, request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), ): """Create a transcription task.""" + if ( + request.push_to_docs_config is not None + and not request_tenant.allowed_push_to_docs + ): + logger.warning( + "Push to docs is not allowed for this tenant (%s).", request_tenant.id + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Push to docs is not allowed for this tenant.", + ) + task = process_audio_transcribe_v2_task.apply_async( - args=[{**request.model_dump(), "tenant_id": request_tenant.id}] + args=[ + { + **request.model_dump(), + "tenant_id": request_tenant.id, + "received_at": datetime.now(timezone.utc), + } + ] + ) + + # We track the request, this also properly initializes the user in the + # analytics system, so that later feature flags work properly + analytics.capture( + settings.posthog_event_request, + request.user_sub, + properties={ + "kind": "transcribe", + "$set": { + "email": request.user_email, + }, + }, ) return TranscribeWebhookPendingPayload(job_id=task.id).model_dump() @@ -38,14 +77,32 @@ async def create_transcribe_task_v2( @router_tasks_v2.post("/async-jobs/summarize") async def create_summarize_task_v2( - request: SummarizeTaskV2Request, + request: SummarizeTaskApiRequest, request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), ): """Create a summarization task.""" task = summarize_v2_task.apply_async( - args=[{**request.model_dump(), "tenant_id": request_tenant.id}] + args=[ + { + **request.model_dump(), + "tenant_id": request_tenant.id, + "received_at": datetime.now(timezone.utc), + } + ] ) + # We track the request, this also properly initializes the user in the + # analytics system, so that later feature flags work properly + analytics.capture( + settings.posthog_event_request, + request.user_sub, + properties={ + "kind": "summarize", + "$set": { + "email": request.user_email, + }, + }, + ) return SummarizeWebhookPendingPayload(job_id=task.id).model_dump() diff --git a/src/summary/summary/core/analytics.py b/src/summary/summary/core/analytics.py index cffe1a37..fb52ef8a 100644 --- a/src/summary/summary/core/analytics.py +++ b/src/summary/summary/core/analytics.py @@ -4,12 +4,14 @@ import json import time from collections import Counter from functools import lru_cache +from urllib.parse import urlsplit, urlunsplit import redis from celery.utils.log import get_task_logger from posthog import Posthog from summary.core.config import get_settings +from summary.core.models import TranscribeTaskJob logger = get_task_logger(__name__) settings = get_settings() @@ -107,23 +109,24 @@ class MetadataManager: """Check if task_id exists in tasks metadata cache.""" return self._redis.exists(self._get_redis_key(task_id)) - def create(self, task_id, task_args): + def create(self, task_id: str, task_payload: TranscribeTaskJob): """Create initial metadata entry for a new task.""" if self._is_disabled or self.has_task_id(task_id): return - # 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() + parts = urlsplit(task_payload.cloud_storage_url) + clean_url = urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) initial_metadata = { "start_time": start_time, "asr_model": settings.whisperx_asr_model, "retries": 0, - "filename": filename, - "email": email, - "queuing_time": round(start_time - received_at, 2), + "filename": clean_url, + "sub": task_payload.user_sub, + # avoid None in redis, it shouldn't happen anyway in prod + "email": task_payload.user_email or "", + "tenant_id": task_payload.tenant_id, + "queuing_time": round(start_time - task_payload.received_at.timestamp(), 2), } self._save_metadata(task_id, initial_metadata) @@ -210,6 +213,6 @@ class MetadataManager: self.clear(task_id) try: - self._analytics.capture(event_name, metadata.get("email"), metadata) + self._analytics.capture(event_name, metadata.get("sub"), metadata) except AnalyticsException: logger.exception("Failed to capture analytics event") diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index ff803cac..091c5fae 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -4,7 +4,7 @@ import json import time -from datetime import datetime +from datetime import datetime, timezone from typing import Any from urllib.parse import urljoin @@ -17,16 +17,15 @@ from requests import exceptions from summary.core.analytics import MetadataManager, get_analytics from summary.core.config import get_settings -from summary.core.file_service import ( - FileService, - FileServiceException, - TranscribeError, -) +from summary.core.docs_service import create_document_in_lasuite_docs +from summary.core.file_service import FileService, FileServiceException, TranscribeError from summary.core.llm_service import LLMException, LLMObservability, LLMService from summary.core.locales import get_locale from summary.core.models import ( - SummarizeTaskV2Payload, - TranscribeTaskV2Payload, + PushToDocsBaseConfig, + RecordingMetadata, + SummarizeTaskJob, + TranscribeTaskJob, ) from summary.core.prompt import ( FORMAT_NEXT_STEPS, @@ -85,29 +84,22 @@ file_service = FileService() def transcribe_audio( *, task_id: str, - recording_filename: str | None = None, language: str, - cloud_storage_url=None, + cloud_storage_url: str, raises: bool = False, ): """Transcribe an audio file using WhisperX. - Downloads the audio from MinIO or a cloud storage URL, sends it to + Downloads the audio from a cloud storage URL, sends it to WhisperX for transcription, and tracks metadata throughout the process. Returns the transcription object, or None if the file could not be retrieved. """ - if bool(recording_filename) == bool(cloud_storage_url): - raise ValueError( - "Either filename or cloud_storage_url must be provided, but not both." - ) - logger.info("Initiating WhisperX client") # Transcription try: with file_service.prepare_audio_file( - remote_object_key=recording_filename, cloud_storage_url=cloud_storage_url, ) as (audio_file, metadata): metadata_manager.track(task_id, {"audio_length": metadata["duration"]}) @@ -196,11 +188,7 @@ def transcribe_audio( cloud_storage_url.split("?", 1)[0] if cloud_storage_url else None ) logger.exception( - ( - "Unexpected error while preparing file | filename: %s " - "| cloud_storage_url: %s" - ), - recording_filename, + ("Unexpected error while preparing file %s "), redacted_cloud_storage_url, ) return None @@ -210,44 +198,34 @@ def transcribe_audio( def resolve_speaker_identities_and_apply_to( - transcription, recording_start_at, recording_end_at, metadata_filename, task_id -): + *, transcription: WhisperXResponse, recording_metadata: RecordingMetadata, task_id +) -> WhisperXResponse: """Assign users to detected speakers and rewrite the transcriptions. Args: transcription: output of meet-whisperx after transcription and diarization - recording_start_at: sourced from LiveKit FileInfo via the egress_ended webhook - recording_end_at: sourced from LiveKit FileInfo via the egress_ended webhook - metadata_filename: name of metadata file containing VAD information in S3 + recording_metadata: Metadata of the recording task_id: current task id, for logging purposes """ - recording_start_dt = ( - datetime.fromisoformat(recording_start_at) if recording_start_at else None - ) - recording_end_dt = ( - datetime.fromisoformat(recording_end_at) if recording_end_at else None - ) - logger.debug( "recording_start_dt: %s ; recording_end_dt: %s", - recording_start_dt, - recording_end_dt, + recording_metadata.started_at, + recording_metadata.ended_at, ) - if (recording_start_dt is None) or (recording_end_dt is None): - logger.debug("Skipping resolve_speaker_identities") - return transcription logger.debug("Running resolve_speaker_identities") try: - metadata = file_service.read_json(metadata_filename) + metadata = file_service.read_cloud_storage_json( + recording_metadata.cloud_storage_url + ) speaker_mapping = resolve_speaker_identities( metadata, - transcription, - recording_start_dt, - recording_end_dt, + transcription.model_dump(), + recording_metadata.started_at, + recording_metadata.ended_at, ) new_transcription = speaker_mapping.apply_to(transcription.model_dump()) - return new_transcription + return WhisperXResponse.model_validate(new_transcription) except FileServiceException as exc: logger.error( @@ -272,12 +250,9 @@ def format_transcript( transcription, context_language: str | None, language: str, - room: str | None, - recording_datetime: str | None, - owner_timezone: str | None, download_link: str | None, form_link: str | None, -) -> tuple[str, str]: +) -> str: """Format a transcription into readable content with a title. Resolves the locale from context_language / language, then uses @@ -290,9 +265,6 @@ def format_transcript( return formatter.format( transcription, - room=room, - recording_datetime=recording_datetime, - owner_timezone=owner_timezone, download_link=download_link, form_link=form_link, ) @@ -317,7 +289,7 @@ def format_actions(llm_output: dict) -> str: def summarize_transcription_internals( - *, owner_id: str, transcript: str, session_id: str + *, distinct_id: str, transcript: str, session_id: str ) -> str: """Generate a summary from the provided transcription text. @@ -328,11 +300,11 @@ def summarize_transcription_internals( """ logger.info( "Starting summarization task | Owner: %s", - owner_id, + distinct_id, ) user_has_tracing_consent = analytics.is_feature_enabled( - "summary-tracing-consent", distinct_id=owner_id + "summary-tracing-consent", distinct_id=distinct_id ) # NOTE: We must instantiate a new LLMObservability client for each task invocation @@ -343,7 +315,7 @@ def summarize_transcription_internals( llm_observability = LLMObservability( user_has_tracing_consent=user_has_tracing_consent, session_id=session_id, - user_id=owner_id, + user_id=distinct_id, ) llm_service = LLMService(llm_observability=llm_observability) @@ -400,6 +372,52 @@ def summarize_transcription_internals( ################################################################################## +def _should_push_to_docs( + payload: TranscribeTaskJob | SummarizeTaskJob, +) -> bool: + """Determines if the transcription should be pushed to docs. + + Based on the payload and settings. + """ + if not payload.push_to_docs_config: + reason = "Push to docs is not requested in the payload" + elif not settings.is_lasuite_docs_integration_enabled: + reason = "Docs integration is disabled" + elif not settings.get_authorized_tenant( + tenant_id=payload.tenant_id + ).allowed_push_to_docs: + reason = "Tenant is not allowed to push to docs" + else: + return True + + logger.info("Push to docs is not requested: %s", reason) + return False + + +def _should_auto_create_summary(payload: TranscribeTaskJob) -> bool: + """Determines if the transcription should have an auto-created summary. + + Based on the payload and settings. + """ + if ( + payload.push_to_docs_config is None + or not payload.push_to_docs_config.auto_create_summary + ): + reason = "Auto create summary is not requested in the payload" + elif not settings.is_summary_enabled: + reason = "Summary feature is disabled" + elif not analytics.is_feature_enabled( + "summary-enabled", + distinct_id=payload.user_sub, + ): + reason = "Summary feature flag return false" + else: + return True + + logger.info("Auto create summary is not requested: %s", reason) + return False + + @celery.task( max_retries=3, queue=settings.call_webhook_queue_v2, @@ -436,7 +454,7 @@ def process_audio_transcribe_v2_task( self: Celery task instance (passed on with bind=True) payload: Serialized dictionary of TranscribeSummarizeTaskCreationV2 """ - payload = TranscribeTaskV2Payload.model_validate(payload) + payload = TranscribeTaskJob.model_validate(payload) logger.info( "Transcribing for object received | Owner: %s", payload.user_sub, @@ -463,6 +481,60 @@ def process_audio_transcribe_v2_task( ) return failure_payload.model_dump() + # Assign speakers and rewrite transcription/diarization output + if settings.is_resolve_speaker_identities_enabled and payload.metadata is not None: + try: + transcription_res = resolve_speaker_identities_and_apply_to( + transcription=transcription_res, + recording_metadata=payload.metadata, + task_id=job_id, + ) + except Exception as e: + logger.error(f"Failed to resolve speaker identities, skipping: {e}") + + should_push_to_docs = _should_push_to_docs(payload) + # We do it synchronously for now + if should_push_to_docs: + if payload.push_to_docs_config is None: + raise ValueError("Push to docs config is missing") + + # Format output + content = format_transcript( + transcription_res.model_dump(), + payload.context_language, + payload.language, + payload.push_to_docs_config.download_link, + payload.push_to_docs_config.form_link, + ) + + create_document_in_lasuite_docs( + content=content, + title=payload.push_to_docs_config.title, + email=payload.push_to_docs_config.user_email, + sub=payload.user_sub, + ) + + if _should_auto_create_summary(payload): + locale = get_locale(payload.context_language, payload.language) + + summarize_v2_task.apply_async( + args=[ + SummarizeTaskJob( + received_at=datetime.now(timezone.utc), + tenant_id=payload.tenant_id, + user_sub=payload.user_sub, + user_email=payload.user_email, + push_to_docs_config=PushToDocsBaseConfig( + user_email=payload.push_to_docs_config.user_email, + title=locale.summary_title_template.format( + title=payload.push_to_docs_config.title + ), + ), + content=content, + ).model_dump() + ], + ) + file_service.store_transcript( transcript=transcription_res, job_id=job_id, @@ -475,9 +547,30 @@ def process_audio_transcribe_v2_task( call_webhook_v2_task.apply_async( args=[success_payload.model_dump(), payload.tenant_id] ) + metadata_manager.capture(job_id, settings.posthog_event_success) + return success_payload.model_dump() +@signals.task_prerun.connect(sender=process_audio_transcribe_v2_task) +def task_started(task_id=None, task=None, args=None, **kwargs): + """Signal handler called before task execution begins.""" + if args: + metadata_manager.create(task_id, TranscribeTaskJob.model_validate(args[0])) + + +@signals.task_retry.connect(sender=process_audio_transcribe_v2_task) +def task_retry_handler(request=None, reason=None, einfo=None, **kwargs): + """Signal handler called when task execution retries.""" + metadata_manager.retry(request.id) + + +@signals.task_failure.connect(sender=process_audio_transcribe_v2_task) +def task_failure_handler(task_id, exception=None, **kwargs): + """Signal handler called when task execution fails permanently.""" + metadata_manager.capture(task_id, settings.posthog_event_failure) + + @signals.task_failure.connect(sender=process_audio_transcribe_v2_task) def handle_transcribe_v2_failed( sender, @@ -533,15 +626,26 @@ def summarize_v2_task( 1. Run summary internals 2. Sends the final summary via webhook. """ - payload = SummarizeTaskV2Payload.model_validate(payload) + payload = SummarizeTaskJob.model_validate(payload) summary = summarize_transcription_internals( - owner_id=payload.user_sub, + distinct_id=payload.user_sub, transcript=payload.content, session_id=self.request.id, ) job_id = self.request.id file_service.store_summary(summary=summary, job_id=job_id) + if _should_push_to_docs(payload): + if payload.push_to_docs_config is None: + raise ValueError("Push to docs config is missing") + + create_document_in_lasuite_docs( + content=summary, + title=payload.push_to_docs_config.title, + email=payload.push_to_docs_config.user_email, + sub=payload.user_sub, + ) + success_payload = SummarizeWebhookSuccessPayload( job_id=job_id, summary_data_url=file_service.get_summary_signed_url(job_id), diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index c3089cf8..8ee4eabf 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -1,9 +1,8 @@ """Application configuration and settings.""" import logging -import os from functools import cached_property, lru_cache -from typing import Annotated, Any, List, Literal, Mapping, Optional, Set +from typing import Annotated, List, Mapping, Optional, Set from fastapi import Depends from pydantic import ( @@ -34,6 +33,12 @@ class AuthorizedTenant(BaseModel): title="Webhook API Key", description="The api_key to authenticate the webhook request.", ) + allowed_push_to_docs: bool = Field( + title="Allow Push to Docs", + description="Whether to allow pushing transcript" + " and summaries to docs for this tenant.", + default=False, + ) class Settings(BaseSettings): @@ -83,7 +88,7 @@ class Settings(BaseSettings): # AI-related settings whisperx_api_key: SecretStr - whisperx_base_url: str = "https://api.openai.com/v1" + whisperx_base_url: Url = "https://api.openai.com/v1" whisperx_asr_model: str = "whisper-1" # ISO 639-1 language code (e.g., "en", "fr", "es") whisperx_default_language: Optional[str] = None @@ -105,10 +110,17 @@ class Settings(BaseSettings): webhook_max_retries: int = 2 webhook_status_forcelist: List[int] = [502, 503, 504] webhook_backoff_factor: float = 0.1 + app_external_user_agent: str = "summary" # Summary related settings is_summary_enabled: bool = True - transcription_satisfaction_form_base_url: Optional[str] = None + + # Docs service configuration + is_lasuite_docs_integration_enabled: bool = False + lasuite_docs_base_url: Url = "https://example.com" + lasuite_docs_server_to_server_api_key: SecretStr = Field( + title="API key for using docs server to server api", default="NO_API_KEY" + ) # Sentry sentry_is_enabled: bool = False @@ -120,6 +132,7 @@ class Settings(BaseSettings): posthog_api_host: Optional[str] = "https://eu.i.posthog.com" posthog_event_failure: str = "transcript-failure" posthog_event_success: str = "transcript-success" + posthog_event_request: str = "transcript-request" # Langfuse (LLM Observability) langfuse_enabled: bool = False @@ -148,6 +161,25 @@ class Settings(BaseSettings): raise ValueError("Duplicate application API api_keys are not allowed") return self + @model_validator(mode="after") + def validate_docs_config(self): + """Validate docs integration configuration.""" + if not self.is_lasuite_docs_integration_enabled: + return self + + if not self.lasuite_docs_base_url: + raise ValueError( + "lasuite_docs_base_url is required when docs integration is enabled" + ) + if self.lasuite_docs_server_to_server_api_key.get_secret_value() in { + "", + "NO_API_KEY", + }: + raise ValueError( + "Valid lasuite_docs_server_to_server_api_key is required when" + " docs integration is enabled" + ) + return self @cached_property def authorized_tenant_api_keys(self) -> frozenset[str]: diff --git a/src/summary/summary/core/docs_service.py b/src/summary/summary/core/docs_service.py new file mode 100644 index 00000000..96c5f130 --- /dev/null +++ b/src/summary/summary/core/docs_service.py @@ -0,0 +1,94 @@ +"""Service for delivering content to external destinations.""" + +import json +import logging +from urllib.parse import urljoin + +from requests import Session +from requests.adapters import HTTPAdapter +from urllib3.util import Retry + +from summary.core.config import get_settings + +settings = get_settings() + +logger = logging.getLogger(__name__) + + +def _create_retry_session(api_key: str | None = None): + """Create an HTTP session configured with retry logic.""" + session = Session() + retries = Retry( + total=settings.webhook_max_retries, + backoff_factor=settings.webhook_backoff_factor, + status_forcelist=settings.webhook_status_forcelist, + allowed_methods={"POST"}, + ) + session.mount("https://", HTTPAdapter(max_retries=retries)) + if api_key: + session.headers.update( + { + "Authorization": f"Bearer {api_key}", + "User-Agent": settings.app_external_user_agent, + } + ) + return session + + +def _post_with_retries(*, url, data, api_key: str | None = None): + """Send POST request with automatic retries.""" + session = _create_retry_session(api_key=api_key) + + try: + response = session.post(url, json=data, timeout=(20, 3 * 60)) + response.raise_for_status() + return response + finally: + session.close() + + +def create_document_in_lasuite_docs( + *, content: str, title: str, email: str, sub: str +) -> None: + """Call the Docs API to create a document on behalf of the user there. + + Builds the payload, sends it with retries, and logs the outcome. + """ + data = { + "title": title, + "content": content, + "email": email, + "sub": sub, + } + + logger.debug( + "Submitting document to docs endpoint: %s", settings.lasuite_docs_base_url + ) + logger.debug( + "Docs payload metadata | title_len=%s content_len=%s has_email=%s has_sub=%s", + len(title), + len(content), + bool(email), + bool(sub), + ) + + response = _post_with_retries( + url=urljoin( + settings.lasuite_docs_base_url, "/api/v1.0/documents/create-for-owner/" + ), + api_key=settings.lasuite_docs_server_to_server_api_key.get_secret_value(), + data=data, + ) + + try: + response_data = response.json() + document_id = response_data.get("id", "N/A") + except (json.JSONDecodeError, AttributeError): + document_id = "Unable to parse response" + + logger.info( + "Delivery success | Document %s submitted (HTTP %s)", + document_id, + response.status_code, + ) + logger.debug("Docs response received (body omitted)") diff --git a/src/summary/summary/core/locales/de.py b/src/summary/summary/core/locales/de.py index 10a8d2c7..985c7390 100644 --- a/src/summary/summary/core/locales/de.py +++ b/src/summary/summary/core/locales/de.py @@ -30,8 +30,5 @@ Einige Punkte, die wir Ihnen empfehlen zu überprüfen: "(externer Link)]({form_link})*\n" ), hallucination_replacement_text="[Text konnte nicht transkribiert werden]", - document_default_title="Transkription", - document_title_template=( - 'Besprechung "{room}" am {room_recording_date} um {room_recording_time}' - ), + summary_title_template="Zusammenfassung von {title}", ) diff --git a/src/summary/summary/core/locales/en.py b/src/summary/summary/core/locales/en.py index 22841d84..c0d51279 100644 --- a/src/summary/summary/core/locales/en.py +++ b/src/summary/summary/core/locales/en.py @@ -30,8 +30,5 @@ A few things we recommend you check: "(external link)]({form_link})*\n" ), hallucination_replacement_text="[Unable to transcribe text]", - document_default_title="Transcription", - document_title_template=( - 'Meeting "{room}" on {room_recording_date} at {room_recording_time}' - ), + summary_title_template="Summary of {title}", ) diff --git a/src/summary/summary/core/locales/fr.py b/src/summary/summary/core/locales/fr.py index aab29617..4eec89ab 100644 --- a/src/summary/summary/core/locales/fr.py +++ b/src/summary/summary/core/locales/fr.py @@ -29,8 +29,5 @@ Quelques points que nous vous conseillons de vérifier : "\n\n*[Donnez nous votre avis sur cette transcription]({form_link})*\n" ), hallucination_replacement_text="[Texte impossible à transcrire]", - document_default_title="Transcription", - document_title_template=( - 'Réunion "{room}" du {room_recording_date} à {room_recording_time}' - ), + summary_title_template="Résumé de {title}", ) diff --git a/src/summary/summary/core/locales/nl.py b/src/summary/summary/core/locales/nl.py index 3da0de7e..ac2cfab9 100644 --- a/src/summary/summary/core/locales/nl.py +++ b/src/summary/summary/core/locales/nl.py @@ -30,8 +30,5 @@ Een paar punten die wij u aanraden te controleren: "(externe link)]({form_link})*\n" ), hallucination_replacement_text="[Tekst kon niet worden getranscribeerd]", - document_default_title="Transcriptie", - document_title_template=( - 'Vergadering "{room}" op {room_recording_date} om {room_recording_time}' - ), + summary_title_template="Samenvatting van {title}", ) diff --git a/src/summary/summary/core/locales/strings.py b/src/summary/summary/core/locales/strings.py index 77a20979..5ae27cc6 100644 --- a/src/summary/summary/core/locales/strings.py +++ b/src/summary/summary/core/locales/strings.py @@ -12,5 +12,4 @@ class LocaleStrings: download_header_template: str form_footer_template: str hallucination_replacement_text: str - document_default_title: str - document_title_template: str + summary_title_template: str diff --git a/src/summary/summary/core/models.py b/src/summary/summary/core/models.py index 11983d15..e91f100b 100644 --- a/src/summary/summary/core/models.py +++ b/src/summary/summary/core/models.py @@ -1,6 +1,8 @@ """Models for the API & Celery tasks creation.""" -from pydantic import BaseModel, Field, field_validator +from datetime import datetime + +from pydantic import AwareDatetime, BaseModel, EmailStr, Field, field_validator from summary.core.config import get_settings from summary.core.types import Url @@ -12,10 +14,60 @@ class SharedV2TaskCreation(BaseModel): """Model that holds basic information for task creation.""" user_sub: str = Field(title="User Sub", description="The user's sub.") + user_email: EmailStr | None = Field( + default=None, + title="User Email", + description="The user's email for analytics purposes.", + ) -class TranscribeTaskV2Request(SharedV2TaskCreation): - """Model for creating a transcribe and summarize task (used for API request).""" +class RecordingMetadata(BaseModel): + """Model for recording metadata.""" + + cloud_storage_url: Url = Field( + title="Cloud Storage URL", + description="The URL of the metadata file for speaker assignement.", + ) + started_at: AwareDatetime = Field(title="Start time of the recording to transcribe") + ended_at: AwareDatetime = Field(title="End time of the recording to transcribe") + + +class PushToDocsBaseConfig(BaseModel): + """Model containing information for pushing transcript and summaries to docs.""" + + user_email: EmailStr = Field( + title="User Email", description="The user's email, future owner of the docs." + ) + title: str = Field(title="Title", description="The title for the created document.") + + +class PushToDocsTranscriptConfig(PushToDocsBaseConfig): + """Model for push to docs information for transcripts.""" + + download_link: Url | None = Field( + default=None, + title="Download Link", + description="The link to download the recording.", + ) + form_link: Url | None = Field( + default=None, + title="Form Link", + description="The link to fill out a form for the recording.", + ) + auto_create_summary: bool = Field( + title="Auto Create Summary Docs", + description="Whether to automatically create a summary " + "for the transcription task and push it to docs.", + default=False, + ) + + +class PushToDocsSummaryConfig(PushToDocsBaseConfig): + """Model for push to docs information for summaries.""" + + +class TranscribeTaskApiRequest(SharedV2TaskCreation): + """Model for creating a transcribe task (used for API request).""" cloud_storage_url: Url = Field( title="Cloud storage URL", @@ -27,7 +79,17 @@ class TranscribeTaskV2Request(SharedV2TaskCreation): description="The language of the context text.", ) language: str = Field( - title="Language", description="The language of the content to summarize." + title="Language", description="The language of the content to transcribe." + ) + metadata: RecordingMetadata | None = Field( + title="Metadata", + description="The metadata for the transcribe task.", + default=None, + ) + push_to_docs_config: PushToDocsTranscriptConfig | None = Field( + title="Push to Docs info", + description="If set, configuration for pushing to docs", + default=None, ) @field_validator("language") @@ -42,19 +104,30 @@ class TranscribeTaskV2Request(SharedV2TaskCreation): return v -class TranscribeTaskV2Payload(TranscribeTaskV2Request): - """Model for creating a transcribe and summarize task (used for actual task creation).""" # noqa: E501 +class TranscribeTaskJob(TranscribeTaskApiRequest): + """Model for creating a transcribe task (used for actual task creation).""" tenant_id: str = Field(title="Tenant ID", description="The ID of the tenant.") + received_at: datetime = Field( + title="Received At", description="The time the task was received." + ) -class SummarizeTaskV2Request(SharedV2TaskCreation): +class SummarizeTaskApiRequest(SharedV2TaskCreation): """Model for creating a summarize task (used for API request).""" content: str = Field(title="Content", description="The content to summarize.") -class SummarizeTaskV2Payload(SummarizeTaskV2Request): +class SummarizeTaskJob(SummarizeTaskApiRequest): """Model for creating a summarize task (used for actual task creation).""" tenant_id: str = Field(title="Tenant ID", description="The ID of the tenant.") + received_at: datetime = Field( + title="Received At", description="The time the task was received." + ) + push_to_docs_config: PushToDocsBaseConfig | None = Field( + title="Push to Docs info", + description="If set, configuration for pushing to docs", + default=None, + ) diff --git a/src/summary/summary/core/shared_models.py b/src/summary/summary/core/shared_models.py index b18c81cc..2e80896c 100644 --- a/src/summary/summary/core/shared_models.py +++ b/src/summary/summary/core/shared_models.py @@ -167,4 +167,5 @@ __all__ = [ "SummarizeWebhookPayloads", "WebhookPayloads", "WhisperXResponse", + "webhook_payload_adapter", ] diff --git a/src/summary/summary/core/transcript_formatter.py b/src/summary/summary/core/transcript_formatter.py index be1ff53e..b9d3e541 100644 --- a/src/summary/summary/core/transcript_formatter.py +++ b/src/summary/summary/core/transcript_formatter.py @@ -1,9 +1,6 @@ """Transcript formatting into readable conversation format with speaker labels.""" import logging -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 @@ -38,16 +35,13 @@ class TranscriptFormatter: return None - def format( # noqa: PLR0913 + def format( self, transcription, - room: str | None = None, - recording_datetime: str | None = None, - owner_timezone: str | None = None, download_link: str | None = None, form_link: str | None = None, - ) -> Tuple[str, str]: - """Format transcription into the final document and its title.""" + ) -> str: + """Format transcription into the final document.""" segments = self._get_segments(transcription) if not segments: @@ -59,9 +53,7 @@ class TranscriptFormatter: if form_link: content = self._add_footer(content, form_link) - title = self._generate_title(room, recording_datetime, owner_timezone) - - return content, title + return content def _remove_hallucinations(self, content: str) -> str: """Remove hallucination patterns from content.""" @@ -104,23 +96,3 @@ class TranscriptFormatter: footer = self._locale.form_footer_template.format(form_link=form_link) return content + footer - - def _generate_title( - self, - 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_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=dt.strftime("%Y-%m-%d"), - room_recording_time=dt.strftime("%H:%M"), - ) diff --git a/src/summary/summary/core/webhook_service.py b/src/summary/summary/core/webhook_service.py index 0e196cb7..2c132e01 100644 --- a/src/summary/summary/core/webhook_service.py +++ b/src/summary/summary/core/webhook_service.py @@ -4,9 +4,6 @@ import json import logging import requests -from requests import Session -from requests.adapters import HTTPAdapter -from urllib3.util import Retry from summary.core.config import get_settings from summary.core.shared_models import ( @@ -18,83 +15,6 @@ settings = get_settings() logger = logging.getLogger(__name__) -def _create_retry_session(api_key: str | None = None): - """Create an HTTP session configured with retry logic.""" - session = Session() - retries = Retry( - total=settings.webhook_max_retries, - backoff_factor=settings.webhook_backoff_factor, - status_forcelist=settings.webhook_status_forcelist, - allowed_methods={"POST"}, - ) - session.mount("https://", HTTPAdapter(max_retries=retries)) - if api_key: - session.headers.update({"Authorization": f"Bearer {api_key}"}) - - return session - - -def _post_with_retries(*, url, data, api_key: str | None = None): - """Send POST request with automatic retries.""" - session = _create_retry_session(api_key=api_key) - - try: - response = session.post(url, json=data) - response.raise_for_status() - return response - finally: - session.close() - - -def call_webhook_v1(*, tenant_id: str, payload: dict) -> None: - """Call webhook with payload a payload and optional token.""" - tenant = settings.get_authorized_tenant(tenant_id=tenant_id) - - logger.debug("Submitting to %s", tenant.webhook_url) - logger.debug("Request payload: %s", json.dumps(payload, indent=2)) - - response = _post_with_retries( - url=tenant.webhook_url, - api_key=tenant.webhook_api_key.get_secret_value(), - data=payload, - ) - - try: - response_data = response.json() - document_id = response_data.get("id", "N/A") - except (json.JSONDecodeError, AttributeError): - document_id = "Unable to parse response" - response_data = response.text - - logger.info( - "Delivery success | Document %s submitted (HTTP %s)", - document_id, - response.status_code, - ) - logger.debug("Full response: %s", response_data) - - -def submit_content(content: str, title: str, email: str, sub: str) -> None: - """Submit content to the configured webhook destination. - - Builds the payload, sends it with retries, and logs the outcome. - - Notes: - Deprecated: Use call_webhook_v2 directly instead. - - Deprecated: - This will route content to the v1 default tenant - """ - data = { - "title": title, - "content": content, - "email": email, - "sub": sub, - } - - call_webhook_v1(payload=data, tenant_id=settings.v1_tenant_id) - - def call_webhook_v2( *, tenant_id: str, @@ -114,6 +34,7 @@ def call_webhook_v2( json=payload.model_dump(), headers={ "Authorization": f"Bearer {tenant.webhook_api_key.get_secret_value()}", + "User-Agent": settings.app_external_user_agent, }, timeout=(10, 20), ) diff --git a/src/summary/tests/api/test_api_tasks_v2.py b/src/summary/tests/api/test_api_tasks_v2.py index 0fac599e..75837bf8 100644 --- a/src/summary/tests/api/test_api_tasks_v2.py +++ b/src/summary/tests/api/test_api_tasks_v2.py @@ -1,5 +1,6 @@ """Integration tests for the V2 task API endpoints.""" +from datetime import datetime from unittest.mock import ANY, MagicMock, patch @@ -31,6 +32,8 @@ class TestTasksV2: } args = mock_apply_async.call_args.kwargs["args"] + received_at = args[0].pop("received_at") + assert isinstance(received_at, datetime) assert args == [ { "user_sub": "remote-001", @@ -38,6 +41,9 @@ class TestTasksV2: "language": "en", "context_language": "fr", "tenant_id": "test-tenant", + "user_email": None, + "metadata": None, + "push_to_docs_config": None, } ] @@ -64,11 +70,14 @@ class TestTasksV2: } args = mock_apply_async.call_args.kwargs["args"] + received_at = args[0].pop("received_at") + assert isinstance(received_at, datetime) assert args == [ { "user_sub": "remote-002", "content": "This is a long meeting transcript to summarize.", "tenant_id": "test-tenant", + "user_email": None, } ]