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, )