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

scripts/process_options.py
349 строк12 KB
maxim.cherepanov
feat: extension ping, email/Kaiten options, drop Redis from multiuser plan
07 авг 2026, 12:46
07 авг 2026, 12:4670ec00d
"""Парсинг и валидация опций транскрибации."""

from __future__ import annotations

import json
from typing import Any

from llm_prompts_loader import selected_ids_from_opts

OUTPUT_FORMATS = frozenset({"txt", "srt"})
LLM_OUTPUT_FORMATS = frozenset({"txt", "odt", "pdf"})
LANGUAGES = frozenset({"auto", "ru", "en", "de", "fr", "es", "it", "pt", "pl"})
WHISPER_MODELS = frozenset({
    "tiny", "base", "small", "medium", "large-v2", "large-v3",
})
SPEECH_MODES = frozenset({"monologue", "dialogue", "meeting"})

DEFAULT_OPTIONS: dict[str, Any] = {
    "language": "ru",
    "model": "large-v3",
    "output_format": "txt",
    "speech_mode": "monologue",
    "exam": False,  # soft Q/A prior при dialogue (взаимоисключ. с interview)
    "interview": False,  # soft Q/A prior при dialogue (взаимоисключ. с exam)
    "dialogue_repair": False,  # опц. LLM-правка меток; default off
    "timestamps": True,
    "vad_filter": True,
    "beam_size": 5,
    "llm_prompt_ids": [],
    # per-prompt primary format: {prompt_id: txt|odt|pdf}; default txt
    "llm_formats": {},
    # legacy global (demoted); fallback если id нет в llm_formats
    "llm_output_format": "txt",
    "screenshots": [],  # [{t_sec, path, filename?}] — кадры для OCR
    # Extension / integrations (email — свойство user; MVP: SMTP + MVP_USER_EMAIL в .env)
    "email_result": False,
    # ID карточки Kaiten (опц.); контекст тянется при старте job, если задан
    "kaiten_card_id": None,
}

BEAM_SIZE_MIN = 1
BEAM_SIZE_MAX = 10


def _normalize_kaiten_card_id(val: Any) -> str | None:
    """ID карточки Kaiten — ровно 8 цифр (строка, чтобы сохранить ведущие нули)."""
    if val is None or val == "":
        return None
    digits = "".join(ch for ch in str(val).strip() if ch.isdigit())
    if len(digits) != 8:
        return None
    return digits

def _clamp_int(val: Any, lo: int, hi: int, default: int) -> int:
    try:
        n = int(val)
    except (TypeError, ValueError):
        return default
    return max(lo, min(hi, n))


def _as_bool(val: Any, default: bool = False) -> bool:
    if val is None:
        return default
    if isinstance(val, bool):
        return val
    if isinstance(val, (int, float)):
        return bool(val)
    s = str(val).strip().lower()
    if s in ("", "none"):
        return default
    return s in ("1", "true", "yes", "on")


def _normalize_fmt(val: Any, *, field: str) -> str:
    fmt = str(val if val is not None else "txt").lower().strip()
    if fmt not in LLM_OUTPUT_FORMATS:
        raise ValueError(f"invalid {field}: {val}")
    return fmt


def normalize_llm_formats(
    raw: Any,
    *,
    prompt_ids: list[str] | None = None,
    legacy_global: str | None = None,
) -> dict[str, str]:
    """{prompt_id: txt|odt|pdf}. Пустой raw + legacy ≠ txt → legacy на все ids."""
    out: dict[str, str] = {}
    if isinstance(raw, dict):
        for k, v in raw.items():
            out[str(k)] = _normalize_fmt(v, field=f"llm_formats[{k}]")
    legacy = _normalize_fmt(
        legacy_global if legacy_global is not None else "txt",
        field="llm_output_format",
    )
    ids = [str(x) for x in (prompt_ids or [])]
    if not out and ids and legacy != "txt":
        return {pid: legacy for pid in ids}
    for pid in ids:
        out.setdefault(pid, "txt")
    return out


def format_for_prompt(opts: dict[str, Any], prompt_id: str) -> str:
    """Формат финала для prompt id (llm_formats → legacy llm_output_format → txt)."""
    fmts = opts.get("llm_formats")
    if isinstance(fmts, dict):
        v = fmts.get(prompt_id)
        if v is None:
            v = fmts.get(str(prompt_id))
        if v is not None:
            fmt = str(v).lower().strip()
            if fmt in LLM_OUTPUT_FORMATS:
                return fmt
    legacy = str(opts.get("llm_output_format") or "txt").lower().strip()
    return legacy if legacy in LLM_OUTPUT_FORMATS else "txt"


def _llm_formats_from_form(form: dict[str, Any], ids: list[str]) -> dict[str, str]:
    collected: dict[str, str] = {}
    raw = form.get("llm_formats")
    if isinstance(raw, dict):
        for k, v in raw.items():
            collected[str(k)] = str(v).lower().strip()
    for k, v in form.items():
        if isinstance(k, str) and k.startswith("llm_format_"):
            collected[k[len("llm_format_") :]] = str(v).lower().strip()
    has_per = bool(collected)
    legacy = str(form.get("llm_output_format") or "txt").lower().strip()
    if legacy not in LLM_OUTPUT_FORMATS:
        legacy = "txt"
    out: dict[str, str] = {}
    for pid in ids:
        if has_per:
            fmt = collected.get(pid, "txt")
        else:
            fmt = legacy
        out[pid] = fmt if fmt in LLM_OUTPUT_FORMATS else "txt"
    return out


def parse_options(raw: str | dict[str, Any] | None) -> dict[str, Any]:
    if raw is None or raw == "":
        data: dict[str, Any] = {}
    elif isinstance(raw, str):
        data = json.loads(raw)
    else:
        data = dict(raw)

    language = str(data.get("language", DEFAULT_OPTIONS["language"])).lower()
    if language not in LANGUAGES:
        raise ValueError(f"invalid language: {language}")

    model = str(data.get("model", DEFAULT_OPTIONS["model"])).lower()
    if model not in WHISPER_MODELS:
        raise ValueError(f"invalid model: {model}")

    out_fmt = str(data.get("output_format", DEFAULT_OPTIONS["output_format"])).lower()
    if out_fmt not in OUTPUT_FORMATS:
        raise ValueError(f"invalid output_format: {out_fmt}")

    llm_out = _normalize_fmt(
        data.get("llm_output_format", DEFAULT_OPTIONS["llm_output_format"]),
        field="llm_output_format",
    )

    speech_raw = data.get("speech_mode", DEFAULT_OPTIONS["speech_mode"])
    speech_mode = str(speech_raw or DEFAULT_OPTIONS["speech_mode"]).lower().strip()
    if speech_mode not in SPEECH_MODES:
        speech_mode = str(DEFAULT_OPTIONS["speech_mode"])

    # legacy flags + llm_prompt_ids
    merged = dict(data)
    ids = selected_ids_from_opts(merged)
    llm_formats = normalize_llm_formats(
        data.get("llm_formats"),
        prompt_ids=ids,
        legacy_global=llm_out,
    )

    exam = _as_bool(data.get("exam"), DEFAULT_OPTIONS["exam"])
    interview = _as_bool(data.get("interview"), DEFAULT_OPTIONS["interview"])
    dialogue_repair = _as_bool(
        data.get("dialogue_repair", data.get("llm_repair")),
        DEFAULT_OPTIONS["dialogue_repair"],
    )
    # legacy interview_exam / qa_roles → interview (если новых флагов нет)
    legacy_qa = _as_bool(
        data.get("interview_exam", data.get("qa_roles")),
        False,
    )
    if legacy_qa and not exam and not interview:
        interview = True
    if speech_mode != "dialogue":
        exam = False
        interview = False
        dialogue_repair = False
    elif exam and interview:
        raise ValueError("exam and interview are mutually exclusive")

    from screenshots import normalize_screenshot_entries

    out: dict[str, Any] = {
        "language": language,
        "model": model,
        "output_format": out_fmt,
        "speech_mode": speech_mode,
        "exam": exam,
        "interview": interview,
        "dialogue_repair": dialogue_repair,
        "timestamps": _as_bool(data.get("timestamps"), DEFAULT_OPTIONS["timestamps"]),
        "vad_filter": _as_bool(data.get("vad_filter"), DEFAULT_OPTIONS["vad_filter"]),
        "beam_size": _clamp_int(
            data.get("beam_size"),
            BEAM_SIZE_MIN,
            BEAM_SIZE_MAX,
            int(DEFAULT_OPTIONS["beam_size"]),
        ),
        "llm_prompt_ids": ids,
        "llm_formats": llm_formats,
        "llm_output_format": llm_out,
        "screenshots": normalize_screenshot_entries(data.get("screenshots")),
        "email_result": _as_bool(
            data.get("email_result"),
            DEFAULT_OPTIONS["email_result"],
        ),
        "kaiten_card_id": _normalize_kaiten_card_id(data.get("kaiten_card_id")),
    }
    # снимок промптов (если уже есть в job) — прокидываем как есть
    snap = data.get("llm_prompts")
    if isinstance(snap, list) and snap:
        out["llm_prompts"] = snap
    if data.get("parent_job_id") is not None:
        out["parent_job_id"] = data.get("parent_job_id")
    return out


def has_transformation(opts: dict[str, Any], input_suffix: str) -> bool:
    _ = opts, input_suffix
    return True


def options_summary(opts: dict[str, Any]) -> str:
    parts = [
        opts.get("language", "ru"),
        opts.get("model", "large-v3"),
        opts.get("output_format", "txt"),
        opts.get("speech_mode", "monologue"),
    ]
    if opts.get("exam"):
        parts.append("exam")
    if opts.get("interview"):
        parts.append("interview")
    if opts.get("dialogue_repair"):
        parts.append("repair")
    if opts.get("timestamps", True):
        parts.append("timestamps")
    if opts.get("vad_filter", True):
        parts.append("vad")
    beam = opts.get("beam_size", 5)
    parts.append(f"beam={beam}")
    shots = opts.get("screenshots") or []
    if isinstance(shots, list) and shots:
        parts.append(f"shots={len(shots)}")
    if opts.get("email_result"):
        parts.append("email")
    kid = opts.get("kaiten_card_id")
    if kid:
        parts.append(f"kaiten={kid}")
    ids = selected_ids_from_opts(opts)
    snap = opts.get("llm_prompts")
    if isinstance(snap, list) and snap:
        aliases = [str(s.get("alias") or s.get("id")) for s in snap if isinstance(s, dict)]
        if aliases:
            parts.append("llm=" + "+".join(aliases))
    elif ids:
        parts.append("llm=" + "+".join(ids))
    fmts = opts.get("llm_formats") if isinstance(opts.get("llm_formats"), dict) else {}
    non_txt = [
        f"{pid}:{fmt}"
        for pid, fmt in sorted(fmts.items())
        if str(fmt).lower() in ("odt", "pdf")
    ]
    if non_txt:
        parts.append("llm_fmt=" + "+".join(non_txt))
    else:
        llm_fmt = str(opts.get("llm_output_format") or "txt").lower()
        if llm_fmt in ("odt", "pdf"):
            parts.append(f"llm_fmt={llm_fmt}")
    return ", ".join(parts)


def job_options_summary(job: dict[str, Any]) -> str:
    raw = job.get("options")
    if not raw:
        return "—"
    try:
        return options_summary(parse_options(raw))
    except (json.JSONDecodeError, ValueError):
        return "—"


def wants_llm(opts: dict[str, Any]) -> bool:
    if isinstance(opts.get("llm_prompts"), list) and opts["llm_prompts"]:
        return True
    return bool(selected_ids_from_opts(opts))


def options_from_form(form: dict[str, Any]) -> dict[str, Any]:
    raw_ids = form.get("llm_prompts")
    if isinstance(raw_ids, list):
        ids = [str(x).strip() for x in raw_ids if str(x).strip()]
    elif isinstance(raw_ids, str) and raw_ids.strip():
        ids = [raw_ids.strip()]
    else:
        ids = []
    # legacy single fields
    for legacy, pid in (
        ("llm_cleanup", "cleanup"),
        ("llm_summary", "summary"),
        ("llm_actions", "actions"),
    ):
        if form.get(legacy) == "on" and pid not in ids:
            ids.append(pid)

    from screenshots import parse_screenshots_from_form

    llm_formats = _llm_formats_from_form(form, ids)

    return {
        "language": form.get("language") or DEFAULT_OPTIONS["language"],
        "model": form.get("model") or DEFAULT_OPTIONS["model"],
        "output_format": form.get("output_format") or DEFAULT_OPTIONS["output_format"],
        "speech_mode": form.get("speech_mode") or DEFAULT_OPTIONS["speech_mode"],
        "exam": form.get("exam") == "on",
        "interview": form.get("interview") == "on",
        "dialogue_repair": form.get("dialogue_repair") == "on",
        "timestamps": form.get("timestamps") == "on",
        "vad_filter": form.get("vad_filter") == "on",
        "beam_size": form.get("beam_size") or str(DEFAULT_OPTIONS["beam_size"]),
        "llm_prompt_ids": ids,
        "llm_formats": llm_formats,
        "llm_output_format": form.get("llm_output_format")
        or DEFAULT_OPTIONS["llm_output_format"],
        "screenshots": parse_screenshots_from_form(form.get("screenshots")),
        "email_result": form.get("email_result") == "on",
        "kaiten_card_id": form.get("kaiten_card_id") or None,
    }