app.py

# app.py
import gradio as gr, os, json, pandas as pd, math
from analyzer import analyze

INTRO = """
# Voice Analyzer (Lite)
Загрузите аудио (wav/mp3/m4a), нажмите **Анализ** и скачайте **JSON**.
Файл автоматически проверяется на соответствие схеме `analysis_schema.json`.
Если что-то не так — появится список ошибок с указанием поля.
"""

# ===== ЛЁГКАЯ ВАЛИДАЦИЯ (БЕЗ ЗАВИСИМОСТЕЙ) =====
def _is_num(x):
    return isinstance(x, (int, float)) and not math.isnan(x)

def _is_opt_num(x):
    return (x is None) or _is_num(x)

def validate_payload(payload: dict) -> list:
    """Возвращает список ошибок (пустой = всё ок)."""
    errors = []

    # 1) Корень
    for key in ["version", "aggregates", "words", "transcript"]:
        if key not in payload:
            errors.append(f"Отсутствует корневое поле `{key}`.")
    if errors:
        return errors

    # версия формата: допускаем 1.0, 1.1, 1.2
    allowed_versions = {"1.0", "1.1", "1.2"}
    ver = str(payload.get("version", ""))
    if ver not in allowed_versions:
        errors.append(f"`version` должен быть одним из {sorted(allowed_versions)}, сейчас: {ver}.")

    # 2) Aggregates
    agg = payload["aggregates"]
    req_agg = [
        "duration_sec","voiced_duration_sec","voiced_ratio","rms_dbfs_mean",
        "f0_mean_hz","f0_median_hz","f0_std_hz","f0_stability",
        "words_count","speech_rate_wps","vad"
    ]
    for k in req_agg:
        if k not in agg:
            errors.append(f"aggregates: отсутствует `{k}`.")

    if "duration_sec" in agg and not _is_num(agg["duration_sec"]):
        errors.append("aggregates.duration_sec должен быть числом ≥ 0.")
    if "voiced_duration_sec" in agg and not _is_num(agg["voiced_duration_sec"]):
        errors.append("aggregates.voiced_duration_sec должен быть числом ≥ 0.")
    if "voiced_ratio" in agg:
        vr = agg["voiced_ratio"]
        if not _is_num(vr) or vr < 0 or vr > 1:
            errors.append("aggregates.voiced_ratio должен быть числом от 0 до 1.")

    for k in ["rms_dbfs_mean","f0_mean_hz","f0_median_hz","f0_std_hz","f0_stability"]:
        if k in agg and not _is_opt_num(agg[k]):
            errors.append(f"aggregates.{k} должен быть числом или null.")
    if "f0_stability" in agg and agg["f0_stability"] is not None:
        fs = agg["f0_stability"]
        if fs < 0 or fs > 1:
            errors.append("aggregates.f0_stability должен быть в диапазоне 0..1.")
    if "words_count" in agg and not isinstance(agg["words_count"], int):
        errors.append("aggregates.words_count должен быть целым числом.")
    if "speech_rate_wps" in agg and not _is_num(agg["speech_rate_wps"]):
        errors.append("aggregates.speech_rate_wps должен быть числом.")
    if "vad" in agg and not isinstance(agg["vad"], str):
        errors.append("aggregates.vad должен быть строкой.")

    # 3) Words
    words = payload["words"]
    if not isinstance(words, list):
        errors.append("`words` должен быть массивом объектов.")
        return errors

    req_word = [
        "start_s","end_s","duration_s","pre_pause_ms","word","prob",
        "f0_mean_hz","f0_z","rms_dbfs","rms_z",
        "accent_score","accent_flag","speed_local_sylps","duration_z"
    ]

    for i, w in enumerate(words[:200]):  # проверяем первые 200 для скорости
        if not isinstance(w, dict):
            errors.append(f"words[{i}] должен быть объектом.")
            continue
        for k in req_word:
            if k not in w:
                errors.append(f"words[{i}]: отсутствует `{k}`.")

        def num_or_null(val, name):
            if val is not None and not _is_num(val):
                errors.append(f"words[{i}].{name} должен быть числом или null.")

        def num_required(val, name):
            if not _is_num(val):
                errors.append(f"words[{i}].{name} должен быть числом.")

        if "start_s" in w: num_required(w["start_s"], "start_s")
        if "end_s" in w:   num_required(w["end_s"], "end_s")
        if "duration_s" in w: num_required(w["duration_s"], "duration_s")
        if "pre_pause_ms" in w and not isinstance(w["pre_pause_ms"], int):
            errors.append(f"words[{i}].pre_pause_ms должен быть целым числом.")
        if "word" in w and not isinstance(w["word"], str):
            errors.append(f"words[{i}].word должен быть строкой.")

        for k in ["prob","f0_mean_hz","f0_z","rms_dbfs","rms_z","speed_local_sylps","duration_z"]:
            if k in w: num_or_null(w[k], k)

        if "accent_score" in w:
            v = w["accent_score"]
            if not _is_num(v) or v < 0 or v > 1:
                errors.append(f"words[{i}].accent_score должен быть числом 0..1.")
        if "accent_flag" in w and w["accent_flag"] not in [0,1]:
            errors.append(f"words[{i}].accent_flag должен быть 0 или 1.")

        if all(k in w for k in ["start_s","end_s"]):
            if _is_num(w["start_s"]) and _is_num(w["end_s"]) and w["end_s"] < w["start_s"]:
                errors.append(f"words[{i}]: end_s < start_s.")

    # 4) Transcript
    if not isinstance(payload["transcript"], str):
        errors.append("`transcript` должен быть строкой.")

    return errors

# ====== ГЛАВНАЯ ФУНКЦИЯ ======
def run_analysis(file):
    if file is None:
        return None, "Сначала загрузите файл.", None
    with open(file.name, "rb") as f:
        data = f.read()

    json_path, info = analyze(data, filename_hint=os.path.basename(file.name))

    try:
        with open(json_path, "r", encoding="utf-8") as f:
            payload = json.load(f)
    except Exception as e:
        err = f"JSON не прочитан: {e}"
        return json_path, err, pd.DataFrame({"error":[str(e)]})

    # ---- Проверка ----
    errors = validate_payload(payload)
    if errors:
        msg = "⚠️ Файл не соответствует схеме:\n- " + "\n- ".join(errors[:20])
        words = payload.get("words", [])[:30]
        df_preview = pd.DataFrame(words) if words else pd.DataFrame({"message":["слов нет"]})
        return json_path, msg, df_preview

    # ---- Всё ок ----
    agg = payload.get("aggregates", {})
    words = payload.get("words", [])

    cols = [
        "start_s","end_s","duration_s","pre_pause_ms","word","prob",
        "f0_mean_hz","f0_z","rms_dbfs","rms_z",
        "accent_score","accent_flag","speed_local_sylps","duration_z"
    ]
    normalized = [{c: (w.get(c, None)) for c in cols} for w in words[:30]]
    df_preview = pd.DataFrame(normalized) if normalized else pd.DataFrame({"message":["слов нет"]})

    log = (
        f"✅ JSON соответствует схеме v{payload.get('version','?')}\n"
        f"Длительность: {agg.get('duration_sec', info.get('duration_sec'))} с\n"
        f"Voiced: {agg.get('voiced_duration_sec', '—')} с  |  ratio: {agg.get('voiced_ratio', '—')}\n"
        f"Слов распознано: {agg.get('words_count', info.get('words'))}\n"
        f"Скорость речи: {agg.get('speech_rate_wps', '—')} слов/сек\n"
        f"RMS среднее: {agg.get('rms_dbfs_mean', '—')} dBFS\n"
        f"F0: mean {agg.get('f0_mean_hz', '—')} Hz, median {agg.get('f0_median_hz', '—')} Hz, std {agg.get('f0_std_hz', '—')}\n"
        f"Стабильность F0: {agg.get('f0_stability', '—')}\n"
        f"VAD: {agg.get('vad', info.get('vad'))}\n"
        f"Время анализа: {info.get('elapsed_sec')} с"
    )

    return json_path, log, df_preview

# ====== UI ======
with gr.Blocks() as demo:
    gr.Markdown(INTRO)
    with gr.Row():
        inp = gr.File(label="Аудио")
    btn = gr.Button("Анализ")
    with gr.Row():
        out_file = gr.File(label="Результаты (JSON)")
        out_log = gr.Textbox(label="Статус и сводка", lines=12)
    gr.Markdown("## Предпросмотр `words` (первые 30 записей)")
    out_df = gr.Dataframe(interactive=False)

    btn.click(run_analysis, inputs=inp, outputs=[out_file, out_log, out_df])

if __name__ == "__main__":
    demo.launch()