gsadmin/meet
1
0
Fork 0
mirror of https://github.com/suitenumerique/meet.git synced 2026-07-26 11:58:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨(summary) extend v2 routes to support visio usecase

Browse Source 
* Update the v2 create transcribe payload to support
publishing directly to docs and performing automatically
a summary if requested
* Note that title computation is to be handled by the caller
now as it makes more sense and avoids throwing a bunch
of parameters to the endpoint.
* All files are send as signed URL now, we don't read
directly from s3 anymore,
* Docs integration is now explicit in summary settings
* To make analytics work properly accross projects,
we use the user sub as distinct id,
this will require a new posthog project to be deployed
to work properly.
We create a post hog event at the creation request processing,
to make sure feature flags work properly after that.
* User email should now be provided to the API, it's not
mandatory to avoid a breaking change.
* Use a specific user agent for better tracking
Finally note that existing helpers don't always make use of
 the pydantic models, so that's why there
are model_dump in some places. To avoid bigger changes.
This commit is contained in:
Florent Chehab
2026-05-28 21:40:13 +02:00
committed by aleb_the_flash
parent 27da57aff5
commit ecf8f0fe3f
15 changed files with 470 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/summary/summary/api/route/tasks_v2.py
+65 -8
View File
@@ -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()
src/summary/summary/core/analytics.py
+12 -9
View File
@@ -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")
src/summary/summary/core/celery_worker.py
+162 -58
View File
@@ -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),
src/summary/summary/core/config.py
+36 -4
View File
@@ -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]:
src/summary/summary/core/docs_service.py
+94
View File
@@ -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)")
src/summary/summary/core/locales/de.py
+1 -4
View File
@@ -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}",
)
src/summary/summary/core/locales/en.py
+1 -4
View File
@@ -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}",
)
src/summary/summary/core/locales/fr.py
+1 -4
View File
@@ -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}",
)
src/summary/summary/core/locales/nl.py
+1 -4
View File
@@ -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}",
)
src/summary/summary/core/locales/strings.py
+1 -2
View File
@@ -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
src/summary/summary/core/models.py
+81 -8
View File
@@ -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,
)
src/summary/summary/core/shared_models.py
+1
View File
@@ -167,4 +167,5 @@ __all__ = [
"SummarizeWebhookPayloads",
"WebhookPayloads",
"WhisperXResponse",
"webhook_payload_adapter",
]
src/summary/summary/core/transcript_formatter.py
+4 -32
View File
@@ -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"),
)
src/summary/summary/core/webhook_service.py
+1 -80
View File
@@ -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),
)
src/summary/tests/api/test_api_tasks_v2.py
+9
View File
@@ -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,
}
]