diff --git a/.env.example b/.env.example index 40a7a89..7186b48 100644 --- a/.env.example +++ b/.env.example @@ -81,6 +81,15 @@ S3_PRESIGN_EXPIRY=3600 # Воркер MAX_CONCURRENT_TASKS=2 +# Очистка диска (необязательно; значения ниже — дефолты). +# TTL workspace упавших задач; кеш транскриптов с ним не связан. +# WORK_DIR_CLEANUP_ENABLED=true +# WORK_DIR_FAILED_RETENTION_HOURS=24 +# WORK_DIR_SWEEP_INTERVAL_MINUTES=60 +# Срок хранения неиспользуемой записи кеша транскриптов (дни, максимум 365, +# не меньше WORK_DIR_FAILED_RETENTION_HOURS). +# TRANSCRIPT_CACHE_RETENTION_DAYS=30 + # Вебхук на терминальные события (опционально; без URL движок работает автономно) # PLATFORM_CALLBACK_URL=https://platform.example/lecturelog/callback # LECTURELOG_WEBHOOK_SECRET= diff --git a/README.md b/README.md index b54ce37..b665477 100644 --- a/README.md +++ b/README.md @@ -358,6 +358,11 @@ python scripts/submit_task.py --base http://my-host:8000/api/v1 status | `DEEPGRAM_LANGUAGE` | Язык Deepgram (по умолчанию `ru`). | | `DEEPGRAM_DETECT_LANGUAGE` | Автоопределение доминирующего языка (`true`/`false`). При `true` фиксированный `DEEPGRAM_LANGUAGE` не отправляется. | | `DEEPGRAM_UTT_SPLIT` | Порог паузы utterance в секундах (по умолчанию `0.8`). | +| `TRANSCRIPT_CACHE_DIR` | Кеш транскриптов (перезапуск без повторной транскрипции). По умолчанию `$WORK_DIR/transcript_cache` (в compose это именованный volume `lecturelog-core-transcript-cache`, переживает деплой). Ключ: хеш аудио + провайдер/модель/язык/utt_split. | +| `WORK_DIR_CLEANUP_ENABLED` | TTL-очистка workspace упавших задач (по умолчанию `true`). | +| `WORK_DIR_FAILED_RETENTION_HOURS` | Срок хранения workspace упавшей задачи в часах (по умолчанию `24`). Кеш транскриптов с этим сроком не связан. | +| `TRANSCRIPT_CACHE_RETENTION_DAYS` | Срок хранения записи кеша транскриптов в днях (по умолчанию `30`, максимум `365`). Записи, не использованные дольше срока (mtime; обновляется при каждом попадании), удаляет тот же sweeper. Не может быть меньше `WORK_DIR_FAILED_RETENTION_HOURS`. | +| `WORK_DIR_SWEEP_INTERVAL_MINUTES` | Период прохода sweeper (по умолчанию `60`). | | `OPENROUTER_API_KEY` | Ключ OpenRouter; LLM-вызовы идут через BYOK Google AI Studio. | | `OPENROUTER_BASE_URL` | Base URL OpenRouter (по умолчанию `https://openrouter.ai/api/v1`). | | `VIDEO_TARGET_RESOLUTION` | Целевое разрешение URL-видео: `144..4320` либо `best`; по умолчанию `720`. | diff --git a/deploy/compose.vps.yml b/deploy/compose.vps.yml index af3fd2f..5abccda 100644 --- a/deploy/compose.vps.yml +++ b/deploy/compose.vps.yml @@ -11,6 +11,9 @@ services: S3_BUCKET: ${S3_BUCKET:-lectures} S3_ACCESS_KEY: ${S3_ACCESS_KEY:?S3_ACCESS_KEY is required} S3_SECRET_KEY: ${S3_SECRET_KEY:?S3_SECRET_KEY is required} + volumes: + # Только кеш транскриптов: переживает пересоздание контейнера при деплое. + - lecturelog-core-transcript-cache:/app/data/transcript_cache ports: - 127.0.0.1:${CORE_HTTP_PORT:-8000}:8000 depends_on: @@ -109,3 +112,5 @@ volumes: name: lecturelog-core-postgres-data lecturelog-core-minio-data: name: lecturelog-core-minio-data + lecturelog-core-transcript-cache: + name: lecturelog-core-transcript-cache diff --git a/deploy/env.core.example b/deploy/env.core.example index bcc19e5..8855853 100644 --- a/deploy/env.core.example +++ b/deploy/env.core.example @@ -61,6 +61,15 @@ S3_PRESIGN_EXPIRY=3600 MAX_CONCURRENT_TASKS=2 +# Очистка диска (необязательно; значения ниже — дефолты). +# TTL workspace упавших задач; кеш транскриптов с ним не связан. +# WORK_DIR_CLEANUP_ENABLED=true +# WORK_DIR_FAILED_RETENTION_HOURS=24 +# WORK_DIR_SWEEP_INTERVAL_MINUTES=60 +# Срок хранения неиспользуемой записи кеша транскриптов (дни, максимум 365, +# не меньше WORK_DIR_FAILED_RETENTION_HOURS). +# TRANSCRIPT_CACHE_RETENTION_DAYS=30 + # URL web-платформы для вебхука ядра. Это не Google OAuth callback. PLATFORM_CALLBACK_URL=https://app.example.com/webhooks/core diff --git a/docker-compose.yml b/docker-compose.yml index 74335ed..c6b4e25 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -72,9 +72,15 @@ services: S3_SECRET_KEY: ${S3_SECRET_KEY:?S3_SECRET_KEY is required} # S3_PUBLIC_ENDPOINT не задан -> presigned выключен (автономный режим): # /uploads и /result-url отдают 409, стрим GET /result работает. + volumes: + # Только кеш транскриптов: он должен переживать пересоздание контейнера + # (деплой нового образа), иначе перезапуск лекции снова платит за STT. + # Остальной WORK_DIR остаётся во writable layer. + - lecturelog-core-transcript-cache:/app/data/transcript_cache ports: - "8000:8000" volumes: pgdata: miniodata: + lecturelog-core-transcript-cache: diff --git a/docs/api-contract.md b/docs/api-contract.md index 4348467..6578b13 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -109,6 +109,14 @@ public-endpoint, иначе `409`. навигацию и собственное представление строит платформа. Способы забрать результат — см. «Два способа забрать результат». +## Транскрипт упавшей задачи + +`GET /tasks/{id}/transcript` для задачи в статусе `failed`, у которой транскрипта нет +(не дошла до STT либо workspace удалён по TTL `WORK_DIR_FAILED_RETENTION_HOURS`), отвечает +`410 Gone` с телом `{"error": "transcript_unavailable", "detail": }`, а не +`202 in_progress`. Сбой самой транскрибации по-прежнему даёт `409 transcribe_failed`. +Клиент не должен поллить `410`. + ## Удаление задачи (идемпотентно) `DELETE /tasks/{id}` инкапсулирует физику хранения: платформа не знает ключей и diff --git a/docs/openapi.json b/docs/openapi.json index 7c33f06..545aff5 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -403,6 +403,30 @@ "title": "TranscriptFailedError", "type": "object" }, + "TranscriptGoneError": { + "properties": { + "detail": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Detail" + }, + "error": { + "title": "Error", + "type": "string" + } + }, + "required": [ + "error" + ], + "title": "TranscriptGoneError", + "type": "object" + }, "TranscriptInvalidFormatError": { "properties": { "allowed": { @@ -915,6 +939,16 @@ }, "description": "Транскрибация завершилась ошибкой" }, + "410": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TranscriptGoneError" + } + } + }, + "description": "Задача упала, транскрипта нет (не дошла до STT или удалён по TTL)" + }, "422": { "content": { "application/json": { diff --git a/lecturelog/api/lifespan.py b/lecturelog/api/lifespan.py index d1c0d6b..bd7be22 100644 --- a/lecturelog/api/lifespan.py +++ b/lecturelog/api/lifespan.py @@ -1,8 +1,10 @@ from __future__ import annotations +import asyncio import logging import os from contextlib import asynccontextmanager +from datetime import timedelta from pathlib import Path from fastapi import FastAPI @@ -15,7 +17,7 @@ ) from lecturelog.application.pipeline_service import PipelineService from lecturelog.application.progress_plan import ProgressPlan -from lecturelog.application.worker import PipelineWorker +from lecturelog.application.worker import PipelineWorker, WorkDirSweeper from lecturelog.config.settings import get_config from lecturelog.infrastructure.export.obsidian_exporter import ObsidianExporter from lecturelog.infrastructure.frames.provider import VideoFrameProvider @@ -32,6 +34,15 @@ logger = logging.getLogger(__name__) +async def _sweep_forever(sweeper: WorkDirSweeper, interval_seconds: float) -> None: + while True: + try: + await sweeper.sweep() + except Exception: + logger.exception("Sweeper: проход завершился ошибкой") + await asyncio.sleep(interval_seconds) + + @asynccontextmanager async def lifespan(app: FastAPI): cfg = get_config() @@ -53,7 +64,12 @@ async def lifespan(app: FastAPI): cooldown = ModelCooldown() llm = LlmClient(openai_client, cooldown, max_tokens=cfg.llm.max_tokens) - transcriber = transcriber_factory(cfg.transcribe) + # Локальный эфемерный scratch для внутренних стадий пайплайна (не S3). + work_dir = Path(os.getenv("WORK_DIR", "/app/data")) + # Кеш транскриптов лежит рядом со scratch, но вне каталогов задач: его не сносит + # очистка workspace задачи. + transcript_cache_dir = cfg.transcribe.cache_dir or work_dir / "transcript_cache" + transcriber = transcriber_factory(cfg.transcribe, cache_dir=transcript_cache_dir) transcribe_model = ( "whisper-large-v3" if cfg.transcribe.provider == "groq" else cfg.transcribe.deepgram_model ) @@ -139,19 +155,34 @@ def frames_factory(video_path: Path, srt_path: Path) -> VideoFrameProvider: worker = PipelineWorker(service=service, concurrency=cfg.worker.max_concurrent_tasks) await worker.start() + sweeper_task: asyncio.Task | None = None + if cfg.worker.cleanup_enabled: + sweeper = WorkDirSweeper( + repository=repo, + work_root=work_dir, + failed_retention=timedelta(hours=cfg.worker.failed_retention_hours), + transcript_cache_dir=transcript_cache_dir, + cache_retention=timedelta(days=cfg.worker.transcript_cache_retention_days), + ) + sweeper_task = asyncio.create_task( + _sweep_forever(sweeper, cfg.worker.sweep_interval_minutes * 60) + ) + app.state.config = cfg app.state.repository = repo app.state.worker = worker app.state.storage = storage app.state.cookie_store = cookie_store app.state.presign_expiry = cfg.s3.presign_expiry - # Локальный эфемерный scratch для внутренних стадий пайплайна (не S3). - app.state.work_dir = Path(os.getenv("WORK_DIR", "/app/data")) + app.state.work_dir = work_dir app.state.llm = llm app.state.prompts_dir = Path("prompts") try: yield finally: + if sweeper_task is not None: + sweeper_task.cancel() + await asyncio.gather(sweeper_task, return_exceptions=True) await worker.stop() # Закрываем HTTP-клиент OpenRouter, чтобы не оставлять открытые сокеты. await openai_client.close() diff --git a/lecturelog/api/routes.py b/lecturelog/api/routes.py index 9d29e96..4b82668 100644 --- a/lecturelog/api/routes.py +++ b/lecturelog/api/routes.py @@ -24,6 +24,7 @@ ResultUrlResponse, TaskStatusResponse, TranscriptFailedError, + TranscriptGoneError, TranscriptInvalidFormatError, TranscriptNotFoundError, UploadUrlRequest, @@ -318,6 +319,10 @@ async def delete_task( "model": TranscriptFailedError, "description": "Транскрибация завершилась ошибкой", }, + 410: { + "model": TranscriptGoneError, + "description": "Задача упала, транскрипта нет (не дошла до STT или удалён по TTL)", + }, }, ) async def get_task_transcript( @@ -375,6 +380,13 @@ async def get_task_transcript( headers={"Content-Disposition": 'attachment; filename="transcript.txt"'}, ) + if not result.ready and task.status == TaskStatus.FAILED: + # Упавшая задача транскрипт не получит: не маскируем под in_progress. + return JSONResponse( + status_code=410, + content={"error": "transcript_unavailable", "detail": task.error}, + ) + if not result.ready: return JSONResponse( status_code=202, diff --git a/lecturelog/api/schemas.py b/lecturelog/api/schemas.py index df0ab7e..ebb54c6 100644 --- a/lecturelog/api/schemas.py +++ b/lecturelog/api/schemas.py @@ -99,6 +99,12 @@ class TranscriptFailedError(BaseModel): detail: str | None = None +class TranscriptGoneError(BaseModel): + # 410: {"error": "transcript_unavailable", "detail": }. + error: str + detail: str | None = None + + class TaskStatusResponse(BaseModel): task_id: str stage: str | None diff --git a/lecturelog/application/factories.py b/lecturelog/application/factories.py index 7807b2b..0cee915 100644 --- a/lecturelog/application/factories.py +++ b/lecturelog/application/factories.py @@ -1,5 +1,8 @@ from __future__ import annotations +from pathlib import Path +from typing import Any + from lecturelog.config.settings import S3Config, TranscribeConfig from lecturelog.domain.media_source import MediaSource, is_video_source from lecturelog.domain.ports import ( @@ -10,8 +13,9 @@ WebhookNotifier, ) from lecturelog.infrastructure.storage.s3_storage import S3Storage +from lecturelog.infrastructure.transcribe.cache import CachingTranscriber from lecturelog.infrastructure.transcribe.deepgram_transcriber import DeepgramTranscriber -from lecturelog.infrastructure.transcribe.groq_transcriber import GroqTranscriber +from lecturelog.infrastructure.transcribe.groq_transcriber import WHISPER_MODEL, GroqTranscriber from lecturelog.infrastructure.webhook.http_notifier import HttpWebhookNotifier @@ -55,7 +59,31 @@ def storage_factory(s3: S3Config) -> Storage: ) -def transcriber_factory(config: TranscribeConfig) -> Transcriber: +def transcribe_cache_settings(config: TranscribeConfig) -> dict[str, Any]: + """Настройки, влияющие на содержимое транскрипта, — часть ключа кеша. + Секреты и endpoint-регион (не меняют текст) сюда не входят.""" + if config.provider == "groq": + return {"provider": "groq", "model": WHISPER_MODEL} + return { + "provider": "deepgram", + "model": config.deepgram_model, + "language": "auto" if config.deepgram_detect_language else config.deepgram_language, + "detect_language": config.deepgram_detect_language, + "utt_split": config.deepgram_utt_split, + } + + +def transcriber_factory(config: TranscribeConfig, *, cache_dir: Path | None = None) -> Transcriber: + """Транскрибер по конфигу; с cache_dir — обёрнутый кешем транскриптов.""" + transcriber = _build_transcriber(config) + if cache_dir is None: + return transcriber + return CachingTranscriber( + transcriber, cache_dir=cache_dir, settings=transcribe_cache_settings(config) + ) + + +def _build_transcriber(config: TranscribeConfig) -> Transcriber: if config.provider == "groq": return GroqTranscriber(groq_api_keys=config.groq_keys) assert config.deepgram_api_key is not None diff --git a/lecturelog/application/worker.py b/lecturelog/application/worker.py index 9fec23f..9912b04 100644 --- a/lecturelog/application/worker.py +++ b/lecturelog/application/worker.py @@ -1,20 +1,192 @@ from __future__ import annotations import asyncio +import json import logging +import os import re import shutil from collections.abc import Callable from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from pathlib import Path +from lecturelog.domain.enums import TaskStatus from lecturelog.domain.media_source import MediaSource from lecturelog.domain.models import Task -from lecturelog.domain.ports import SlideProvider +from lecturelog.domain.ports import SlideProvider, TaskRepository logger = logging.getLogger(__name__) _TASK_ID_RE = re.compile(r"[0-9a-f]{32}\Z") +# Запись кеша транскриптов: каталог с именем-хешем sha256. +_CACHE_ENTRY_RE = re.compile(r"[0-9a-f]{64}\Z") + +# Служебный marker терминального состояния внутри workspace задачи: TTL считается по нему, +# а не по mtime медиафайлов. Каталоги-соседи задач (results_tmp, кеш транскриптов) sweeper +# не трогает: кеш обязан пережить workspace упавшей задачи, иначе перезапуск заплатит за STT. +TERMINAL_MARKER = ".terminal.json" +RESERVED_DIRS = frozenset({"results_tmp", "transcript_cache"}) + +Clock = Callable[[], datetime] + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +def _is_task_id(name: str) -> bool: + """Имя - task_id (32 hex): единственная проверка для cleanup воркера и sweeper.""" + return _TASK_ID_RE.fullmatch(name) is not None + + +def _safe_task_dir(path: Path, work_root: Path) -> bool: + """Каталог задачи: имя - task_id, не symlink, прямой потомок work_root.""" + if not _is_task_id(path.name) or path.is_symlink(): + return False + return path.resolve().parent == work_root.resolve() + + +def write_terminal_marker(work_dir: Path, status: TaskStatus, clock: Clock = utc_now) -> None: + """Атомарно записать marker (статус и UTC-время). Best-effort, без текста ошибки.""" + payload = json.dumps({"status": status.value, "at": clock().astimezone(UTC).isoformat()}) + tmp = work_dir / (TERMINAL_MARKER + ".tmp") + try: + tmp.write_text(payload, encoding="utf-8") + os.replace(tmp, work_dir / TERMINAL_MARKER) + except OSError as exc: + logger.warning("Воркер: не записан marker work_dir=%s: %s", work_dir, exc) + + +def _read_terminal_time(work_dir: Path) -> datetime | None: + try: + raw = json.loads((work_dir / TERMINAL_MARKER).read_text(encoding="utf-8")) + at = datetime.fromisoformat(raw["at"]) + except (OSError, ValueError, KeyError, TypeError): + return None + return at if at.tzinfo else at.replace(tzinfo=UTC) + + +@dataclass +class SweepResult: + scanned: int = 0 + deleted: int = 0 + skipped: int = 0 + cache_deleted: int = 0 + + +class WorkDirSweeper: + """Страховочная TTL-очистка workspace упавших задач (расширение очистки воркера). + + Удаляет только каталоги задач в статусе FAILED, чей marker старше срока хранения. + PENDING/PROCESSING/DONE/INTERRUPTED и каталоги без строки в БД не трогает. + + Если задан `transcript_cache_dir` с `cache_retention`, удаляет и записи кеша + транскриптов, не использовавшиеся дольше срока (по mtime записи: кеш + обновляет его при каждом попадании). Срок кеша не меньше срока workspace + упавшей задачи - это проверяет конфиг, - иначе перезапуск не нашёл бы транскрипт. + """ + + def __init__( + self, + repository: TaskRepository, + work_root: Path, + failed_retention: timedelta, + clock: Clock = utc_now, + *, + transcript_cache_dir: Path | None = None, + cache_retention: timedelta | None = None, + ): + self._repo = repository + self._root = work_root + self._retention = failed_retention + self._clock = clock + self._cache_dir = transcript_cache_dir + self._cache_retention = cache_retention + self._lock = asyncio.Lock() + + async def sweep(self) -> SweepResult: + result = SweepResult() + if self._lock.locked(): # параллельные проходы запрещены + return result + async with self._lock: + try: + entries = await asyncio.to_thread(lambda: sorted(self._root.iterdir())) + except OSError as exc: + logger.warning("Sweeper: не прочитан WORK_DIR %s: %s", self._root, exc) + return result + for entry in entries: + if entry.name in RESERVED_DIRS: + continue + result.scanned += 1 + try: + deleted = await self._sweep_one(entry) + except Exception as exc: # ошибка одного каталога не останавливает проход + logger.warning("Sweeper: сбой на %s: %s", entry, exc) + deleted = False + if deleted: + result.deleted += 1 + else: + result.skipped += 1 + result.cache_deleted = await self._sweep_cache() + logger.info( + "Sweeper: scanned=%d deleted=%d skipped=%d cache_deleted=%d", + result.scanned, + result.deleted, + result.skipped, + result.cache_deleted, + ) + return result + + async def _sweep_cache(self) -> int: + """Удалить записи кеша транскриптов старше срока; сбой записи проход не прерывает.""" + if self._cache_dir is None or self._cache_retention is None: + return 0 + try: + entries = await asyncio.to_thread(lambda: sorted(self._cache_dir.iterdir())) + except OSError as exc: + if not isinstance(exc, FileNotFoundError): + logger.warning("Sweeper: не прочитан кеш транскриптов %s: %s", self._cache_dir, exc) + return 0 + now = self._clock() + deleted = 0 + for entry in entries: + try: + if ( + not _CACHE_ENTRY_RE.fullmatch(entry.name) + or entry.is_symlink() + or not entry.is_dir() + ): + continue + used_at = datetime.fromtimestamp(entry.stat().st_mtime, UTC) + if now - used_at < self._cache_retention: + continue + await asyncio.to_thread(shutil.rmtree, entry) + logger.info("Sweeper: удалена устаревшая запись кеша транскриптов %s", entry.name) + deleted += 1 + except Exception as exc: # ошибка одной записи не останавливает проход + logger.warning("Sweeper: сбой на записи кеша %s: %s", entry, exc) + return deleted + + async def _sweep_one(self, entry: Path) -> bool: + if not _safe_task_dir(entry, self._root): + return False + task = await self._repo.get(entry.name) + if task is None or task.status != TaskStatus.FAILED: + return False + now = self._clock() + at = _read_terminal_time(entry) + if at is not None and at > now: + logger.warning("Sweeper: marker из будущего, откат на mtime task=%s", entry.name) + at = None + if at is None: + # Marker нет/битый/из будущего: считаем от mtime каталога. + at = datetime.fromtimestamp(entry.stat().st_mtime, UTC) + if now - at < self._retention: + return False + await asyncio.to_thread(shutil.rmtree, entry) + logger.info("Sweeper: удалён workspace упавшей задачи task=%s", entry.name) + return True @dataclass @@ -30,8 +202,9 @@ class PipelineJob: class PipelineWorker: - def __init__(self, service, concurrency: int): + def __init__(self, service, concurrency: int, clock: Clock = utc_now): self._service = service + self._clock = clock self._concurrency = concurrency self._queue: asyncio.Queue[PipelineJob] = asyncio.Queue() self._consumers: list[asyncio.Task] = [] @@ -46,7 +219,7 @@ async def _cleanup_completed_work_dir(self, job: PipelineJob, result_path: str) """Best-effort удалить scratch только после подтверждённого S3-результата.""" if result_path != f"results/{job.task_id}/": return - if not _TASK_ID_RE.fullmatch(job.task_id) or job.work_dir.name != job.task_id: + if not _is_task_id(job.task_id) or job.work_dir.name != job.task_id: logger.error( "Воркер: небезопасный work_dir для cleanup task=%s path=%s", job.task_id, @@ -86,6 +259,10 @@ async def _consume(self) -> None: await self._cleanup_completed_work_dir(job, result_path) except Exception as exc: # задача уже помечена FAILED в repo logger.warning("Воркер: задача %s завершилась ошибкой: %s", job.task_id, exc) + if job.work_dir.is_dir() and _safe_task_dir(job.work_dir, job.work_dir.parent): + await asyncio.to_thread( + write_terminal_marker, job.work_dir, TaskStatus.FAILED, self._clock + ) finally: self._queue.task_done() diff --git a/lecturelog/config/settings.py b/lecturelog/config/settings.py index 25b012f..25dc939 100644 --- a/lecturelog/config/settings.py +++ b/lecturelog/config/settings.py @@ -1,6 +1,7 @@ from __future__ import annotations from functools import cached_property, lru_cache +from pathlib import Path from typing import Literal from urllib.parse import urlsplit @@ -24,6 +25,9 @@ class TranscribeConfig(BaseSettings): deepgram_language: str = Field("ru", alias="DEEPGRAM_LANGUAGE") deepgram_detect_language: bool = Field(False, alias="DEEPGRAM_DETECT_LANGUAGE") deepgram_utt_split: float = Field(0.8, alias="DEEPGRAM_UTT_SPLIT", gt=0) + # Кеш транскриптов для перезапуска без повторной оплаты STT. Пусто — каталог + # transcript_cache внутри WORK_DIR (вне workspace отдельных задач). + cache_dir: Path | None = Field(None, alias="TRANSCRIPT_CACHE_DIR") @model_validator(mode="after") def validate_provider(self) -> TranscribeConfig: @@ -177,6 +181,29 @@ class S3Config(BaseSettings): class WorkerConfig(BaseSettings): model_config = _BASE max_concurrent_tasks: int = Field(2, alias="MAX_CONCURRENT_TASKS") + # TTL-очистка workspace упавших задач (см. docs/plans/2026-07-15-disk-cleanup-design.md). + cleanup_enabled: bool = Field(True, alias="WORK_DIR_CLEANUP_ENABLED") + failed_retention_hours: float = Field( + 24, alias="WORK_DIR_FAILED_RETENTION_HOURS", ge=0, le=24 * 3650 + ) + sweep_interval_minutes: float = Field( + 60, alias="WORK_DIR_SWEEP_INTERVAL_MINUTES", gt=0, le=60 * 24 * 365 + ) + # Срок хранения неиспользуемой записи кеша транскриптов (по mtime записи). + # Не меньше срока workspace упавшей задачи: кеш для того и нужен, чтобы + # перезапуск после удаления workspace не платил за STT повторно. + transcript_cache_retention_days: float = Field( + 30, alias="TRANSCRIPT_CACHE_RETENTION_DAYS", gt=0, le=365 + ) + + @model_validator(mode="after") + def validate_cache_retention(self) -> WorkerConfig: + if self.transcript_cache_retention_days * 24 < self.failed_retention_hours: + raise ValueError( + "TRANSCRIPT_CACHE_RETENTION_DAYS (в часах) не может быть меньше " + "WORK_DIR_FAILED_RETENTION_HOURS" + ) + return self class MediaConfig(BaseSettings): diff --git a/lecturelog/infrastructure/transcribe/cache.py b/lecturelog/infrastructure/transcribe/cache.py new file mode 100644 index 0000000..88f1378 --- /dev/null +++ b/lecturelog/infrastructure/transcribe/cache.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import asyncio +import contextlib +import hashlib +import json +import logging +import os +import tempfile +from pathlib import Path +from typing import Any + +from lecturelog.domain.ports import ProgressCallback, Transcriber, UsageCallback +from lecturelog.infrastructure.transcribe.common import emit_progress + +logger = logging.getLogger(__name__) + +# Версия формата записи и постобработки SRT: поднять, если меняется то, как из ответа +# провайдера получается транскрипт (старые записи тогда перестанут находиться). +CACHE_FORMAT_VERSION = 1 +_CHUNK = 1024 * 1024 + + +def _hash_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + while chunk := source.read(_CHUNK): + digest.update(chunk) + return digest.hexdigest() + + +class CachingTranscriber(Transcriber): + """Переиспользует транскрипт по ключу «содержимое аудио + настройки транскрипции». + + Запись кладётся в `cache_dir/<ключ>/transcript.srt` сразу после успешной + транскрипции, то есть до всех следующих стадий: падение задачи дальше не + теряет транскрипт. Каталог кеша живёт вне workspace задачи и не удаляется + вместе с ним. Любой сбой кеша (чтение, запись) — это промах, а не ошибка задачи. + + При попадании usage не эмитится: оплаченная транскрипция второй раз + в учёт расходов не попадает. + """ + + def __init__(self, inner: Transcriber, *, cache_dir: Path, settings: dict[str, Any]) -> None: + self._inner = inner + self._cache_dir = cache_dir + self._settings = settings + + def _entry(self, audio_hash: str) -> Path: + # Ключ — хеш аудио вместе с канонической записью настроек: смена любой + # настройки даёт другой каталог. + payload = json.dumps( + {"v": CACHE_FORMAT_VERSION, "audio": audio_hash, "settings": self._settings}, + sort_keys=True, + ensure_ascii=False, + ) + return self._cache_dir / hashlib.sha256(payload.encode("utf-8")).hexdigest() + + def _lookup(self, entry: Path) -> str | None: + try: + text = (entry / "transcript.srt").read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return None + if not text.strip(): + return None + with contextlib.suppress(OSError): + os.utime(entry) # свежесть записи для будущей чистки по сроку + return text + + def _store(self, entry: Path, srt_path: Path) -> None: + try: + text = srt_path.read_text(encoding="utf-8") + if not text.strip(): + return + entry.mkdir(parents=True, exist_ok=True) + # Уникальный tmp: две задачи с одним ключом не пишут в один файл. + fd, tmp_name = tempfile.mkstemp(dir=entry, prefix=".transcript.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as tmp_file: + tmp_file.write(text) + os.replace(tmp_name, entry / "transcript.srt") + except BaseException: + with contextlib.suppress(OSError): + os.unlink(tmp_name) + raise + except (OSError, UnicodeDecodeError) as exc: + logger.warning("Кеш транскриптов: запись %s не сохранена: %s", entry.name, exc) + + async def transcribe( + self, + audio_path: Path, + output_dir: Path, + on_progress: ProgressCallback | None = None, + on_usage: UsageCallback | None = None, + ) -> Path: + entry: Path | None = None + try: + entry = self._entry(await asyncio.to_thread(_hash_file, audio_path)) + except OSError as exc: + logger.warning("Кеш транскриптов: не удалось хешировать аудио: %s", exc) + if entry is not None: + cached = await asyncio.to_thread(self._lookup, entry) + if cached is not None: + logger.info("Кеш транскриптов: попадание %s, транскрипция пропущена", entry.name) + output_dir.mkdir(parents=True, exist_ok=True) + target = output_dir / "transcript.srt" + tmp = output_dir / ".transcript.srt.tmp" + tmp.write_text(cached, encoding="utf-8") + tmp.replace(target) + await emit_progress(on_progress, 100) + return target + + srt_path = await self._inner.transcribe( + audio_path=audio_path, + output_dir=output_dir, + on_progress=on_progress, + on_usage=on_usage, + ) + if entry is not None: + await asyncio.to_thread(self._store, entry, srt_path) + return srt_path diff --git a/tests/integration/test_api_contract.py b/tests/integration/test_api_contract.py index 7524d01..d64e99e 100644 --- a/tests/integration/test_api_contract.py +++ b/tests/integration/test_api_contract.py @@ -274,6 +274,20 @@ def test_transcript_failed_on_transcribe_409(client, repo): assert r.json()["error"] == "transcribe_failed" +def test_transcript_failed_without_transcript_410(client, repo): + """Упавшая после транскрипции задача, workspace которой убран: транскрипта нет и не будет.""" + repo.tasks["t"] = Task( + task_id="t", + source_kind="audio", + status=TaskStatus.FAILED, + stage=PipelineStage.STRUCTURIZE, + error="llm down", + ) + r = client.get("/api/v1/tasks/t/transcript") + assert r.status_code == 410 + assert r.json() == {"error": "transcript_unavailable", "detail": "llm down"} + + def test_transcript_in_progress_202(client, repo): repo.tasks["t"] = Task( task_id="t", diff --git a/tests/unit/test_transcript_cache.py b/tests/unit/test_transcript_cache.py new file mode 100644 index 0000000..f0c9e29 --- /dev/null +++ b/tests/unit/test_transcript_cache.py @@ -0,0 +1,337 @@ +import contextlib +import shutil +from pathlib import Path + +import pytest + +from lecturelog.application.factories import transcribe_cache_settings, transcriber_factory +from lecturelog.application.pipeline_service import PipelineService +from lecturelog.application.progress_plan import ProgressPlan +from lecturelog.config.settings import TranscribeConfig +from lecturelog.domain.enums import TaskStatus +from lecturelog.domain.media_source import AudioSource +from lecturelog.domain.models import Section, Task, Topic +from lecturelog.infrastructure.transcribe.cache import CachingTranscriber +from tests.unit.test_pipeline_service import ( + FakeCutter, + FakeExporter, + InMemoryRepo, +) + +SRT = "1\n00:00:00,000 --> 00:00:01,000\nпривет\n" + + +class CountingTranscriber: + """Поддельный транскрибер: считает вызовы и шлёт usage, как настоящий.""" + + def __init__(self): + self.calls = 0 + + async def transcribe(self, audio_path, output_dir, on_progress=None, on_usage=None): + self.calls += 1 + if on_usage: + await on_usage({"audio_seconds": 120, "provider": "groq", "model": "whisper-large-v3"}) + output_dir.mkdir(parents=True, exist_ok=True) + target = output_dir / "transcript.srt" + target.write_text(SRT, encoding="utf-8") + return target + + +class OkStructurizer: + async def structurize( + self, srt_path, slide_images, output_dir, on_progress=None, on_usage=None + ): + sec = Section(title="s", start="0:00", end="1:00", content="c", slide_indices=[]) + return [Topic(title="T", start="0:00", end="1:00", sections=[sec], slide_indices=[])] + + +class BrokenStructurizer: + async def structurize( + self, srt_path, slide_images, output_dir, on_progress=None, on_usage=None + ): + raise RuntimeError("llm down") + + +class BrokenCutter: + async def cut(self, source_path, sections, output_dir): + raise RuntimeError("ffmpeg down") + + +def _audio(tmp_path: Path, payload: bytes = b"audio-bytes") -> Path: + path = tmp_path / "a.mp3" + path.write_bytes(payload) + return path + + +def _cached(inner, tmp_path, **settings): + return CachingTranscriber(inner, cache_dir=tmp_path / "cache", settings=settings or {"m": "a"}) + + +async def _run(tmp_path, transcriber, task_id, *, structurizer=None, cutter=None, audio=None): + repo = InMemoryRepo() + task = Task(task_id=task_id, source_kind="audio") + await repo.create(task) + service = PipelineService( + repository=repo, + transcriber=transcriber, + structurizer=structurizer or OkStructurizer(), + audio_cutter=cutter or FakeCutter(), + exporter=FakeExporter(), + progress_plan_factory=ProgressPlan.for_audio, + ) + with contextlib.suppress(RuntimeError): + await service.run( + task=task, + source=AudioSource(path=audio or _audio(tmp_path)), + slide_provider=None, + work_dir=tmp_path / "work" / task_id, + ) + return await repo.get(task_id) + + +@pytest.mark.asyncio +async def test_rerun_after_structurize_failure_does_not_transcribe_again(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + + failed = await _run(tmp_path, transcriber, "t1", structurizer=BrokenStructurizer()) + assert failed.status == TaskStatus.FAILED + retry = await _run(tmp_path, transcriber, "t2") + + assert retry.status == TaskStatus.DONE + assert inner.calls == 1 + # транскрипт попал в рабочий каталог новой задачи (его читает /transcript) + assert (tmp_path / "work" / "t2" / "transcribe" / "transcript.srt").read_text( + encoding="utf-8" + ) == SRT + + +@pytest.mark.asyncio +async def test_transcript_survives_failure_at_later_stage(tmp_path): + # падение на нарезке (стадия после структуризации) тоже не теряет транскрипт + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + + failed = await _run(tmp_path, transcriber, "t1", cutter=BrokenCutter()) + assert failed.status == TaskStatus.FAILED + await _run(tmp_path, transcriber, "t1") # ретрай той же задачи + + assert inner.calls == 1 + + +@pytest.mark.asyncio +async def test_survives_new_service_instance_and_wiped_work_dir(tmp_path): + # кеш лежит вне workspace задачи: пересоздание сервиса и удаление work_dir его не трогают + inner = CountingTranscriber() + await _run(tmp_path, _cached(inner, tmp_path), "t1", structurizer=BrokenStructurizer()) + shutil.rmtree(tmp_path / "work") + await _run(tmp_path, _cached(inner, tmp_path), "t2") + + assert inner.calls == 1 + + +@pytest.mark.asyncio +async def test_different_audio_content_transcribes_again(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + + await _run(tmp_path, transcriber, "t1") + await _run(tmp_path, transcriber, "t2", audio=_audio(tmp_path, b"other-audio")) + + assert inner.calls == 2 + + +@pytest.mark.asyncio +async def test_same_content_under_other_filename_hits_cache(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + other = tmp_path / "renamed.wav" + other.write_bytes(b"audio-bytes") + + await _run(tmp_path, transcriber, "t1") + await _run(tmp_path, transcriber, "t2", audio=other) + + assert inner.calls == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "changed", + [{"model": "nova-2"}, {"language": "en"}, {"provider": "groq"}, {"detect_language": True}], +) +async def test_changed_transcription_setting_transcribes_again(tmp_path, changed): + base = {"provider": "deepgram", "model": "nova-3", "language": "ru", "detect_language": False} + inner = CountingTranscriber() + + await _run(tmp_path, _cached(inner, tmp_path, **base), "t1") + await _run(tmp_path, _cached(inner, tmp_path, **{**base, **changed}), "t2") + + assert inner.calls == 2 + + +@pytest.mark.asyncio +async def test_cache_hit_does_not_record_transcribe_usage_again(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + + first = await _run(tmp_path, transcriber, "t1") + second = await _run(tmp_path, transcriber, "t2") + + assert first.usage["transcribe"]["audio_seconds"] == 120 + assert "transcribe" not in second.usage + assert second.usage["total"]["audio_seconds"] == 0 + + +@pytest.mark.asyncio +async def test_failed_transcription_is_not_cached(tmp_path): + class Flaky(CountingTranscriber): + async def transcribe(self, *args, **kwargs): + if self.calls == 0: + self.calls += 1 + raise RuntimeError("groq down") + return await super().transcribe(*args, **kwargs) + + inner = Flaky() + transcriber = _cached(inner, tmp_path) + + first = await _run(tmp_path, transcriber, "t1") + second = await _run(tmp_path, transcriber, "t2") + + assert first.status == TaskStatus.FAILED + assert second.status == TaskStatus.DONE + assert inner.calls == 2 + + +@pytest.mark.asyncio +async def test_corrupt_cache_entry_is_treated_as_miss(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + await _run(tmp_path, transcriber, "t1") + for entry in (tmp_path / "cache").rglob("*.srt"): + entry.write_text("", encoding="utf-8") + + await _run(tmp_path, transcriber, "t2") + + assert inner.calls == 2 + + +@pytest.mark.asyncio +async def test_unwritable_cache_does_not_fail_transcription(tmp_path): + blocker = tmp_path / "cache" + blocker.write_text("файл вместо каталога", encoding="utf-8") + inner = CountingTranscriber() + + result = await _run(tmp_path, CachingTranscriber(inner, cache_dir=blocker, settings={}), "t1") + + assert result.status == TaskStatus.DONE + assert inner.calls == 1 + + +def test_cache_settings_cover_deepgram_options(monkeypatch): + monkeypatch.setenv("TRANSCRIBE_PROVIDER", "deepgram") + monkeypatch.setenv("DEEPGRAM_API_KEY", "secret-token") + + def settings(**env): + for name in ("DEEPGRAM_MODEL", "DEEPGRAM_LANGUAGE", "DEEPGRAM_UTT_SPLIT"): + monkeypatch.delenv(name, raising=False) + for name, value in env.items(): + monkeypatch.setenv(name, value) + return transcribe_cache_settings(TranscribeConfig(_env_file=None)) + + base = settings() + assert settings(DEEPGRAM_MODEL="nova-2") != base + assert settings(DEEPGRAM_LANGUAGE="en") != base + assert settings(DEEPGRAM_UTT_SPLIT="1.2") != base + assert "secret-token" not in str(base) # секреты в ключ не попадают + + +def test_cache_settings_differ_between_providers(monkeypatch): + monkeypatch.setenv("DEEPGRAM_API_KEY", "k") + monkeypatch.setenv("GROQ_API_KEYS", "k1") + monkeypatch.setenv("TRANSCRIBE_PROVIDER", "groq") + groq = transcribe_cache_settings(TranscribeConfig(_env_file=None)) + monkeypatch.setenv("TRANSCRIBE_PROVIDER", "deepgram") + deepgram = transcribe_cache_settings(TranscribeConfig(_env_file=None)) + + assert groq != deepgram + + +def test_factory_wraps_transcriber_only_when_cache_dir_given(monkeypatch, tmp_path): + monkeypatch.setenv("TRANSCRIBE_PROVIDER", "groq") + monkeypatch.setenv("GROQ_API_KEYS", "k1") + config = TranscribeConfig(_env_file=None) + + assert not isinstance(transcriber_factory(config), CachingTranscriber) + assert isinstance(transcriber_factory(config, cache_dir=tmp_path), CachingTranscriber) + + +def _entry_dir(tmp_path: Path) -> Path: + (entry,) = [p for p in (tmp_path / "cache").iterdir() if p.is_dir()] + return entry + + +@pytest.mark.asyncio +async def test_concurrent_writer_tmp_name_does_not_collide(tmp_path): + # другая задача с тем же ключом прямо сейчас пишет свой tmp под старым общим именем + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + await _run(tmp_path, transcriber, "t0") + entry = _entry_dir(tmp_path) + shutil.rmtree(entry) + entry.mkdir() + (entry / ".transcript.srt.tmp").mkdir() # занятое общее имя + + await _run(tmp_path, transcriber, "t1") + await _run(tmp_path, transcriber, "t2") + + assert inner.calls == 2 # t0 и t1; t2 уже попал в кеш + + +@pytest.mark.asyncio +async def test_failed_store_leaves_no_tmp_files(tmp_path): + inner = CountingTranscriber() + transcriber = _cached(inner, tmp_path) + await _run(tmp_path, transcriber, "t0") + entry = _entry_dir(tmp_path) + shutil.rmtree(entry) + entry.mkdir() + (entry / "transcript.srt").mkdir() # replace упадёт + + await _run(tmp_path, transcriber, "t1") + + assert [p.name for p in entry.iterdir()] == ["transcript.srt"] + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="нужен ffmpeg") +@pytest.mark.asyncio +async def test_extracted_audio_is_byte_deterministic(tmp_path): + # ключ видео-задач = хеш извлечённой дорожки, поэтому извлечение обязано быть детерминированным + import asyncio + + from lecturelog.infrastructure.media.video_ingestor import VideoIngestor + + video = tmp_path / "v.mp4" + proc = await asyncio.create_subprocess_exec( + *( + [ + "ffmpeg", + "-loglevel", + "error", + "-y", + "-f", + "lavfi", + "-i", + "sine=frequency=440:duration=2", + ] + ), + *(["-f", "lavfi", "-i", "color=c=red:s=64x64:d=2", "-shortest"]), + str(video), + ) + await proc.wait() + ingestor = VideoIngestor(cookie_store=None, target_resolution="720") + + first = await ingestor.extract_audio(video, tmp_path / "x1") + await asyncio.sleep(1.1) # метки времени в контейнере не должны попадать в байты + second = await ingestor.extract_audio(video, tmp_path / "x2") + + assert first.read_bytes() == second.read_bytes() diff --git a/tests/unit/test_transcript_cache_sweep.py b/tests/unit/test_transcript_cache_sweep.py new file mode 100644 index 0000000..ea2ccea --- /dev/null +++ b/tests/unit/test_transcript_cache_sweep.py @@ -0,0 +1,141 @@ +"""Чистка кеша транскриптов sweeper-ом: по mtime записи, с поддельными часами.""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from lecturelog.application.worker import WorkDirSweeper +from lecturelog.config.settings import WorkerConfig + +NOW = datetime(2026, 7, 20, 12, 0, tzinfo=UTC) +RETENTION = timedelta(days=30) +KEY_OLD = "a" * 64 +KEY_FRESH = "b" * 64 + + +class _NoTasks: + async def get(self, task_id): + return None + + +def _entry(cache_dir: Path, key: str, age: timedelta) -> Path: + entry = cache_dir / key + entry.mkdir(parents=True) + (entry / "transcript.srt").write_text("1\n00:00:00,000 --> 00:00:01,000\nтекст\n") + ts = (NOW - age).timestamp() + os.utime(entry, (ts, ts)) + return entry + + +def _sweeper(work_root: Path, cache_dir: Path | None, retention: timedelta | None = RETENTION): + return WorkDirSweeper( + _NoTasks(), + work_root, + timedelta(hours=24), + clock=lambda: NOW, + transcript_cache_dir=cache_dir, + cache_retention=retention, + ) + + +@pytest.mark.asyncio +async def test_запись_старше_срока_удаляется_а_свежая_остаётся(tmp_path): + cache = tmp_path / "transcript_cache" + old = _entry(cache, KEY_OLD, RETENTION + timedelta(seconds=1)) + fresh = _entry(cache, KEY_FRESH, RETENTION - timedelta(seconds=1)) + + result = await _sweeper(tmp_path, cache).sweep() + + assert not old.exists() + assert fresh.exists() + assert result.cache_deleted == 1 + + +@pytest.mark.asyncio +async def test_запись_ровно_на_границе_срока_удаляется(tmp_path): + cache = tmp_path / "transcript_cache" + entry = _entry(cache, KEY_OLD, RETENTION) + + await _sweeper(tmp_path, cache).sweep() + + assert not entry.exists() + + +@pytest.mark.asyncio +async def test_без_срока_кеш_не_чистится(tmp_path): + cache = tmp_path / "transcript_cache" + entry = _entry(cache, KEY_OLD, timedelta(days=3650)) + + await _sweeper(tmp_path, cache, retention=None).sweep() + + assert entry.exists() + + +@pytest.mark.asyncio +async def test_посторонние_объекты_в_каталоге_кеша_не_трогаются(tmp_path): + cache = tmp_path / "transcript_cache" + outside = tmp_path / "outside" + outside.mkdir() + (outside / "keep.txt").write_text("x") + cache.mkdir() + link = cache / KEY_OLD + link.symlink_to(outside, target_is_directory=True) + stray_file = cache / (KEY_FRESH) + stray_file.write_text("x") + not_hash = cache / "not-a-hash" + not_hash.mkdir() + for path in (stray_file, not_hash): + os.utime(path, (0, 0)) + + await _sweeper(tmp_path, cache).sweep() + + assert (outside / "keep.txt").exists() + assert link.is_symlink() + assert stray_file.exists() + assert not_hash.exists() + + +@pytest.mark.asyncio +async def test_кеш_вне_work_dir_тоже_чистится_а_отсутствующий_каталог_не_ошибка(tmp_path): + work = tmp_path / "work" + work.mkdir() + cache = tmp_path / "elsewhere" / "cache" + await _sweeper(work, cache).sweep() # каталога нет - проход не падает + + entry = _entry(cache, KEY_OLD, RETENTION + timedelta(days=1)) + await _sweeper(work, cache).sweep() + + assert not entry.exists() + + +def test_срок_кеша_по_умолчанию_30_дней(monkeypatch): + monkeypatch.delenv("TRANSCRIPT_CACHE_RETENTION_DAYS", raising=False) + assert WorkerConfig().transcript_cache_retention_days == 30 + + +@pytest.mark.parametrize( + "env", + [ + {"TRANSCRIPT_CACHE_RETENTION_DAYS": "0"}, + {"TRANSCRIPT_CACHE_RETENTION_DAYS": "-1"}, + {"TRANSCRIPT_CACHE_RETENTION_DAYS": "366"}, + # Кеш не должен жить меньше workspace упавшей задачи (48 ч > 1 дня). + {"TRANSCRIPT_CACHE_RETENTION_DAYS": "1", "WORK_DIR_FAILED_RETENTION_HOURS": "48"}, + ], +) +def test_неверный_срок_кеша_отклоняется(monkeypatch, env): + for key, value in env.items(): + monkeypatch.setenv(key, value) + with pytest.raises(ValidationError): + WorkerConfig() + + +def test_срок_кеша_равный_сроку_workspace_допустим(monkeypatch): + monkeypatch.setenv("TRANSCRIPT_CACHE_RETENTION_DAYS", "2") + monkeypatch.setenv("WORK_DIR_FAILED_RETENTION_HOURS", "48") + assert WorkerConfig().transcript_cache_retention_days == 2 diff --git a/tests/unit/test_workdir_sweeper_edge_cases.py b/tests/unit/test_workdir_sweeper_edge_cases.py new file mode 100644 index 0000000..ba62e9c --- /dev/null +++ b/tests/unit/test_workdir_sweeper_edge_cases.py @@ -0,0 +1,594 @@ +# ruff: noqa: E501 +"""Краевые случаи TTL-очистки workspace упавших задач (WorkDirSweeper, marker, _sweep_forever).""" + +import asyncio +import json +import os +import shutil +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from lecturelog.api import dependencies as deps +from lecturelog.api.app import create_app +from lecturelog.api.lifespan import _sweep_forever +from lecturelog.application.worker import ( + TERMINAL_MARKER, + PipelineJob, + PipelineWorker, + WorkDirSweeper, + write_terminal_marker, +) +from lecturelog.config.settings import WorkerConfig +from lecturelog.domain.enums import PipelineStage, TaskStatus +from lecturelog.domain.media_source import AudioSource +from lecturelog.domain.models import Task + +NOW = datetime(2026, 7, 20, 12, 0, tzinfo=UTC) +RETENTION = timedelta(hours=24) +TID = "a" * 32 +TID2 = "b" * 32 + + +class Repo: + def __init__(self, statuses=None, default=None, stage=None): + self.statuses = dict(statuses or {}) + self.default = default + self.stage = stage + self.fail_for: set[str] = set() + + async def get(self, task_id): + if task_id in self.fail_for: + raise RuntimeError("db down") + st = self.statuses.get(task_id, self.default) + if st is None: + return None + return Task(task_id=task_id, source_kind="audio", status=st, stage=self.stage, error="boom") + + +def _sweeper(tmp_path, repo, retention=RETENTION, now=NOW): + return WorkDirSweeper(repo, tmp_path, retention, clock=lambda: now) + + +def _ws(root: Path, tid=TID, marker_at=None, mtime_age=None, raw_marker=None): + d = root / tid + d.mkdir() + (d / "raw.bin").write_bytes(b"x") + if marker_at is not None: + (d / TERMINAL_MARKER).write_text(json.dumps({"status": "failed", "at": marker_at})) + if raw_marker is not None: + (d / TERMINAL_MARKER).write_bytes(raw_marker) + if mtime_age is not None: + ts = (NOW - mtime_age).timestamp() + os.utime(d, (ts, ts)) + return d + + +def _iso(delta): + return (NOW - delta).isoformat() + + +# ---------- срок: границы ---------- + + +async def test_marker_ровно_на_границе_срока_удаляется_потому_что_срок_истёк_включительно(tmp_path): + d = _ws(tmp_path, marker_at=_iso(RETENTION)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert not d.exists() + + +async def test_marker_на_секунду_моложе_срока_сохраняется_потому_что_срок_не_истёк(tmp_path): + d = _ws( + tmp_path, marker_at=_iso(RETENTION - timedelta(seconds=1)), mtime_age=timedelta(days=30) + ) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() + + +async def test_marker_приоритетнее_старого_mtime_каталога_потому_что_срок_считается_от_падения( + tmp_path, +): + d = _ws(tmp_path, marker_at=_iso(timedelta(minutes=5)), mtime_age=timedelta(days=30)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() + + +async def test_marker_без_таймзоны_трактуется_как_utc_и_свежий_не_удаляется(tmp_path): + naive = (NOW - timedelta(minutes=5)).replace(tzinfo=None).isoformat() + d = _ws(tmp_path, marker_at=naive, mtime_age=timedelta(days=30)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() + + +async def test_marker_с_нестандартным_смещением_сравнивается_по_абсолютному_времени(tmp_path): + # 5 минут назад, записано в +14:00: по "голым" часам это будущее/прошлое на 14 ч. + at = (NOW - timedelta(minutes=5)).astimezone(timezone_plus14()).isoformat() + d = _ws(tmp_path, marker_at=at, mtime_age=timedelta(days=30)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() + + +def timezone_plus14(): + from datetime import timezone + + return timezone(timedelta(hours=14)) + + +async def test_marker_из_далёкого_будущего_не_делает_workspace_бессмертным_потому_что_диск_утечёт_навсегда( + tmp_path, +): + # Сбитые часы/ручная правка: at=9999 год. Срок "ещё не истёк" вечно, mtime каталога 30 дней. + d = _ws(tmp_path, marker_at="9999-12-31T00:00:00+00:00", mtime_age=timedelta(days=30)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert not d.exists(), "marker из будущего должен считаться недостоверным и уступать mtime" + + +# ---------- битый marker -> откат на mtime, никогда не падение ---------- + +BAD_MARKERS = [ + b"", + b"{", + b"[]", + b"null", + b'"2026-01-01"', + b'{"status": "failed"}', + b'{"at": 12345}', + b'{"at": null}', + b'{"at": ""}', + '{"at": "вчера"}'.encode(), + b"\xff\xfe\x00garbage", +] + + +@pytest.mark.parametrize("raw", BAD_MARKERS) +async def test_битый_marker_не_роняет_проход_и_старый_workspace_удаляется_по_mtime(tmp_path, raw): + d = _ws(tmp_path, raw_marker=raw, mtime_age=timedelta(days=3)) + res = await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert res.deleted == 1 and not d.exists() + + +@pytest.mark.parametrize("raw", BAD_MARKERS) +async def test_битый_marker_у_свежего_workspace_не_приводит_к_удалению(tmp_path, raw): + d = _ws(tmp_path, raw_marker=raw, mtime_age=timedelta(minutes=1)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() + + +async def test_marker_являющийся_каталогом_не_роняет_проход(tmp_path): + d = _ws(tmp_path, mtime_age=timedelta(minutes=1)) + (d / TERMINAL_MARKER).mkdir() + res = await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert d.exists() and res.deleted == 0 + + +# ---------- статус в БД решает, marker — нет ---------- + + +@pytest.mark.parametrize( + "status", [TaskStatus.PENDING, TaskStatus.PROCESSING, TaskStatus.DONE, TaskStatus.INTERRUPTED] +) +async def test_старый_marker_не_удаляет_workspace_задачи_не_в_FAILED_потому_что_решает_статус_в_БД( + tmp_path, status +): + d = _ws(tmp_path, marker_at=_iso(timedelta(days=10)), mtime_age=timedelta(days=10)) + await _sweeper(tmp_path, Repo({TID: status})).sweep() + assert (d / "raw.bin").exists() + + +async def test_каталог_без_строки_в_БД_не_удаляется_потому_что_это_может_быть_создание_задачи( + tmp_path, +): + d = _ws(tmp_path, marker_at=_iso(timedelta(days=10)), mtime_age=timedelta(days=10)) + await _sweeper(tmp_path, Repo()).sweep() + assert d.exists() + + +async def test_marker_со_статусом_failed_не_заменяет_проверку_БД_когда_БД_говорит_processing( + tmp_path, +): + d = _ws(tmp_path, marker_at=_iso(timedelta(days=10))) + await _sweeper(tmp_path, Repo({TID: TaskStatus.PROCESSING})).sweep() + assert d.exists() + + +# ---------- имена, зарезервированные каталоги, симлинки ---------- + + +async def test_зарезервированные_каталоги_и_кеш_не_удаляются_даже_если_БД_ответит_FAILED_на_любой_id( + tmp_path, +): + keep = [] + for name in ( + "results_tmp", + "transcript_cache", + "f" * 64, + "A" * 32, + "a" * 31, + "a" * 33, + "not-task", + ".hidden", + ): + p = tmp_path / name + p.mkdir() + (p / "x").write_text("x") + os.utime(p, (0, 0)) + keep.append(p) + await _sweeper(tmp_path, Repo(default=TaskStatus.FAILED)).sweep() + assert all(p.exists() for p in keep) + + +async def test_symlink_с_именем_задачи_не_ведёт_к_удалению_цели_потому_что_rmtree_не_должен_выходить_из_work_dir( + tmp_path, +): + outside = tmp_path.parent / (tmp_path.name + "_outside") + outside.mkdir() + (outside / "precious").write_text("data") + (tmp_path / TID).symlink_to(outside, target_is_directory=True) + await _sweeper(tmp_path, Repo(default=TaskStatus.FAILED)).sweep() + assert (outside / "precious").exists() + shutil.rmtree(outside) + + +async def test_symlink_внутри_workspace_удаляется_без_удаления_цели_за_пределами_каталога(tmp_path): + outside = tmp_path.parent / (tmp_path.name + "_outside2") + outside.mkdir() + (outside / "precious").write_text("data") + d = _ws(tmp_path, marker_at=_iso(timedelta(days=2))) + (d / "link").symlink_to(outside, target_is_directory=True) + await _sweeper(tmp_path, Repo({TID: TaskStatus.FAILED})).sweep() + assert not d.exists() and (outside / "precious").exists() + shutil.rmtree(outside) + + +async def test_обычный_файл_с_именем_задачи_не_останавливает_проход_по_остальным_каталогам( + tmp_path, +): + (tmp_path / ("0" * 32)).write_text("file") + d = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + await _sweeper(tmp_path, Repo(default=TaskStatus.FAILED)).sweep() + assert not d.exists() + + +async def test_work_dir_сам_симлинк_на_реальный_каталог_не_считается_небезопасным_и_чистится( + tmp_path, +): + real = tmp_path / "real" + real.mkdir() + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + d = _ws(real, marker_at=_iso(timedelta(days=2))) + sw = WorkDirSweeper(Repo({TID: TaskStatus.FAILED}), link, RETENTION, clock=lambda: NOW) + await sw.sweep() + assert not d.exists() + + +# ---------- отказы ---------- + + +async def test_ошибка_БД_на_одной_задаче_не_мешает_чистке_соседней(tmp_path): + a = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + b = _ws(tmp_path, TID2, marker_at=_iso(timedelta(days=2))) + repo = Repo(default=TaskStatus.FAILED) + repo.fail_for = {TID} + await _sweeper(tmp_path, repo).sweep() + assert a.exists() and not b.exists() + + +async def test_ошибка_БД_не_приводит_к_удалению_потому_что_неизвестный_статус_не_значит_FAILED( + tmp_path, +): + a = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + repo = Repo(default=TaskStatus.FAILED) + repo.fail_for = {TID} + await _sweeper(tmp_path, repo).sweep() + assert a.exists() + + +async def test_rmtree_с_PermissionError_не_роняет_проход_и_идёт_дальше(tmp_path, monkeypatch): + a = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + b = _ws(tmp_path, TID2, marker_at=_iso(timedelta(days=2))) + real = shutil.rmtree + + def flaky(p, *a_, **k): + if Path(p).name == TID: + raise PermissionError("ro") + return real(p, *a_, **k) + + monkeypatch.setattr(shutil, "rmtree", flaky) + res = await _sweeper(tmp_path, Repo(default=TaskStatus.FAILED)).sweep() + assert a.exists() and not b.exists() and res.deleted == 1 + + +async def test_отсутствующий_WORK_DIR_даёт_пустой_результат_без_исключения(tmp_path): + res = await _sweeper(tmp_path / "нет", Repo(default=TaskStatus.FAILED)).sweep() + assert res.scanned == 0 and res.deleted == 0 + + +async def test_WORK_DIR_являющийся_файлом_даёт_пустой_результат_без_исключения(tmp_path): + f = tmp_path / "file" + f.write_text("x") + res = await _sweeper(f, Repo(default=TaskStatus.FAILED)).sweep() + assert res.deleted == 0 + + +async def test_пустой_WORK_DIR_ничего_не_делает(tmp_path): + res = await _sweeper(tmp_path, Repo(default=TaskStatus.FAILED)).sweep() + assert (res.scanned, res.deleted, res.skipped) == (0, 0, 0) + + +async def test_нулевой_срок_хранения_удаляет_только_FAILED_а_не_всё(tmp_path): + a = _ws(tmp_path, TID, marker_at=_iso(timedelta(0))) + b = _ws(tmp_path, TID2, marker_at=_iso(timedelta(0))) + await _sweeper( + tmp_path, Repo({TID: TaskStatus.FAILED, TID2: TaskStatus.DONE}), retention=timedelta(0) + ).sweep() + assert not a.exists() and b.exists() + + +# ---------- конкуренция ---------- + + +async def test_параллельный_второй_проход_того_же_sweeper_ничего_не_делает_пока_идёт_первый( + tmp_path, monkeypatch +): + _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + gate = asyncio.Event() + started = asyncio.Event() + repo = Repo(default=TaskStatus.FAILED) + orig = repo.get + + async def slow_get(tid): + started.set() + await gate.wait() + return await orig(tid) + + repo.get = slow_get + sw = _sweeper(tmp_path, repo) + first = asyncio.create_task(sw.sweep()) + await started.wait() + second = await sw.sweep() + gate.set() + r1 = await first + assert second.scanned == 0 and r1.deleted == 1 + + +async def test_два_sweeper_на_одном_каталоге_как_два_процесса_не_падают_и_workspace_удалён( + tmp_path, +): + d = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + repo = Repo(default=TaskStatus.FAILED) + r = await asyncio.gather(_sweeper(tmp_path, repo).sweep(), _sweeper(tmp_path, repo).sweep()) + assert not d.exists() and sum(x.deleted for x in r) >= 1 + + +async def test_отмена_прохода_во_время_ожидания_БД_пробрасывает_CancelledError_а_не_глотает_её( + tmp_path, +): + _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + repo = Repo(default=TaskStatus.FAILED) + + async def hang(tid): + await asyncio.sleep(3600) + + repo.get = hang + t = asyncio.create_task(_sweeper(tmp_path, repo).sweep()) + await asyncio.sleep(0.01) + t.cancel() + with pytest.raises(asyncio.CancelledError): + await t + + +async def test_после_отмены_прохода_lock_освобождён_и_следующий_проход_работает(tmp_path): + d = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + repo = Repo(default=TaskStatus.FAILED) + orig = repo.get + + async def hang(tid): + await asyncio.sleep(3600) + + repo.get = hang + sw = _sweeper(tmp_path, repo) + t = asyncio.create_task(sw.sweep()) + await asyncio.sleep(0.01) + t.cancel() + await asyncio.gather(t, return_exceptions=True) + repo.get = orig + await sw.sweep() + assert not d.exists() + + +# ---------- _sweep_forever ---------- + + +async def test_sweep_forever_переживает_исключение_прохода_и_делает_следующий(): + calls = [] + + class S: + async def sweep(self): + calls.append(1) + if len(calls) == 1: + raise RuntimeError("boom") + + t = asyncio.create_task(_sweep_forever(S(), 0.01)) + await asyncio.sleep(0.1) + t.cancel() + await asyncio.gather(t, return_exceptions=True) + assert len(calls) >= 2 + + +async def test_sweep_forever_отменяется_во_сне_без_зависания(): + class S: + async def sweep(self): + return None + + t = asyncio.create_task(_sweep_forever(S(), 3600)) + await asyncio.sleep(0.01) + t.cancel() + res = await asyncio.wait_for(asyncio.gather(t, return_exceptions=True), 1) + assert isinstance(res[0], asyncio.CancelledError) + + +# ---------- конфиг ---------- + + +@pytest.mark.parametrize( + "env", + [ + {"WORK_DIR_SWEEP_INTERVAL_MINUTES": "0"}, + {"WORK_DIR_SWEEP_INTERVAL_MINUTES": "-5"}, + {"WORK_DIR_FAILED_RETENTION_HOURS": "-1"}, + {"WORK_DIR_FAILED_RETENTION_HOURS": "nan"}, + ], +) +def test_конфиг_отвергает_нулевой_отрицательный_интервал_и_отрицательный_срок_потому_что_иначе_busy_loop_или_мгновенное_удаление( + monkeypatch, env +): + for k, v in env.items(): + monkeypatch.setenv(k, v) + with pytest.raises(Exception): # noqa: B017 + WorkerConfig() + + +def test_конфиг_по_умолчанию_включён_24_часа_60_минут(monkeypatch): + for k in ( + "WORK_DIR_CLEANUP_ENABLED", + "WORK_DIR_FAILED_RETENTION_HOURS", + "WORK_DIR_SWEEP_INTERVAL_MINUTES", + ): + monkeypatch.delenv(k, raising=False) + c = WorkerConfig() + assert (c.cleanup_enabled, c.failed_retention_hours, c.sweep_interval_minutes) == (True, 24, 60) + + +# ---------- marker и воркер ---------- + + +class FailingService: + async def run(self, task, source, slide_provider, work_dir, **kw): + raise RuntimeError("секретный токен sk-123 в тексте ошибки") + + +class OkLocalService: + async def run(self, task, source, slide_provider, work_dir, **kw): + return str(work_dir / "export" / "result.zip") + + +class OkS3Service: + async def run(self, task, source, slide_provider, work_dir, **kw): + return f"results/{task.task_id}/" + + +class _T: + def __init__(self, tid): + self.task_id = tid + + +def _job(work_dir, tid=TID): + return PipelineJob( + task_id=tid, + task=_T(tid), + source=AudioSource(path=Path("/a.mp3")), + slide_provider=None, + work_dir=work_dir, + ) + + +async def _run(worker, *jobs): + await worker.start() + for j in jobs: + await worker.enqueue(j) + await worker.stop() + + +async def test_упавшая_задача_получает_marker_с_utc_временем_и_без_текста_ошибки(tmp_path): + d = tmp_path / TID + d.mkdir() + await _run(PipelineWorker(FailingService(), 1, clock=lambda: NOW), _job(d)) + raw = (d / TERMINAL_MARKER).read_text() + data = json.loads(raw) + assert data["status"] == "failed" and datetime.fromisoformat(data["at"]) == NOW + assert "sk-123" not in raw + assert not (d / (TERMINAL_MARKER + ".tmp")).exists() + + +async def test_успешная_задача_не_получает_marker_потому_что_он_маркирует_только_падение(tmp_path): + d = tmp_path / TID + d.mkdir() + await _run(PipelineWorker(OkLocalService(), 1, clock=lambda: NOW), _job(d)) + assert not (d / TERMINAL_MARKER).exists() + + +async def test_падение_с_исчезнувшим_workspace_не_создаёт_каталог_заново_и_не_роняет_воркер( + tmp_path, +): + d = tmp_path / TID + await _run(PipelineWorker(FailingService(), 1, clock=lambda: NOW), _job(d)) + assert not d.exists() + + +async def test_падение_в_каталоге_с_нештатным_именем_не_пишет_marker(tmp_path): + d = tmp_path / "custom" + d.mkdir() + await _run(PipelineWorker(FailingService(), 1, clock=lambda: NOW), _job(d)) + assert not (d / TERMINAL_MARKER).exists() + + +async def test_не_записанный_marker_не_убивает_consumer_и_следующая_задача_обрабатывается(tmp_path): + bad = tmp_path / TID + bad.mkdir() + (bad / (TERMINAL_MARKER + ".tmp")).mkdir() # write_text упадёт IsADirectoryError + good = tmp_path / TID2 + good.mkdir() + await _run(PipelineWorker(FailingService(), 1, clock=lambda: NOW), _job(bad), _job(good, TID2)) + assert (good / TERMINAL_MARKER).exists() + + +async def test_marker_после_ошибки_cleanup_DONE_задачи_не_делает_её_workspace_удаляемой(tmp_path): + # DONE-задача: даже если marker оказался в каталоге, sweeper смотрит в БД. + d = tmp_path / TID + d.mkdir() + write_terminal_marker(d, TaskStatus.FAILED, lambda: NOW - timedelta(days=9)) + await _sweeper(tmp_path, Repo({TID: TaskStatus.DONE})).sweep() + assert d.exists() + + +def test_write_terminal_marker_в_отсутствующий_каталог_не_бросает(tmp_path): + write_terminal_marker(tmp_path / "нет", TaskStatus.FAILED, lambda: NOW) + + +def test_write_terminal_marker_с_наивными_часами_не_сдвигает_время_относительно_utc(tmp_path): + d = tmp_path / TID + d.mkdir() + write_terminal_marker(d, TaskStatus.FAILED, lambda: NOW.replace(tzinfo=None)) + at = datetime.fromisoformat(json.loads((d / TERMINAL_MARKER).read_text())["at"]) + assert at.tzinfo is not None + + +# ---------- стык со слоем чтения workspace (API) ---------- + + +def _client(tmp_path, repo): + app = create_app() + app.dependency_overrides[deps.get_repository] = lambda: repo + app.dependency_overrides[deps.get_work_dir] = lambda: tmp_path + app.dependency_overrides[deps.get_storage] = lambda: None + return TestClient(app) + + +async def test_транскрипт_FAILED_задачи_после_чистки_не_маскируется_под_идущую_обработку_202_in_progress( + tmp_path, +): + # Задача упала после транскрибации (STRUCTURIZE), SRT был доступен; после TTL-чистки + # роут отвечает 202 "in_progress, retry later" вечно: клиент поллит мёртвую задачу. + d = _ws(tmp_path, TID, marker_at=_iso(timedelta(days=2))) + (d / "transcribe").mkdir() + (d / "transcribe" / "transcript.srt").write_text("1\n00:00:00,000 --> 00:00:01,000\nпривет\n") + repo = Repo({TID: TaskStatus.FAILED}, stage=PipelineStage.STRUCTURIZE) + client = _client(tmp_path, repo) + assert client.get(f"/api/v1/tasks/{TID}/transcript").status_code == 200 # до чистки + await _sweeper(tmp_path, repo).sweep() + assert not d.exists() + resp = client.get(f"/api/v1/tasks/{TID}/transcript") + assert not (resp.status_code == 202 and resp.json().get("status") == "in_progress") diff --git a/tests/unit/test_worker.py b/tests/unit/test_worker.py index 0ed6b1b..ef9715f 100644 --- a/tests/unit/test_worker.py +++ b/tests/unit/test_worker.py @@ -153,3 +153,109 @@ async def test_worker_refuses_to_remove_workspace_with_unexpected_name(tmp_path) await worker.stop() assert (work_dir / "raw.bin").exists() + + +# --- TTL-очистка workspace упавших задач --------------------------------------------- + +from datetime import UTC, datetime, timedelta # noqa: E402 + +from lecturelog.application.worker import WorkDirSweeper # noqa: E402 +from lecturelog.domain.enums import TaskStatus # noqa: E402 + + +class FakeClock: + def __init__(self): + self.now = datetime(2026, 7, 1, 12, 0, tzinfo=UTC) + + def __call__(self): + return self.now + + def advance(self, **kw): + self.now += timedelta(**kw) + + +class _Repo: + def __init__(self, statuses): + self._s = statuses + + async def get(self, task_id): + st = self._s.get(task_id) + if st is None: + return None + t = _Task(task_id) + t.status = st + return t + + +A, B, C, D = ("a" * 32, "b" * 32, "c" * 32, "d" * 32) + + +async def _fail_job(tmp_path, tid, clock): + job = _completed_job(tmp_path, tid) + worker = PipelineWorker(service=FailingService(), concurrency=1, clock=clock) + await worker.start() + await worker.enqueue(job) + await worker.stop() + + +@pytest.mark.asyncio +async def test_failed_workspace_deleted_only_after_ttl(tmp_path): + clock = FakeClock() + await _fail_job(tmp_path, A, clock) + repo = _Repo({A: TaskStatus.FAILED}) + sweeper = WorkDirSweeper(repo, tmp_path, timedelta(hours=24), clock) + + clock.advance(hours=23, minutes=59) + assert (await sweeper.sweep()).deleted == 0 + assert (tmp_path / A).exists() + + clock.advance(minutes=2) + assert (await sweeper.sweep()).deleted == 1 + assert not (tmp_path / A).exists() + + +@pytest.mark.asyncio +async def test_sweeper_keeps_active_done_unknown_and_reserved_dirs(tmp_path): + clock = FakeClock() + repo = _Repo({A: TaskStatus.PROCESSING, B: TaskStatus.PENDING, C: TaskStatus.DONE}) + names = (A, B, C, D, "results_tmp", "transcript_cache", "not-a-task") + for name in names: + (tmp_path / name).mkdir() + (tmp_path / name / "x").write_bytes(b"x") + sweeper = WorkDirSweeper(repo, tmp_path, timedelta(hours=24), clock) + + clock.advance(days=365) + assert (await sweeper.sweep()).deleted == 0 + for name in names: + assert (tmp_path / name).exists() + + +@pytest.mark.asyncio +async def test_transcript_cache_outlives_failed_workspace(tmp_path): + # Кеш транскриптов - сосед workspace: удаление упавшей задачи его не задевает. + clock = FakeClock() + await _fail_job(tmp_path, A, clock) + entry = tmp_path / "transcript_cache" / ("f" * 64) + entry.mkdir(parents=True) + (entry / "transcript.srt").write_text("1\n") + sweeper = WorkDirSweeper(_Repo({A: TaskStatus.FAILED}), tmp_path, timedelta(hours=24), clock) + + clock.advance(hours=25) + assert (await sweeper.sweep()).deleted == 1 + assert (entry / "transcript.srt").exists() + + +@pytest.mark.asyncio +async def test_sweeper_skips_symlink_and_continues(tmp_path): + clock = FakeClock() + await _fail_job(tmp_path, B, clock) + outside = tmp_path / "outside" + outside.mkdir() + (tmp_path / A).symlink_to(outside, target_is_directory=True) + repo = _Repo({A: TaskStatus.FAILED, B: TaskStatus.FAILED}) + sweeper = WorkDirSweeper(repo, tmp_path, timedelta(hours=1), clock) + + clock.advance(hours=2) + assert (await sweeper.sweep()).deleted == 1 + assert outside.exists() and (tmp_path / A).is_symlink() + assert not (tmp_path / B).exists()