app.pyi

import json
import os
import tempfile
from typing import Any, Dict, List, Optional

import gradio as gr

from pipeline import IncidentPipeline, IncidentResult, serialize_result
from preprocess import truncate_logs

from gradio.events import Dependency

class DownloadOnlyFile(gr.File):
    """Файл только для скачивания, скрытый из OpenAPI-схемы Gradio."""

    is_template = True

    @property
    def skip_api(self) -> bool:
        return True
    from typing import Callable, Literal, Sequence, Any, TYPE_CHECKING
    from gradio.blocks import Block
    if TYPE_CHECKING:
        from gradio.components import Timer


pipeline = IncidentPipeline()


LABEL_DISPLAY = {
    "oom": "Переполнение памяти (OOM)",
    "timeout": "Таймаут",
    "auth_failure": "Ошибка аутентификации/авторизации",
    "db_connection": "Сбой подключения к базе данных",
    "dns_resolution": "Ошибка DNS",
    "tls_handshake": "Ошибка TLS-рукопожатия",
    "crashloop": "CrashLoop / повторные рестарты",
    "null_pointer": "NullPointer / None reference",
    "resource_exhaustion": "Исчерпание ресурсов",
    "network_partition": "Сетевая изоляция",
}

SOURCE_DISPLAY = {
    "python": "Python",
    "java": "Java",
    "node": "Node.js",
    "k8s": "Kubernetes",
    "auto": "Auto",
}

SIGNATURE_DISPLAY = {
    "stacktrace": "стектрейс",
    "timestamps": "таймстемпы",
    "log_levels": "уровни логов",
    "k8s": "ошибки Kubernetes",
    "oom": "признаки OOM",
    "timeout": "упоминания таймаута",
}

SPEC_SUFFIX = "_specific"


def human_label(label: str) -> str:
    if label.endswith(SPEC_SUFFIX):
        base = label[: -len(SPEC_SUFFIX)]
        source_name = SOURCE_DISPLAY.get(base, base)
        return f"Категория, специфичная для {source_name}"
    return LABEL_DISPLAY.get(label, label)


def human_signature(sig: str) -> str:
    return SIGNATURE_DISPLAY.get(sig, sig)


def env_flag(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.lower() in ("1", "true", "yes", "on")


def format_incident_section(result: IncidentResult) -> str:
    alt_text = ", ".join(
        f"{human_label(a['label'])} ({a['score']:.2f})" for a in result.incident_alternatives
    )
    sigs = ", ".join(human_signature(sig) for sig in result.signatures) if result.signatures else "нет"
    return (
        f"**Инцидент:** {human_label(result.incident_label)} (уверенность {result.incident_score:.2f})\n\n"
        f"**Альтернативы:** {alt_text if alt_text else 'н/д'}\n\n"
        f"**Обнаруженные сигнатуры:** {sigs}"
    )


def format_cause_section(result: IncidentResult) -> str:
    checks_md = "\n".join([f"- {c}" for c in result.checks])
    return f"**Вероятная причина:** {result.likely_cause}\n\n**Проверки / следующие шаги:**\n{checks_md}"


def analyze_logs(logs: str, source: str, use_retrieval: bool, use_nli: bool, verbosity: int):
    try:
        res = pipeline.process(
            logs,
            source=source,
            use_retrieval=use_retrieval,
            use_nli=use_nli,
            verbosity=verbosity,
        )
    except Exception as exc:
        message = f"Ошибка: {exc}"
        empty_table: List[List[Any]] = []
        return (
            message,
            "",
            "",
            empty_table,
            empty_table,
            None,
            f"Сбой: {exc}",
        )

    retrieval_rows = [
        [r["title"], round(r["score"], 3), r["path"], r["excerpt"]]
        for r in res.retrieved
    ]
    verification_rows = [
        [v["hypothesis"], v["label"], round(v["score"], 3)] for v in res.verification
    ]
    state_payload = serialize_result(res)
    return (
        format_incident_section(res),
        res.explanation,
        format_cause_section(res),
        retrieval_rows,
        verification_rows,
        state_payload,
        "Анализ завершён.",
    )


def ticket_template(state: Optional[str], logs: str) -> str:
    if not state:
        return "Сначала запустите анализ."
    try:
        parsed = json.loads(state) if isinstance(state, str) else state
    except Exception:
        return "Состояние повреждено. Повторите анализ."
    clipped_logs = truncate_logs(logs, head_lines=30, tail_lines=10, max_lines=60)
    checks = parsed.get("checks") or []
    checks_md = "\n".join(f"- {c}" for c in checks)
    summary = f"{human_label(parsed.get('incident_label','?'))} — {parsed.get('explanation','')[:180]}"
    template = (
        f"Сводка:\n{summary}\n\n"
        f"Шаги для воспроизведения:\n- Опишите последовательность, которая привела к сбою.\n- Приложите проблемный запрос или данные.\n\n"
        f"Ожидаемый результат:\n- Сервис успешно обрабатывает запрос.\n\n"
        f"Фактический результат:\n- {parsed.get('likely_cause','')}\n\n"
        f"Проверки / дальнейшие шаги:\n{checks_md}\n\n"
        f"Фрагмент логов:\n{clipped_logs}\n"
    )
    return template


def export_json(state: Optional[str]):
    if not state:
        return None
    # If state is dict, dump; if already JSON string, use as-is.
    data = json.dumps(state, ensure_ascii=False, indent=2) if isinstance(state, dict) else state
    tmp = tempfile.NamedTemporaryFile("w", delete=False, suffix=".json", encoding="utf-8")
    tmp.write(data)
    tmp.flush()
    tmp.close()
    return tmp.name


with gr.Blocks(title="Анализатор логов") as demo:
    gr.Markdown("# Анализатор логов\nВставьте логи/стектрейс и получите тип инцидента, объяснения и подсказки по расследованию.")
    # Скрытое поле для сериализованного состояния.
    state_box = gr.Textbox(visible=False, show_label=False)

    with gr.Row():
        with gr.Column(scale=5):
            logs_input = gr.Textbox(lines=20, label="Логи / стек", placeholder="Вставьте логи сюда...")
            source_dropdown = gr.Dropdown(
                ["auto", "python", "java", "node", "k8s"],
                value="auto",
                label="Источник",
            )
            use_retrieval = gr.Checkbox(value=True, label="Использовать поиск по базе знаний")
            use_nli = gr.Checkbox(value=False, label="Проверять гипотезы (NLI)")
            verbosity_slider = gr.Slider(0, 2, value=1, step=1, label="Детализация объяснения")
            analyze_btn = gr.Button("Анализировать")
            ticket_btn = gr.Button("Сформировать шаблон тикета")
            export_btn = gr.Button("Экспорт JSON")
            json_output = DownloadOnlyFile(label="Экспорт JSON")
            status = gr.Markdown("Готово.")
        with gr.Column(scale=6):
            with gr.Tab("Тип инцидента"):
                incident_md = gr.Markdown()
            with gr.Tab("Пояснение"):
                explanation_md = gr.Markdown()
            with gr.Tab("Причина и проверки"):
                cause_md = gr.Markdown()
            with gr.Tab("Найденные ранбуки"):
                retrieval_df = gr.Dataframe(
                    headers=["Название", "Сходство", "Путь", "Фрагмент"],
                    datatype=["str", "number", "str", "str"],
                    interactive=False,
                )
            with gr.Tab("Проверка гипотез"):
                verification_df = gr.Dataframe(
                    headers=["Гипотеза", "Метка", "Счёт"],
                    datatype=["str", "str", "number"],
                    interactive=False,
                )
            with gr.Tab("Шаблон тикета"):
                ticket_md = gr.Markdown()

    analyze_btn.click(
        fn=analyze_logs,
        inputs=[logs_input, source_dropdown, use_retrieval, use_nli, verbosity_slider],
        outputs=[incident_md, explanation_md, cause_md, retrieval_df, verification_df, state_box, status],
    )

    ticket_btn.click(
        fn=ticket_template,
        inputs=[state_box, logs_input],
        outputs=ticket_md,
    )

    export_btn.click(
        fn=export_json,
        inputs=state_box,
        outputs=json_output,
    )


if __name__ == "__main__":
    in_hf_space = bool(os.getenv("SPACE_ID") or os.getenv("HF_SPACE"))
    share_flag = False if in_hf_space else env_flag("GRADIO_SHARE", default=False)
    host = os.getenv("GRADIO_HOST") or os.getenv("GRADIO_SERVER_NAME") or "127.0.0.1"
    port = int(os.getenv("PORT") or os.getenv("GRADIO_SERVER_PORT") or 7860)
    demo.queue(api_open=False).launch(
        server_name=host,
        server_port=port,
        share=share_flag,
        show_api=False,
    )