"""Оркестрация AI-пайплайна пост-обработки сеанса. Единственная задача-диспетчер `run_pipeline(session_id)`: смотрит текущее `pipeline_status` сеанса и эффективную конфигурацию инстанса (`services.instance_settings.load_effective_config` — настройки из БД поверх дефолтов `config/plugins.yaml`), продолжает работу с последнего успешного шага. Ожидание завершения записи треков (`egress_ended` приходит позже `room_finished`) — через celery-retry с backoff внутри самой задачи. Сама суммаризация здесь не выполняется: `run_pipeline` доводит сеанс до `pipeline_status='summarizing'` и передаёт эстафету задаче `workers.tasks.summarize.summarize_session` — по имени через `app.send_task`, без импорта модуля суммаризации (транскрайбер- процесс не должен тянуть его код). Надёжность постановки `summarize_session`: временная недоступность брокера Redis в момент `app.send_task` не должна ронять `run_pipeline` — фразы к этому моменту уже закоммичены, и `failed` из-за одного лишь сбоя постановки задачи стёр бы уже проделанную работу транскрибации. Поэтому: (1) общий хелпер `workers.tasks.dispatch. send_task_with_retry` делает несколько попыток с коротким backoff (переиспользуется и `summarize_session` для постановки `notify_session`); (2) если все попытки исчерпаны — ошибка логируется, но `pipeline_status` остаётся `summarizing` (не откатывается, не переводится в `failed`), а восстановление берёт на себя периодическая задача `workers.tasks. maintenance.recover_stuck_summaries` (уровень 2 защиты). """ import logging import uuid from datetime import timedelta from pathlib import Path from typing import Protocol from celery.exceptions import MaxRetriesExceededError from sqlalchemy import delete from core.plugins.config import InstanceConfig from core.plugins.factory import create_transcriber from core.plugins.transcriber import Segment from models.audio_track import SessionAudioTrack from models.phrase import Phrase from models.session import ConferenceSession from repositories.conferences import AudioTrackRepository from services.instance_settings import load_effective_config from workers.celery_app import app as app from workers.db import open_session, run_async from workers.tasks.dispatch import send_task_with_retry from workers.transcription.phrases import build_phrases logger = logging.getLogger(__name__) RETRY_COUNTDOWN_S = 30 """Пауза (сек) перед повторной попыткой, пока треки ещё дописываются egress'ом.""" _TRANSCRIBABLE_PIPELINE_STATUSES = frozenset({"recording", "transcribing"}) """Статусы сеанса, с которых допустим (повторный) запуск шага транскрибации — guard идемпотентности: если пайплайн уже ушёл дальше (`summarizing` и позже) или зафиксирован как `failed`, повторный вызов задачи — no-op.""" class RetryableTask(Protocol): """Минимальный протокол объекта задачи с методом `retry` (для тестируемости). Реальный `celery.Task` (bound self) удовлетворяет протоколу структурно — отдельный импорт `celery.Task` как типа не нужен. """ def retry(self, countdown: int | None = None) -> None: ... @app.task(name="workers.tasks.pipeline.run_pipeline", bind=True, max_retries=20, acks_late=True) def run_pipeline(self: RetryableTask, session_id: str) -> None: """Точка входа Celery — синхронная обёртка над асинхронной логикой диспетчера.""" run_async(lambda: run_pipeline_async(self, uuid.UUID(session_id))) async def run_pipeline_async( task: RetryableTask, session_id: uuid.UUID, *, plugins_config: InstanceConfig | None = None, ) -> None: """Диспетчер: продолжить AI-пайплайн сеанса `session_id` с последнего успешного шага. Шаги: 1. Сеанс не найден либо ещё не завершён (`t_end IS NULL`) — выход. 2. `transcriber.enabled=false` — выход, `pipeline_status` не меняется. 3. Пайплайн уже прошёл шаг транскрибации (`pipeline_status` не в `{recording, transcribing}`) — выход (идемпотентность повторного вызова). 4. Есть треки со статусом `recording` (egress ещё пишет) — `task.retry`; исчерпание попыток — зависшие треки помечаются `failed`, работа продолжается с остальными треками. 5. `pipeline_status='transcribing'`, commit. 6. Транскрибация треков со статусом `recorded` и `segments IS NULL` (уже транскрибированные при прошлом прогоне — пропускаются); commit ПОСЛЕ КАЖДОГО трека — точка возобновления при падении процесса посередине (acks_late). 7. Ни одного трека не транскрибировано (все failed либо треков нет) — `pipeline_status='failed'`, выход. 8. Реконструкция фраз (`build_phrases`, ТЗ §1.3) → DELETE+INSERT `phrases` одной транзакцией → `pipeline_status='summarizing'`, commit → постановка задачи `workers.tasks.summarize.summarize_session` в очередь по умолчанию (её слушает базовый `worker`, а не `worker-transcriber`) — через общий `workers.tasks.dispatch.send_task_with_retry` (retry на сбой брокера, см. его докстринг и `workers.tasks.maintenance.recover_stuck_summaries`). """ async with open_session() as session: session_record = await session.get(ConferenceSession, session_id) if session_record is None or session_record.t_end is None: logger.warning( "run_pipeline: сеанс %s не найден либо ещё не завершён (t_end IS NULL)", session_id, ) return cfg = plugins_config or await load_effective_config(session) if not cfg.transcriber.enabled: logger.info( "run_pipeline: transcriber отключён (enabled=false) — сеанс %s пропущен", session_id, ) return if session_record.pipeline_status not in _TRANSCRIBABLE_PIPELINE_STATUSES: logger.info( "run_pipeline: сеанс %s уже прошёл шаг транскрибации (pipeline_status=%s) — no-op", session_id, session_record.pipeline_status, ) return track_repo = AudioTrackRepository(session) # Сортировка по `started_at` — детерминированный порядок обработки # (точка возобновления по-трекового коммита должна быть # предсказуемой между прогонами, см. тест идемпотентности №12). tracks = sorted(await track_repo.list_by_session(session_id), key=lambda t: t.started_at) still_recording = [track for track in tracks if track.status == "recording"] if still_recording: try: task.retry(countdown=RETRY_COUNTDOWN_S) except MaxRetriesExceededError: for track in still_recording: track.status = "failed" await session.commit() logger.warning( "run_pipeline: исчерпаны попытки ожидания egress для %d треков сеанса %s " "— помечены failed", len(still_recording), session_id, ) else: # Реальный `Task.retry()` сам бросает исключение Retry (не # возвращает управление) — сюда попадаем только с # моком/заглушкой `task.retry` в тестах. return session_record.pipeline_status = "transcribing" await session.commit() transcriber = create_transcriber(cfg.transcriber) for track in tracks: if track.status != "recorded" or track.segments is not None: continue # уже транскрибирован на прошлом прогоне — идемпотентность if not track.file_path or not Path(track.file_path).exists(): track.status = "failed" logger.warning( "run_pipeline: файл трека %s не найден (%s) — трек помечен failed", track.id, track.file_path, ) await session.commit() continue segments = transcriber.transcribe(track.file_path, cfg.transcriber.language) track.segments = [ {"start": segment.start, "end": segment.end, "text": segment.text} for segment in segments ] track.status = "transcribed" await session.commit() if not any(track.status == "transcribed" for track in tracks): session_record.pipeline_status = "failed" await session.commit() logger.warning( "run_pipeline: все треки сеанса %s провалены либо треков нет — pipeline failed", session_id, ) return segments_by_participant, track_offsets = _collect_transcribed(tracks, session_record) phrases = build_phrases(segments_by_participant, track_offsets) await session.execute(delete(Phrase).where(Phrase.session_id == session_id)) for phrase in phrases: session.add( Phrase( participant_id=phrase.participant_id, session_id=session_id, data=phrase.text, t_start=session_record.t_start + timedelta(seconds=phrase.start), t_end=session_record.t_start + timedelta(seconds=phrase.end), ) ) session_record.pipeline_status = "summarizing" await session.commit() # По имени задачи, без импорта `workers.tasks.summarize` — модуль # суммаризации не должен становиться зависимостью # транскрайбер-процесса. Отправляем безусловно: если # `summarizer.enabled=false`, задача сама завершится по своему guard'у # (фразы уже сохранены, `pipeline_status='summarizing'` без summary — # задокументированное завершение пайплайна после phrases). sent = await send_task_with_retry( "workers.tasks.summarize.summarize_session", args=[str(session_id)] ) logger.info( "run_pipeline: сеанс %s — реконструировано %d фраз, статус=summarizing, " "постановка задачи суммаризации %s", session_id, len(phrases), "выполнена" if sent else "не удалась (см. предыдущий error) — ждём recovery", ) def _collect_transcribed( tracks: list[SessionAudioTrack], session_record: ConferenceSession ) -> tuple[dict[uuid.UUID, list[Segment]], dict[uuid.UUID, float]]: """Собрать сегменты и смещения транскрибированных треков для `build_phrases`. Смещение трека — `(track.started_at - session.t_start).total_seconds()` («Ключевые архитектурные решения», п.5). Ключ обоих словарей — `participant_id`: одному участнику соответствует один аудиотрек сеанса (одно окно присутствия — один микрофон, ADR-002). """ segments_by_participant: dict[uuid.UUID, list[Segment]] = {} track_offsets: dict[uuid.UUID, float] = {} for track in tracks: if track.status != "transcribed" or not track.segments: continue offset = (track.started_at - session_record.t_start).total_seconds() track_offsets[track.participant_id] = offset segments_by_participant.setdefault(track.participant_id, []).extend( Segment(start=raw["start"], end=raw["end"], text=raw["text"]) for raw in track.segments ) return segments_by_participant, track_offsets