scripts/mobile_api.py - master - Max_Cherep/super-transcription

scripts/mobile_api.py
293 строки9 KB
maxim.cherepanov
feat(extension): E6 shots over WS, E7 Autostart, ST_ filenames
09 авг 2026, 14:20
09 авг 2026, 14:20e5f250b
"""JSON/multipart API для мобильного клиента (Android)."""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

from db import create_job, get_job, has_active_job, has_active_job_for_input, list_jobs
from download_utils import DownloadError, job_download_artifacts, prepare_download
from ffmpeg_ops import INPUT_EXTENSIONS, VIDEO_EXTENSIONS
from input_integrity import (
    delete_input_integrity,
    ensure_input_integrity,
    is_input_corrupted,
    schedule_input_check,
)
from job_cancel import request_cancel
from messaging import publish_job
from process_options import (
    DEFAULT_OPTIONS,
    has_transformation,
    job_options_summary,
    parse_options,
)
from screenshots import discover_screenshot_entries, normalize_screenshot_entries

INPUT_DIR = Path(os.environ.get("INPUT_DIR", "/app/input"))
OUTPUT_DIR = Path(os.environ.get("OUTPUT_DIR", "/app/output"))
MEDIA_EXTENSIONS = INPUT_EXTENSIONS

# Пресеты = дефолты пайплайна + типичные языки
MOBILE_PRESETS: dict[str, dict[str, Any]] = {
    "ru_txt": {**DEFAULT_OPTIONS},
    "ru_srt": {**DEFAULT_OPTIONS, "output_format": "srt"},
    "en_txt": {**DEFAULT_OPTIONS, "language": "en"},
    "en_srt": {**DEFAULT_OPTIONS, "language": "en", "output_format": "srt"},
    "auto_txt": {**DEFAULT_OPTIONS, "language": "auto"},
}


def list_presets() -> list[dict[str, Any]]:
    return [
        {"id": key, "options": dict(opts)}
        for key, opts in MOBILE_PRESETS.items()
    ]


def default_options() -> dict[str, Any]:
    return dict(DEFAULT_OPTIONS)


def _output_path_for(input_path: Path, output_format: str) -> Path:
    stem = input_path.stem
    ext = output_format if output_format.startswith(".") else f".{output_format}"
    return OUTPUT_DIR / f"{stem}{ext}"


def _enqueue_process(input_path: Path, options: dict[str, Any]) -> int | None:
    input_path = input_path.resolve()
    opts = parse_options(options)
    shots = normalize_screenshot_entries(
        opts.get("screenshots"),
        for_filename=input_path.name,
    )
    if not shots:
        shots = discover_screenshot_entries(INPUT_DIR, input_path.name)
    opts["screenshots"] = shots
    out_fmt = opts["output_format"]
    output_path = _output_path_for(input_path, out_fmt)
    str_in = str(input_path)
    str_out = str(output_path)

    if not has_transformation(opts, input_path.suffix):
        return None

    if has_active_job(str_in, str_out):
        return None

    if is_input_corrupted(input_path):
        return None

    job_id = create_job(
        input_path.name,
        str_in,
        str_out,
        options=opts,
        output_format=out_fmt,
        job_type="transcribe",
    )
    publish_job(job_id)
    return job_id


def resolve_process_options(
    *,
    preset: str | None,
    options: dict[str, Any] | None,
    asr_available: bool = True,
) -> dict[str, Any]:
    _ = asr_available
    if preset:
        if preset not in MOBILE_PRESETS:
            raise ValueError(f"unknown preset: {preset}")
        raw = dict(MOBILE_PRESETS[preset])
        if options:
            raw.update(options)
    elif options:
        raw = {**DEFAULT_OPTIONS, **dict(options)}
    else:
        raise ValueError("preset or options required")
    return parse_options(raw)


def list_input_files() -> list[dict[str, Any]]:
    if not INPUT_DIR.exists():
        return []
    files: list[dict[str, Any]] = []
    for path in sorted(
        p for p in INPUT_DIR.iterdir()
        if p.is_file() and p.suffix.lower() in MEDIA_EXTENSIONS
    ):
        resolved = str(path.resolve())
        stat = path.stat()
        integrity = ensure_input_integrity(path)
        files.append({
            "name": path.name,
            "size_bytes": stat.st_size,
            "busy": has_active_job_for_input(resolved),
            "corrupted": integrity["corrupted"],
            "pending": bool(integrity.get("pending")),
            "error_message": integrity["error_message"],
            "is_video": path.suffix.lower() in VIDEO_EXTENSIONS,
        })
    return files


async def upload_files(uploads: list[tuple[str, bytes]]) -> dict[str, Any]:
    INPUT_DIR.mkdir(parents=True, exist_ok=True)
    saved_media = 0
    errors: list[str] = []
    saved_names: list[str] = []
    corrupted_names: list[str] = []

    for name, content in uploads:
        safe_name = Path(name).name
        suffix = Path(safe_name).suffix.lower()
        dest = INPUT_DIR / safe_name

        if suffix not in MEDIA_EXTENSIONS:
            errors.append(f"{safe_name}: неподдерживаемый формат")
            continue

        dest.write_bytes(content)
        saved_media += 1
        saved_names.append(safe_name)
        # Полный ffmpeg-decode в фоне: иначе upload/UI зависают на крупных video.
        schedule_input_check(dest)

    return {
        "saved_audio": saved_media,
        "saved_media": saved_media,
        "saved_names": saved_names,
        "corrupted_names": corrupted_names,
        "errors": errors,
    }


def enqueue_process_filenames(
    filenames: list[str],
    options: dict[str, Any],
) -> dict[str, Any]:
    if not filenames:
        raise ValueError("filenames required")

    queued: list[int] = []
    skipped_missing: list[str] = []
    skipped_noop: list[str] = []
    skipped_busy: list[str] = []
    skipped_corrupted: list[str] = []

    for raw in filenames:
        name = Path(raw).name
        path = INPUT_DIR / name
        if not path.is_file():
            skipped_missing.append(name)
            continue
        if is_input_corrupted(path):
            skipped_corrupted.append(name)
            continue
        if not has_transformation(options, path.suffix):
            skipped_noop.append(name)
            continue
        job_id = _enqueue_process(path, options)
        if job_id is None:
            skipped_busy.append(name)
        else:
            queued.append(job_id)

    return {
        "queued": len(queued),
        "job_ids": queued,
        "skipped_missing": skipped_missing,
        "skipped_noop": skipped_noop,
        "skipped_busy": skipped_busy,
        "skipped_corrupted": skipped_corrupted,
    }


def delete_input_file(filename: str) -> dict[str, Any]:
    name = Path(filename).name
    path = (INPUT_DIR / name).resolve()
    if not path.is_file() or path.suffix.lower() not in MEDIA_EXTENSIONS:
        raise ValueError("file not found")
    if path.parent.resolve() != INPUT_DIR.resolve():
        raise ValueError("invalid path")
    if has_active_job_for_input(str(path)):
        raise ValueError("file busy")
    path.unlink()
    delete_input_integrity(name)
    return {"deleted": name}


def list_ready_jobs() -> list[dict[str, Any]]:
    ready: list[dict[str, Any]] = []
    for job in list_jobs():
        if job.get("status") != "done":
            continue
        try:
            _entries, _df, _dd, download_filename = job_download_artifacts(
                job, input_dir=INPUT_DIR, output_dir=OUTPUT_DIR,
            )
        except DownloadError:
            continue
        ready.append({
            "id": job["id"],
            "filename": job["filename"],
            "output_format": job.get("output_format"),
            "download_filename": download_filename,
            "finished_at": job.get("finished_at"),
            "options_summary": job_options_summary(job),
        })
    return ready


def _job_public(job: dict[str, Any]) -> dict[str, Any]:
    status = job.get("status") or "queued"
    return {
        "id": job["id"],
        "filename": job["filename"],
        "status": status,
        "job_type": job.get("job_type") or "transcribe",
        "output_format": job.get("output_format"),
        "options_summary": job_options_summary(job),
        "created_at": job.get("created_at"),
        "started_at": job.get("started_at"),
        "finished_at": job.get("finished_at"),
        "duration_sec": job.get("duration_sec"),
        "progress_pct": job.get("progress_pct"),
        "progress_detail": job.get("progress_detail"),
        "error_message": job.get("error_message"),
        "cancel_requested": bool(job.get("cancel_requested")),
        "can_cancel": status in ("queued", "processing"),
    }


def list_all_jobs(limit: int = 100) -> list[dict[str, Any]]:
    limit = max(1, min(int(limit), 200))
    return [_job_public(job) for job in list_jobs(limit=limit)]


def cancel_job(job_id: int) -> dict[str, Any]:
    ok, message = request_cancel(job_id)
    if not ok:
        if message == "задача не найдена":
            raise ValueError("job not found")
        if message == "задачу нельзя прервать":
            raise ValueError("job not cancellable")
        raise ValueError(message)
    job = get_job(job_id)
    return {
        "message": message,
        "job": _job_public(job) if job else None,
    }


def get_job_download(job_id: int):
    job = get_job(job_id)
    if job is None:
        raise DownloadError("job not found")
    if job.get("status") != "done":
        raise DownloadError("job not ready")
    return job, prepare_download(job, input_dir=INPUT_DIR, output_dir=OUTPUT_DIR)