faglo/main_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: claude final touches

Browse Source
This commit is contained in:
Daniil
2026-03-17 18:11:23 +03:00
parent 4b90925c2a
commit 0299949553
21 changed files with 1915 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpv3/modules/tasks/schemas.py
+10 -12
View File
@@ -37,9 +37,7 @@ class SilenceRemoveRequest(Schema):
min_silence_duration_ms: int = Field(
default=200, description="Minimum silence duration in milliseconds"
)
silence_threshold_db: int = Field(
default=16, description="Silence threshold in decibels"
)
silence_threshold_db: int = Field(default=16, description="Silence threshold in decibels")
padding_ms: int = Field(
default=100, description="Padding around non-silent segments in milliseconds"
)
@@ -53,9 +51,7 @@ class SilenceDetectRequest(Schema):
min_silence_duration_ms: int = Field(
default=200, description="Minimum silence duration in milliseconds"
)
silence_threshold_db: int = Field(
default=16, description="Silence threshold in decibels"
)
silence_threshold_db: int = Field(default=16, description="Silence threshold in decibels")
padding_ms: int = Field(
default=100, description="Padding around non-silent segments in milliseconds"
)
@@ -67,9 +63,7 @@ class SilenceApplyRequest(Schema):
file_key: str = Field(..., description="Storage key of the input file")
out_folder: str = Field(..., description="Output folder for processed file")
project_id: UUID | None = Field(default=None, description="Associated project ID")
output_name: str | None = Field(
default=None, description="Display name for the output file"
)
output_name: str | None = Field(default=None, description="Display name for the output file")
cuts: list[dict] = Field(
..., description="Cut regions: [{'start_ms': int, 'end_ms': int}, ...]"
)
@@ -103,6 +97,12 @@ class CaptionsGenerateRequest(Schema):
folder: str = Field(..., description="Output folder for rendered video")
transcription_id: UUID = Field(..., description="ID of the transcription to use")
project_id: UUID | None = Field(default=None, description="Associated project ID")
preset_id: UUID | None = Field(
default=None, description="Caption style preset ID (mutually exclusive with style_config)"
)
style_config: dict | None = Field(
default=None, description="Inline caption style config (overrides preset_id)"
)
class FrameExtractRequest(Schema):
@@ -110,9 +110,7 @@ class FrameExtractRequest(Schema):
file_key: str = Field(..., description="S3 key of the video file")
project_id: UUID | None = Field(default=None, description="Associated project ID")
regenerate: bool = Field(
default=False, description="Delete existing frames and re-extract"
)
regenerate: bool = Field(default=False, description="Delete existing frames and re-extract")
# --- Response schemas ---
cpv3/modules/tasks/service.py
+483 -74
View File
@@ -10,17 +10,19 @@ import json
import logging
import time
import uuid
from pathlib import Path
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import dramatiq # type: ignore[import-untyped]
from dramatiq.brokers.redis import RedisBroker # type: ignore[import-untyped]
import httpx
from dramatiq.brokers.redis import RedisBroker # type: ignore[import-untyped]
from sqlalchemy.ext.asyncio import AsyncSession
from cpv3.infrastructure.deps import _get_storage_service
from cpv3.infrastructure.settings import get_settings
from cpv3.infrastructure.storage.utils import get_user_folder
from cpv3.modules.files.repository import FileRepository
from cpv3.modules.files.schemas import FileCreate
from cpv3.modules.jobs.models import Job
@@ -46,7 +48,6 @@ from cpv3.modules.tasks.schemas import (
TaskWebhookEvent,
TranscriptionGenerateRequest,
)
from cpv3.infrastructure.storage.utils import get_user_folder
from cpv3.modules.notifications.service import NotificationService
from cpv3.modules.transcription.repository import TranscriptionRepository
from cpv3.modules.transcription.schemas import TranscriptionCreate
@@ -61,6 +62,7 @@ JOB_STATUS_PENDING: JobStatusEnum = "PENDING"
JOB_STATUS_RUNNING: JobStatusEnum = "RUNNING"
JOB_STATUS_DONE: JobStatusEnum = "DONE"
JOB_STATUS_FAILED: JobStatusEnum = "FAILED"
JOB_STATUS_CANCELLED: JobStatusEnum = "CANCELLED"
JOB_TYPE_MEDIA_PROBE: JobTypeEnum = "MEDIA_PROBE"
JOB_TYPE_SILENCE_REMOVE: JobTypeEnum = "SILENCE_REMOVE"
@@ -94,6 +96,7 @@ MESSAGE_PROBING_MEDIA = "Probing media"
MESSAGE_PROCESSING = "Processing"
MESSAGE_CONVERTING = "Converting"
MESSAGE_RENDERING_CAPTIONS = "Rendering captions"
MESSAGE_CANCELLED = "Отменено пользователем"
MESSAGE_EXTRACTING_FRAMES = "Извлечение кадров"
MESSAGE_UPLOADING_FRAMES = "Загрузка кадров"
MESSAGE_DELETING_OLD_FRAMES = "Удаление старых кадров"
@@ -116,6 +119,13 @@ MESSAGE_APPLYING_CUTS = "Применение вырезок"
PROGRESS_THROTTLE_SECONDS = 3.0
ACTIVE_JOB_STATUSES = (JOB_STATUS_PENDING, JOB_STATUS_RUNNING)
DRAMATIQ_BROKER_REF_SEPARATOR = ":"
class JobCancelledError(RuntimeError):
"""Raised when a job was cancelled before completion."""
# ---------------------------------------------------------------------------
# Dramatiq broker setup
# ---------------------------------------------------------------------------
@@ -163,9 +173,7 @@ def _send_webhook_event(webhook_url: str, event: TaskWebhookEvent) -> None:
"""Send a task webhook event to the API."""
payload = event.model_dump(mode="json", exclude_none=True)
try:
response = httpx.post(
webhook_url, json=payload, timeout=WEBHOOK_TIMEOUT_SECONDS
)
response = httpx.post(webhook_url, json=payload, timeout=WEBHOOK_TIMEOUT_SECONDS)
response.raise_for_status()
except Exception:
logger.exception("Failed to send task webhook event to %s", webhook_url)
@@ -197,17 +205,69 @@ def _run_async(coro: Any) -> Any:
loop.close()
def _serialize_broker_reference(queue_name: str, redis_message_id: str) -> str:
"""Serialize queue name and Dramatiq redis message id into one field."""
return f"{queue_name}{DRAMATIQ_BROKER_REF_SEPARATOR}{redis_message_id}"
def _parse_broker_reference(broker_id: str | None) -> tuple[str, str] | None:
"""Parse queue name and Dramatiq redis message id from stored broker_id."""
if not broker_id or DRAMATIQ_BROKER_REF_SEPARATOR not in broker_id:
return None
queue_name, redis_message_id = broker_id.split(DRAMATIQ_BROKER_REF_SEPARATOR, 1)
if not queue_name or not redis_message_id:
return None
return queue_name, redis_message_id
def _get_job_status_sync(job_id: uuid.UUID) -> JobStatusEnum | None:
"""Read current job status using a sync connection (safe for Dramatiq workers)."""
import psycopg2
settings = get_settings()
dsn = (
f"host={settings.postgres_host} port={settings.postgres_port} "
f"dbname={settings.postgres_database} "
f"user={settings.postgres_user} password={settings.postgres_password}"
)
try:
conn = psycopg2.connect(dsn)
try:
with conn.cursor() as cur:
cur.execute("SELECT status FROM jobs WHERE id = %s", (str(job_id),))
row = cur.fetchone()
return row[0] if row else None
finally:
conn.close()
except Exception:
logger.warning("Failed to check job status for %s", job_id, exc_info=True)
return None
def _raise_if_job_cancelled(job_id: uuid.UUID) -> None:
"""Stop worker execution when the job is already cancelled in the database."""
status = _get_job_status_sync(job_id)
if status == JOB_STATUS_CANCELLED:
raise JobCancelledError(MESSAGE_CANCELLED)
# ---------------------------------------------------------------------------
# Dramatiq actors
# ---------------------------------------------------------------------------
@dramatiq.actor(max_retries=3, min_backoff=1000)
@dramatiq.actor(max_retries=0)
def media_probe_actor(job_id: str, webhook_url: str, file_key: str) -> None:
"""Probe media file to extract metadata."""
from cpv3.modules.media.service import probe_media
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("media_probe_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -237,6 +297,9 @@ def media_probe_actor(job_id: str, webhook_url: str, file_key: str) -> None:
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("media_probe_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("media_probe_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -247,10 +310,9 @@ def media_probe_actor(job_id: str, webhook_url: str, file_key: str) -> None:
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=3, min_backoff=1000)
@dramatiq.actor(max_retries=0)
def silence_remove_actor(
job_id: str,
webhook_url: str,
@@ -264,6 +326,11 @@ def silence_remove_actor(
from cpv3.modules.media.service import remove_silence
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("silence_remove_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -306,6 +373,9 @@ def silence_remove_actor(
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("silence_remove_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("silence_remove_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -316,10 +386,9 @@ def silence_remove_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=3, min_backoff=1000)
@dramatiq.actor(max_retries=0)
def silence_detect_actor(
job_id: str,
webhook_url: str,
@@ -332,6 +401,11 @@ def silence_detect_actor(
from cpv3.modules.media.service import detect_silence
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("silence_detect_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -369,6 +443,9 @@ def silence_detect_actor(
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("silence_detect_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("silence_detect_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -379,10 +456,9 @@ def silence_detect_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=3, min_backoff=1000)
@dramatiq.actor(max_retries=0)
def silence_apply_actor(
job_id: str,
webhook_url: str,
@@ -395,6 +471,11 @@ def silence_apply_actor(
from cpv3.modules.media.service import apply_silence_cuts
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("silence_apply_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -436,6 +517,9 @@ def silence_apply_actor(
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("silence_apply_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("silence_apply_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -446,10 +530,9 @@ def silence_apply_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=3, min_backoff=1000)
@dramatiq.actor(max_retries=0)
def media_convert_actor(
job_id: str,
webhook_url: str,
@@ -461,6 +544,11 @@ def media_convert_actor(
from cpv3.modules.media.service import convert_to_mp4
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("media_convert_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -482,9 +570,7 @@ def media_convert_actor(
progress_pct=PROGRESS_MEDIA_CONVERT,
),
)
result = _run_async(
convert_to_mp4(storage, file_key=file_key, out_folder=out_folder)
)
result = _run_async(convert_to_mp4(storage, file_key=file_key, out_folder=out_folder))
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -499,6 +585,9 @@ def media_convert_actor(
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("media_convert_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("media_convert_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -509,10 +598,9 @@ def media_convert_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=2, min_backoff=2000)
@dramatiq.actor(max_retries=0)
def transcription_generate_actor(
job_id: str,
webhook_url: str,
@@ -528,6 +616,11 @@ def transcription_generate_actor(
)
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("transcription_generate_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -548,11 +641,15 @@ def transcription_generate_actor(
raise ValueError(ERROR_NO_AUDIO_STREAM)
# Extract probe metadata for artifact creation
duration_seconds = float(probe.format.duration) if probe.format and probe.format.duration else 0.0
duration_seconds = (
float(probe.format.duration) if probe.format and probe.format.duration else 0.0
)
video_stream = next((s for s in probe.streams if s.codec_type == "video"), None)
probe_meta = {
"duration_seconds": duration_seconds,
"frame_rate": _parse_frame_rate(video_stream.r_frame_rate) if video_stream and video_stream.r_frame_rate else None,
"frame_rate": _parse_frame_rate(video_stream.r_frame_rate)
if video_stream and video_stream.r_frame_rate
else None,
"width": video_stream.width if video_stream else None,
"height": video_stream.height if video_stream else None,
}
@@ -573,9 +670,9 @@ def transcription_generate_actor(
if now - last_report_time < PROGRESS_THROTTLE_SECONDS:
return
last_report_time = now
mapped = PROGRESS_TRANSCRIPTION_START + (
pct / 100.0
) * (PROGRESS_TRANSCRIPTION_END - PROGRESS_TRANSCRIPTION_START)
mapped = PROGRESS_TRANSCRIPTION_START + (pct / 100.0) * (
PROGRESS_TRANSCRIPTION_END - PROGRESS_TRANSCRIPTION_START
)
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -617,18 +714,9 @@ def transcription_generate_actor(
finished_at=_utc_now(),
),
)
except (ValueError, RuntimeError) as exc:
logger.exception(
"transcription_generate_actor failed (non-transient): %s", job_uuid
)
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
status=JOB_STATUS_FAILED,
error_message=str(exc),
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("transcription_generate_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("transcription_generate_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -639,22 +727,31 @@ def transcription_generate_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=2, min_backoff=2000)
RENDER_POLL_INTERVAL_SECONDS = 5
RENDER_POLL_TIMEOUT_SECONDS = 600
@dramatiq.actor(max_retries=0)
def captions_generate_actor(
job_id: str,
webhook_url: str,
video_s3_path: str,
folder: str,
transcription_json: dict,
style_config: dict | None = None,
) -> None:
"""Generate captions on video."""
"""Generate captions on video (async via Remotion + BullMQ)."""
from cpv3.modules.captions.service import generate_captions
from cpv3.modules.transcription.schemas import Document
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("captions_generate_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -673,21 +770,96 @@ def captions_generate_actor(
),
)
document = Document.model_validate(transcription_json)
output_path = _run_async(
# Call Remotion with callback_url so it sends progress webhooks directly
render_id = _run_async(
generate_captions(
video_s3_path=video_s3_path, folder=folder, transcription=document
video_s3_path=video_s3_path,
folder=folder,
transcription=document,
style_config=style_config,
callback_url=webhook_url,
render_id=job_id,
)
)
_raise_if_job_cancelled(job_uuid)
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
status=JOB_STATUS_DONE,
current_message=MESSAGE_COMPLETED,
progress_pct=PROGRESS_COMPLETE,
output_data={"output_path": output_path},
finished_at=_utc_now(),
current_message=MESSAGE_RENDERING_CAPTIONS,
output_data={"render_id": render_id},
),
)
# Polling fallback: wait for Remotion to finish rendering
# Primary progress is delivered via Remotion → webhook directly
settings = get_settings()
elapsed = 0.0
last_polled_status: str | None = None
while elapsed < RENDER_POLL_TIMEOUT_SECONDS:
_raise_if_job_cancelled(job_uuid)
time.sleep(RENDER_POLL_INTERVAL_SECONDS)
elapsed += RENDER_POLL_INTERVAL_SECONDS
try:
resp = httpx.get(
f"{settings.remotion_service_url}/api/render/{render_id}",
timeout=10,
)
resp.raise_for_status()
data = resp.json()
except Exception:
logger.warning("Render poll failed for %s, retrying...", render_id)
continue
status = data.get("status")
if status != last_polled_status:
logger.info(
"Remotion render %s status=%s progress=%s callback_delivered=%s",
render_id,
status,
data.get("progress_pct"),
data.get("callback_delivered"),
)
last_polled_status = status
if status == "done":
if data.get("callback_delivered") is True:
# Remotion already sent DONE webhook — exit cleanly
return
output_path = data.get("output_path")
if not output_path:
raise RuntimeError(
"Remotion render finished without output_path in polling response"
)
logger.warning(
"Remotion render %s finished without confirmed DONE webhook, sending fallback completion",
render_id,
)
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
status=JOB_STATUS_DONE,
progress_pct=PROGRESS_COMPLETE,
current_message="Готово",
output_data={"output_path": str(output_path)},
finished_at=_utc_now(),
),
)
return
if status == "failed":
error = data.get("error", "Render failed")
raise RuntimeError(f"Remotion render failed: {error}")
raise TimeoutError(
f"Render {render_id} did not complete within {RENDER_POLL_TIMEOUT_SECONDS}s"
)
except JobCancelledError:
logger.info("captions_generate_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("captions_generate_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -698,10 +870,9 @@ def captions_generate_actor(
finished_at=_utc_now(),
),
)
raise
@dramatiq.actor(max_retries=2, min_backoff=2000)
@dramatiq.actor(max_retries=0)
def frame_extract_actor(
job_id: str,
webhook_url: str,
@@ -717,6 +888,11 @@ def frame_extract_actor(
)
job_uuid = uuid.UUID(job_id)
try:
_raise_if_job_cancelled(job_uuid)
except JobCancelledError:
logger.info("frame_extract_actor cancelled: %s", job_uuid)
return
_send_webhook_event(
webhook_url,
TaskWebhookEvent(
@@ -738,9 +914,7 @@ def frame_extract_actor(
progress_pct=PROGRESS_FRAME_EXTRACT_START,
),
)
old_meta = _run_async(
read_frames_metadata(storage, frames_folder=frames_folder)
)
old_meta = _run_async(read_frames_metadata(storage, frames_folder=frames_folder))
if old_meta is not None:
_run_async(
delete_frames(
@@ -797,6 +971,9 @@ def frame_extract_actor(
finished_at=_utc_now(),
),
)
except JobCancelledError:
logger.info("frame_extract_actor cancelled: %s", job_uuid)
return
except Exception as exc:
logger.exception("frame_extract_actor failed: %s", job_uuid)
_send_webhook_event(
@@ -807,7 +984,6 @@ def frame_extract_actor(
finished_at=_utc_now(),
),
)
raise
# ---------------------------------------------------------------------------
@@ -824,6 +1000,97 @@ class TaskService:
self._event_repo = JobEventRepository(session)
self._webhook_repo = WebhookRepository(session)
async def _update_job_broker_reference(self, job: Job, broker_reference: str) -> Job:
"""Persist the transport-specific broker reference after enqueueing."""
job.broker_id = broker_reference
await self._session.commit()
await self._session.refresh(job)
return job
async def _find_duplicate_active_job(
self,
*,
requester: User,
job_type: JobTypeEnum,
project_id: uuid.UUID | None,
input_data: dict,
) -> Job | None:
"""Reuse an already running job for the same request payload."""
jobs = await self._job_repo.list_active_by_type(
requester=requester,
job_type=job_type,
project_id=project_id,
statuses=ACTIVE_JOB_STATUSES,
)
for job in jobs:
if job.input_data == input_data:
return job
return None
async def _create_cancellation_notification(self, job: Job) -> None:
"""Emit a single cancellation notification to the current user."""
if job.user_id is None:
return
notification_service = NotificationService(self._session)
await notification_service.create_task_notification(
user_id=job.user_id,
job=job,
event=TaskWebhookEvent(
status=JOB_STATUS_CANCELLED,
current_message=MESSAGE_CANCELLED,
finished_at=job.finished_at,
),
)
def _cancel_dramatiq_message_sync(self, broker_id: str | None) -> None:
"""Remove a queued Dramatiq message from Redis when possible."""
broker_reference = _parse_broker_reference(broker_id)
if broker_reference is None:
return
queue_name, redis_message_id = broker_reference
namespace = _redis_broker.namespace
queue_key = f"{namespace}:{queue_name}"
queue_messages_key = f"{queue_key}.msgs"
delayed_queue_key = f"{queue_key}.DQ"
delayed_messages_key = f"{delayed_queue_key}.msgs"
acks_pattern = f"{namespace}:__acks__.*.{queue_name}"
pipeline = _redis_broker.client.pipeline()
pipeline.lrem(queue_key, 1, redis_message_id)
pipeline.hdel(queue_messages_key, redis_message_id)
pipeline.lrem(delayed_queue_key, 1, redis_message_id)
pipeline.hdel(delayed_messages_key, redis_message_id)
for key in _redis_broker.client.scan_iter(match=acks_pattern):
pipeline.srem(key, redis_message_id)
pipeline.execute()
async def _cancel_dramatiq_message(self, broker_id: str | None) -> None:
"""Run Redis queue cleanup for a Dramatiq message off the event loop."""
await asyncio.to_thread(self._cancel_dramatiq_message_sync, broker_id)
async def _cancel_caption_render(self, job: Job) -> None:
"""Cancel the downstream Remotion render when it has already been queued."""
if job.job_type != JOB_TYPE_CAPTIONS_GENERATE:
return
output_data = job.output_data or {}
render_id = output_data.get("render_id") or str(job.id)
settings = get_settings()
async with httpx.AsyncClient(timeout=WEBHOOK_TIMEOUT_SECONDS) as client:
response = await client.delete(
f"{settings.remotion_service_url}/api/render/{render_id}"
)
if response.status_code == 404:
return
response.raise_for_status()
async def _create_job_and_webhook(
self,
*,
@@ -873,21 +1140,31 @@ class TaskService:
project_id=project_id,
input_data=input_data,
)
actor.send(job_id=str(job.id), webhook_url=webhook_url, **actor_kwargs)
message = actor.send(job_id=str(job.id), webhook_url=webhook_url, **actor_kwargs)
redis_message_id = message.options.get("redis_message_id")
if redis_message_id:
broker_reference = _serialize_broker_reference(
message.queue_name,
str(redis_message_id),
)
await self._update_job_broker_reference(job, broker_reference)
return TaskSubmitResponse(
job_id=job.id,
webhook_url=webhook_url,
status=JOB_STATUS_PENDING,
)
async def record_webhook_event(
self, *, job_id: uuid.UUID, event: TaskWebhookEvent
) -> Job:
async def record_webhook_event(self, *, job_id: uuid.UUID, event: TaskWebhookEvent) -> Job:
"""Apply a webhook event to the job and store a job event record."""
job = await self._job_repo.get_by_id(job_id)
if job is None:
raise ValueError(f"Job {job_id} not found")
if job.status in (JOB_STATUS_DONE, JOB_STATUS_FAILED, JOB_STATUS_CANCELLED):
logger.info("Ignoring webhook for terminal job %s (status=%s)", job_id, job.status)
return job
job_update = JobUpdate(
status=event.status,
project_pct=event.progress_pct,
@@ -906,24 +1183,23 @@ class TaskService:
)
# Save artifacts BEFORE sending notifications so data exists when frontend refetches
if (
job.job_type == JOB_TYPE_TRANSCRIPTION_GENERATE
and event.status == JOB_STATUS_DONE
):
if job.job_type == JOB_TYPE_TRANSCRIPTION_GENERATE and event.status == JOB_STATUS_DONE:
try:
await self._save_transcription_artifacts(job)
except Exception:
logger.exception(
"Failed to save transcription artifacts for job %s", job_id
)
logger.exception("Failed to save transcription artifacts for job %s", job_id)
if job.job_type == JOB_TYPE_MEDIA_CONVERT and event.status == JOB_STATUS_DONE:
try:
await self._save_convert_artifacts(job)
except Exception:
logger.exception(
"Failed to save convert artifacts for job %s", job_id
)
logger.exception("Failed to save convert artifacts for job %s", job_id)
if job.job_type == JOB_TYPE_CAPTIONS_GENERATE and event.status == JOB_STATUS_DONE:
try:
await self._save_captions_artifacts(job)
except Exception:
logger.exception("Failed to save captions artifacts for job %s", job_id)
# Push real-time notification via WebSocket (after artifacts are persisted)
if job.user_id is not None:
@@ -937,6 +1213,50 @@ class TaskService:
return job
async def cancel_job(self, job: Job) -> Job:
"""Cancel a job, clean queued transport state and ignore late webhooks."""
if job.status in (JOB_STATUS_DONE, JOB_STATUS_FAILED, JOB_STATUS_CANCELLED):
return job
try:
await self._cancel_dramatiq_message(job.broker_id)
except Exception:
logger.exception("Failed to cancel Dramatiq message for job %s", job.id)
try:
await self._cancel_caption_render(job)
except Exception:
logger.exception("Failed to cancel caption render for job %s", job.id)
finished_at = _utc_now()
job = await self._job_repo.update(
job,
JobUpdate(
status=JOB_STATUS_CANCELLED,
current_message=MESSAGE_CANCELLED,
finished_at=finished_at,
),
)
await self._event_repo.create(
JobEventCreate(
job_id=job.id,
event_type=f"{EVENT_TYPE_STATUS_PREFIX}{JOB_STATUS_CANCELLED}",
payload={
"status": JOB_STATUS_CANCELLED,
"current_message": MESSAGE_CANCELLED,
"finished_at": finished_at.isoformat(),
},
)
)
try:
await self._create_cancellation_notification(job)
except Exception:
logger.exception("Failed to create cancellation notification for job %s", job.id)
return job
async def _save_transcription_artifacts(self, job: Job) -> None:
"""Create Transcription, ArtifactMediaFile and File records."""
input_data = job.input_data or {}
@@ -980,9 +1300,9 @@ class TaskService:
user_folder = get_user_folder(user)
json_bytes = json.dumps(document, ensure_ascii=False).encode("utf-8")
# Build display name: "Транскрипция <video_name>.json"
# Build display name: "Транскрибация <video_name>.json"
video_stem = Path(source_file.original_filename).stem
transcription_filename = f"Транскрипция {video_stem}.json"
transcription_filename = f"Транскрибация {video_stem}.json"
artifact_key = await storage.upload_fileobj(
fileobj=io.BytesIO(json_bytes),
@@ -1061,7 +1381,7 @@ class TaskService:
stem = Path(source_file.original_filename).stem
else:
stem = Path(file_key).stem
converted_filename = f"{stem}.mp4"
converted_filename = f"Конвертированое видео {stem}.mp4"
# Create File record for the converted MP4 (no project_id — only reachable via artifact)
converted_file = await file_repo.create(
@@ -1091,6 +1411,73 @@ class TaskService:
logger.info("Saved convert artifacts for job %s", job.id)
async def _save_captions_artifacts(self, job: Job) -> None:
"""Create File and ArtifactMediaFile records for captioned video."""
input_data = job.input_data or {}
output_data = job.output_data or {}
video_s3_path: str = input_data["video_s3_path"]
project_id: uuid.UUID | None = (
uuid.UUID(input_data["project_id"]) if input_data.get("project_id") else None
)
output_path: str = output_data["output_path"]
# Resolve user
user_repo = UserRepository(self._session)
user = await user_repo.get_by_id(job.user_id) # type: ignore[arg-type]
if user is None:
logger.warning("User %s not found, skipping captions artifact save", job.user_id)
return
# Get file size from S3
storage = _get_storage_service()
file_size = await storage.size(output_path)
# Derive output filename from source video
file_repo = FileRepository(self._session)
source_file = await file_repo.get_by_path(video_s3_path)
if source_file is not None:
stem = Path(source_file.original_filename).stem
else:
stem = Path(video_s3_path).stem
captioned_filename = f"Видео с субтитрами {stem}.mp4"
# Create File record
captioned_file = await file_repo.create(
requester=user,
data=FileCreate(
project_id=project_id,
original_filename=captioned_filename,
path=output_path,
storage_backend="S3",
mime_type="video/mp4",
size_bytes=file_size,
file_format="mp4",
is_uploaded=True,
),
)
# Create ArtifactMediaFile record
artifact_repo = ArtifactRepository(self._session)
await artifact_repo.create(
data=ArtifactMediaFileCreate(
project_id=project_id,
file_id=captioned_file.id,
media_file_id=None,
artifact_type="RENDERED_VIDEO",
),
)
# Update job output_data with file_id so frontend can reference it
updated_output = dict(output_data)
updated_output["file_id"] = str(captioned_file.id)
job = await self._job_repo.update(
job, JobUpdate(output_data=updated_output)
)
logger.info("Saved captions artifacts for job %s (file_id=%s)", job.id, captioned_file.id)
async def submit_media_probe(
self, *, requester: User, request: MediaProbeRequest
) -> TaskSubmitResponse:
@@ -1238,6 +1625,22 @@ class TaskService:
self, *, requester: User, request: CaptionsGenerateRequest
) -> TaskSubmitResponse:
"""Submit captions generation task."""
from cpv3.modules.captions.service import CaptionPresetService
input_data = request.model_dump(mode="json")
existing_job = await self._find_duplicate_active_job(
requester=requester,
job_type=JOB_TYPE_CAPTIONS_GENERATE,
project_id=request.project_id,
input_data=input_data,
)
if existing_job is not None:
return TaskSubmitResponse(
job_id=existing_job.id,
webhook_url=_build_webhook_url(existing_job.id),
status=existing_job.status,
)
transcription_repo = TranscriptionRepository(self._session)
transcription = await transcription_repo.get_by_id(request.transcription_id)
if transcription is None:
@@ -1245,20 +1648,26 @@ class TaskService:
user_folder = get_user_folder(requester)
resolved_folder = (
f"{user_folder}/{request.folder}"
if request.folder
else f"{user_folder}/output_files"
f"{user_folder}/{request.folder}" if request.folder else f"{user_folder}/output_files"
)
# Resolve style config from preset or inline override
preset_service = CaptionPresetService(self._session)
style_config = await preset_service.resolve_style_config(
preset_id=request.preset_id,
inline_config=request.style_config,
)
return await self._submit_task(
requester=requester,
job_type=JOB_TYPE_CAPTIONS_GENERATE,
project_id=request.project_id,
input_data=request.model_dump(mode="json"),
input_data=input_data,
actor=captions_generate_actor,
actor_kwargs={
"video_s3_path": request.video_s3_path,
"folder": resolved_folder,
"transcription_json": transcription.document,
"style_config": style_config,
},
)