From 849eb0ff480665a233fa1dea8959e3b359876391 Mon Sep 17 00:00:00 2001 From: nekenny Date: Fri, 3 Apr 2026 22:18:24 +0500 Subject: [PATCH] =?UTF-8?q?feat:=20Celery=20setup=20and=20transcription=20?= =?UTF-8?q?task=20=E2=80=94=20Whisper=20+=20FFmpeg=20pipeline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- backend/app/schemas/transcription.py | 23 +++++++ backend/app/services/transcription.py | 68 ++++++++++++++++++++ backend/app/tasks/__init__.py | 0 backend/app/tasks/celery_app.py | 27 ++++++++ backend/app/tasks/transcription.py | 69 +++++++++++++++++++++ backend/requirements.txt | 1 + backend/tests/test_transcription_service.py | 41 ++++++++++++ 7 files changed, 229 insertions(+) create mode 100644 backend/app/schemas/transcription.py create mode 100644 backend/app/services/transcription.py create mode 100644 backend/app/tasks/__init__.py create mode 100644 backend/app/tasks/celery_app.py create mode 100644 backend/app/tasks/transcription.py create mode 100644 backend/tests/test_transcription_service.py diff --git a/backend/app/schemas/transcription.py b/backend/app/schemas/transcription.py new file mode 100644 index 0000000..edb90fd --- /dev/null +++ b/backend/app/schemas/transcription.py @@ -0,0 +1,23 @@ +import uuid +from datetime import datetime + +from pydantic import BaseModel + + +class TranscriptionSegment(BaseModel): + start: float + end: float + text: str + + +class TranscriptionResponse(BaseModel): + id: uuid.UUID + recording_id: uuid.UUID + text: str + segments: list[TranscriptionSegment] + language: str | None + whisper_model: str + processing_time: float | None + created_at: datetime + + model_config = {"from_attributes": True} diff --git a/backend/app/services/transcription.py b/backend/app/services/transcription.py new file mode 100644 index 0000000..ab8eb56 --- /dev/null +++ b/backend/app/services/transcription.py @@ -0,0 +1,68 @@ +import os +import subprocess +import time +import uuid + +from app.config import settings + +_model = None + + +def _get_model(): + global _model + if _model is None: + from faster_whisper import WhisperModel + _model = WhisperModel( + settings.whisper_model, + device=settings.whisper_device, + compute_type=settings.whisper_compute_type, + ) + return _model + + +def extract_audio(video_path: str) -> str: + audio_path = os.path.join( + settings.upload_dir, f"{uuid.uuid4().hex}.wav" + ) + subprocess.run( + [ + "ffmpeg", "-i", video_path, + "-vn", "-acodec", "pcm_s16le", + "-ar", "16000", "-ac", "1", + audio_path, "-y", + ], + check=True, + capture_output=True, + ) + return audio_path + + +def transcribe_audio(audio_path: str) -> dict: + model = _get_model() + start_time = time.time() + + segments_iter, info = model.transcribe( + audio_path, + beam_size=5, + language=None, + vad_filter=True, + ) + + segments = [] + full_text_parts = [] + for segment in segments_iter: + segments.append({ + "start": round(segment.start, 2), + "end": round(segment.end, 2), + "text": segment.text.strip(), + }) + full_text_parts.append(segment.text.strip()) + + processing_time = time.time() - start_time + + return { + "text": " ".join(full_text_parts), + "segments": segments, + "language": info.language, + "processing_time": round(processing_time, 2), + } diff --git a/backend/app/tasks/__init__.py b/backend/app/tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/tasks/celery_app.py b/backend/app/tasks/celery_app.py new file mode 100644 index 0000000..7dd3095 --- /dev/null +++ b/backend/app/tasks/celery_app.py @@ -0,0 +1,27 @@ +from celery import Celery +from celery.schedules import crontab + +from app.config import settings + +celery_app = Celery( + "meeting_protocol", + broker=settings.redis_url, + backend=settings.redis_url, +) + +celery_app.conf.update( + task_serializer="json", + accept_content=["json"], + result_serializer="json", + timezone="UTC", + enable_utc=True, + task_track_started=True, + beat_schedule={ + "cleanup-old-data": { + "task": "app.tasks.cleanup.cleanup_old_data", + "schedule": crontab(hour=3, minute=0), + }, + }, +) + +celery_app.autodiscover_tasks(["app.tasks"]) diff --git a/backend/app/tasks/transcription.py b/backend/app/tasks/transcription.py new file mode 100644 index 0000000..1ca706d --- /dev/null +++ b/backend/app/tasks/transcription.py @@ -0,0 +1,69 @@ +import os +import uuid + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.tasks.celery_app import celery_app +from app.config import settings +from app.models.recording import Recording, RecordingStatus +from app.models.transcription import Transcription +from app.services.transcription import extract_audio, transcribe_audio + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +sync_db_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace( + "postgresql+psycopg2", "postgresql" +) +sync_engine = create_engine(sync_db_url) +SyncSession = sessionmaker(sync_engine) + + +@celery_app.task(bind=True, name="app.tasks.transcription.transcribe_recording") +def transcribe_recording(self, recording_id: str): + rec_uuid = uuid.UUID(recording_id) + + with SyncSession() as db: + recording = db.execute( + select(Recording).where(Recording.id == rec_uuid) + ).scalar_one_or_none() + + if not recording: + return {"error": "Recording not found"} + + recording.status = RecordingStatus.processing + db.commit() + + try: + file_path = recording.file_path + audio_path = file_path + + video_formats = {"mp4", "webm", "mkv", "avi", "mov"} + if recording.format in video_formats: + audio_path = extract_audio(file_path) + + result = transcribe_audio(audio_path) + + if audio_path != file_path and os.path.exists(audio_path): + os.remove(audio_path) + + transcription = Transcription( + recording_id=rec_uuid, + text=result["text"], + segments=result["segments"], + language=result["language"], + whisper_model=settings.whisper_model, + processing_time=result["processing_time"], + ) + db.add(transcription) + recording.status = RecordingStatus.done + db.commit() + db.refresh(transcription) + + return {"transcription_id": str(transcription.id), "status": "done"} + + except Exception as e: + recording.status = RecordingStatus.error + db.commit() + raise self.retry(exc=e, max_retries=1, countdown=30) diff --git a/backend/requirements.txt b/backend/requirements.txt index f746c51..c2185dd 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -6,6 +6,7 @@ python-multipart==0.0.20 # Database sqlalchemy[asyncio]==2.0.36 asyncpg==0.30.0 +psycopg2-binary==2.9.10 alembic==1.14.1 # Validation diff --git a/backend/tests/test_transcription_service.py b/backend/tests/test_transcription_service.py new file mode 100644 index 0000000..93884ab --- /dev/null +++ b/backend/tests/test_transcription_service.py @@ -0,0 +1,41 @@ +import os +import pytest +from unittest.mock import patch, MagicMock + +from app.services.transcription import extract_audio, transcribe_audio + + +@pytest.mark.asyncio +async def test_extract_audio_calls_ffmpeg(): + with patch("app.services.transcription.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + result = extract_audio("/tmp/test.mp4") + assert result.endswith(".wav") + mock_run.assert_called_once() + args = mock_run.call_args + assert "ffmpeg" in args[0][0] + + +@pytest.mark.asyncio +async def test_transcribe_audio_returns_expected_structure(): + mock_segment = MagicMock() + mock_segment.start = 0.0 + mock_segment.end = 5.0 + mock_segment.text = " Hello world " + + mock_info = MagicMock() + mock_info.language = "en" + + with patch("app.services.transcription._get_model") as mock_model: + mock_model.return_value.transcribe.return_value = ( + iter([mock_segment]), + mock_info, + ) + result = transcribe_audio("/tmp/test.wav") + + assert "text" in result + assert result["text"] == "Hello world" + assert len(result["segments"]) == 1 + assert result["segments"][0]["start"] == 0.0 + assert result["language"] == "en" + assert "processing_time" in result