"""Встроенные slim-tools для LLM-оркестратора: SQL (preview/execute), логи, сводка, классификация.""" from __future__ import annotations import json import re import sqlite3 import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from flask import Flask from app.services.admin_dashboard_service import sqlite_bind_paths, tail_text_file from app.services.client_log_upload import ( client_logs_root_resolved, latest_uploaded_log_file, safe_path_segment, ) # --------------------------------------------------------------------------- # OpenAI-style tool definitions (llama-server /v1/chat/completions) # --------------------------------------------------------------------------- SLIM_TOOL_DEFINITIONS: List[Dict[str, Any]] = [ { "type": "function", "function": { "name": "slim_sql", "description": ( "Выполнить безопасный запрос к SQLite WESP (только SELECT). " "Режим preview — план EXPLAIN без данных; execute — строки результата " "(только если пользователь включил execute в запросе чата)." ), "parameters": { "type": "object", "properties": { "bind": { "type": "string", "description": "База: recipes или reports", "enum": ["recipes", "reports"], }, "sql": {"type": "string", "description": "Один оператор SELECT или WITH … SELECT"}, "mode": { "type": "string", "enum": ["preview", "execute"], "description": "preview по умолчанию; execute только при явном разрешении в чате", }, }, "required": ["bind", "sql"], }, }, }, { "type": "function", "function": { "name": "slim_extract", "description": ( "Извлечь из лога ошибки, IP, HTTP-коды, метки времени, характерные строки." ), "parameters": { "type": "object", "properties": { "source": { "type": "string", "enum": ["server_log", "client_log", "raw_text"], "description": "server_log — WESP_ADMIN_LOG_PATH; client_log — выгрузки клиентов; raw_text — переданный текст", }, "client_id": { "type": "string", "description": "Для client_log: node_id или сегмент каталога клиента", }, "text": { "type": "string", "description": "Для raw_text: фрагмент лога (несколько тысяч символов)", }, }, "required": ["source"], }, }, }, { "type": "function", "function": { "name": "slim_summary", "description": "Сжать большой лог до 1–2 предложений (экстрактивно, без второго LLM).", "parameters": { "type": "object", "properties": { "source": { "type": "string", "enum": ["server_log", "client_log", "raw_text"], }, "client_id": {"type": "string"}, "text": {"type": "string"}, }, "required": ["source"], }, }, }, { "type": "function", "function": { "name": "slim_xsum", "description": "Как slim_summary, но жёстче: максимум два коротких предложения.", "parameters": { "type": "object", "properties": { "source": { "type": "string", "enum": ["server_log", "client_log", "raw_text"], }, "client_id": {"type": "string"}, "text": {"type": "string"}, }, "required": ["source"], }, }, }, { "type": "function", "function": { "name": "slim_sentiment", "description": ( "Оценить операционный тон лога: critical / warn / normal / backup_routine и уверенность 0..1." ), "parameters": { "type": "object", "properties": { "source": { "type": "string", "enum": ["server_log", "client_log", "raw_text"], }, "client_id": {"type": "string"}, "text": {"type": "string"}, }, "required": ["source"], }, }, }, { "type": "function", "function": { "name": "slim_emotions", "description": ( "Грубая эмоциональная окраска событий в логе: stressed / calm / mixed (эвристика, не NLP-модель)." ), "parameters": { "type": "object", "properties": { "source": { "type": "string", "enum": ["server_log", "client_log", "raw_text"], }, "client_id": {"type": "string"}, "text": {"type": "string"}, }, "required": ["source"], }, }, }, ] _IP_RE = re.compile( r"\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b" ) _HTTP_CODE_RE = re.compile(r"\b(?:HTTP/\S+\s+)?(\d{3})\b|status[:\s=]+(\d{3})\b", re.I) _ERR_HINT_RE = re.compile( r"(error|exception|traceback|fatal|critical|failed|errno|segmentation|panic|" r"ошибк|исключен|трассиров)\b", re.I, ) _BAD_SQL_KW = re.compile( r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|ATTACH|DETACH|REPLACE|" r"VACUUM|REINDEX|PRAGMA|TRUNCATE)\b", re.I, ) def _tool_result( *, ok: bool, tool: str, data: Any = None, error: Optional[str] = None, warnings: Optional[List[str]] = None, truncated: bool = False, ) -> Dict[str, Any]: return { "ok": ok, "tool": tool, "data": data, "error": error, "warnings": warnings or [], "truncated": truncated, } def _validate_select_only(sql: str) -> str: s = (sql or "").strip() if not s: raise ValueError("Пустой SQL.") if s.endswith(";"): s = s[:-1].strip() if ";" in s: raise ValueError("Разрешён ровно один SQL-оператор (без «;» внутри).") if not re.match(r"^\s*(SELECT|WITH)\b", s, re.I): raise ValueError("Разрешены только SELECT или WITH … SELECT.") if _BAD_SQL_KW.search(s): raise ValueError("Запрещённые ключевые слова в SQL (только чтение).") return s def _open_sqlite_readonly(path: Path) -> sqlite3.Connection: uri = path.resolve().as_uri() + "?mode=ro" conn = sqlite3.connect(uri, uri=True, timeout=12.0) conn.row_factory = sqlite3.Row return conn def _resolve_log_text( app: Flask, *, source: str, client_id: str = "", raw_text: str = "", max_bytes: int, max_lines: int, ) -> Tuple[str, List[str]]: warnings: List[str] = [] src = (source or "").strip().lower() if src == "raw_text": t = (raw_text or "").strip() if not t: return "", ["raw_text: пусто"] if len(t.encode("utf-8", errors="replace")) > max_bytes: t = t.encode("utf-8", errors="replace")[:max_bytes].decode("utf-8", errors="replace") warnings.append("Текст обрезан по лимиту байт.") return t, warnings if src == "server_log": raw = (app.config.get("WESP_ADMIN_LOG_PATH") or "").strip() if not raw: return "", ["WESP_ADMIN_LOG_PATH не задан."] p = Path(raw).expanduser() if not p.is_absolute(): p = Path(str(app.config.get("BASE_DIR") or ".")) / p p = p.resolve() lines, err = tail_text_file(p, max_lines=max_lines, max_bytes=max_bytes) if err: return "", [err] return "\n".join(lines), warnings if src == "client_log": upload_dir = str(app.config.get("WESP_CLIENT_LOG_UPLOAD_DIR") or "").strip() if not upload_dir: return "", ["WESP_CLIENT_LOG_UPLOAD_DIR не задан."] base_dir = str(app.config.get("BASE_DIR") or ".") try: root = client_logs_root_resolved(upload_dir, base_dir) except ValueError as e: return "", [str(e)] cid = safe_path_segment(client_id, max_len=80) if client_id else "" if not cid: return "", ["Для client_log укажите client_id."] d = root / cid latest = latest_uploaded_log_file(d) if latest is None: return "", [f"Нет файлов лога в каталоге клиента «{cid}»."] lines, err = tail_text_file(latest, max_lines=max_lines, max_bytes=max_bytes) if err: return "", [err] return "\n".join(lines), warnings return "", [f"Неизвестный source: {source!r}"] def _extract_features(text: str) -> Dict[str, Any]: lines = text.splitlines() ip_set = set(_IP_RE.findall(text)) if text else set() http_codes: List[str] = [] for m in _HTTP_CODE_RE.finditer(text or ""): g = m.group(1) or m.group(2) if g: http_codes.append(g) err_lines = [ln for ln in lines if _ERR_HINT_RE.search(ln)] # топ «сигнатур» — первые уникальные короткие строки с error sigs: List[str] = [] seen = set() for ln in err_lines[:80]: key = ln.strip()[:160] if key and key not in seen: seen.add(key) sigs.append(key) if len(sigs) >= 12: break return { "line_count": len(lines), "ipv4": sorted(ip_set)[:40], "http_like_codes": http_codes[:40], "error_line_samples": sigs, } def _summarize_text(text: str, *, strict_short: bool) -> str: if not text.strip(): return "Лог пуст или не прочитан." lines = text.splitlines() head = lines[:15] tail = lines[-25:] if len(lines) > 25 else [] err_lines = [ln for ln in lines if _ERR_HINT_RE.search(ln)] pick: List[str] = [] for block in (err_lines[:20], tail, head): for ln in block: s = ln.strip() if s and s not in pick: pick.append(s) if len(pick) >= 30: break if len(pick) >= 30: break blob = " ".join(pick)[:1200] if strict_short: return ( f"Кратко: всего строк ~{len(lines)}; ключевые фрагменты: {blob[:500]}" + ("…" if len(blob) > 500 else "") ) return ( f"В логе ~{len(lines)} строк. Сжато: {blob[:900]}" + ("…" if len(blob) > 900 else "") ) def _classify_operational(text: str) -> Dict[str, Any]: t = (text or "").lower() critical_hits = sum( 1 for w in ("fatal", "traceback", "segmentation", "panic", "critical", "emergency") if w in t ) err_hits = sum(1 for w in ("error", "exception", "failed", "errno") if w in t) backup_hits = sum( 1 for w in ("backup", "rotat", "archive", "snapshot", "dump") if w in t ) ok_hits = sum(1 for w in (" ok", "200 ", "success", "completed") if w in t) label = "normal" if critical_hits >= 1 or err_hits >= 5: label = "critical" elif err_hits >= 1: label = "warn" elif backup_hits >= 2 and err_hits == 0: label = "backup_routine" elif ok_hits >= 3 and err_hits == 0: label = "normal" # confidence heuristic confidence = min(1.0, 0.35 + 0.1 * (critical_hits + err_hits + min(backup_hits, 3))) return {"label": label, "confidence": round(confidence, 2), "hints": {"errors": err_hits, "critical": critical_hits, "backup": backup_hits}} def _classify_emotions(text: str) -> Dict[str, Any]: t = (text or "").lower() stress = sum( 1 for w in ("error", "fatal", "panic", "failed", "critical", "alarm") if w in t ) calm = sum(1 for w in (" ok", "success", "ready", "listening", "started") if w in t) if stress >= 3: label = "stressed" elif stress == 0 and calm >= 2: label = "calm" else: label = "mixed" return {"label": label, "confidence": round(min(1.0, 0.4 + 0.08 * (stress + calm)), 2)} def run_slim_sql( app: Flask, args: Dict[str, Any], *, chat_sql_mode: str, max_rows: int, ) -> Dict[str, Any]: bind = str(args.get("bind") or "").strip().lower() sql_raw = str(args.get("sql") or "") mode = str(args.get("mode") or "preview").strip().lower() if mode not in ("preview", "execute"): mode = "preview" if bind not in sqlite_bind_paths(app): return _tool_result(ok=False, tool="slim_sql", error=f"Неизвестный bind «{bind}».") try: sql = _validate_select_only(sql_raw) except ValueError as e: return _tool_result(ok=False, tool="slim_sql", error=str(e)) paths = sqlite_bind_paths(app) path = paths[bind] if not path.is_file(): return _tool_result(ok=False, tool="slim_sql", error=f"Файл БД не найден: {path}") if mode == "execute" and chat_sql_mode != "execute": return _tool_result( ok=False, tool="slim_sql", error="Режим execute отключён для этого запроса чата (выберите «Выполнять SELECT» в UI или sql_mode=execute).", ) t0 = time.monotonic() try: conn = _open_sqlite_readonly(path) except sqlite3.Error as e: return _tool_result(ok=False, tool="slim_sql", error=f"SQLite: {e}") warnings: List[str] = [] try: if mode == "preview": plan_cur = conn.execute(f"EXPLAIN QUERY PLAN {sql}") plan_rows = [dict(row) for row in plan_cur.fetchall()] return _tool_result( ok=True, tool="slim_sql", data={ "bind": bind, "mode": "preview", "plan": plan_rows[:200], "sql_echo": sql[:2000], "elapsed_sec": round(time.monotonic() - t0, 4), }, warnings=warnings, ) # execute: читаем не более max_rows строк (остальное отбрасываем; тяжёлый запрос всё равно может грузить СУБД) cur = conn.execute(sql) colnames = [d[0] for d in cur.description] if cur.description else [] out_rows: List[sqlite3.Row] = [] truncated = False for i, row in enumerate(cur): if i >= max_rows: truncated = True break out_rows.append(row) serialized = [] for r in out_rows: serialized.append({colnames[i]: r[i] for i in range(len(colnames))}) if truncated: warnings.append(f"Строк больше лимита ({max_rows}); результат обрезан.") return _tool_result( ok=True, tool="slim_sql", data={ "bind": bind, "mode": "execute", "columns": colnames, "rows": serialized, "row_count": len(serialized), "elapsed_sec": round(time.monotonic() - t0, 4), }, warnings=warnings, truncated=truncated, ) except sqlite3.Error as e: return _tool_result(ok=False, tool="slim_sql", error=f"SQLite: {e}") finally: try: conn.close() except Exception: pass def dispatch_slim_tool( app: Flask, name: str, raw_arguments: str, *, chat_sql_mode: str, max_rows: int, max_log_bytes: int, max_log_lines: int, ) -> Dict[str, Any]: """Парсит arguments JSON и вызывает обработчик.""" try: args = json.loads(raw_arguments or "{}") except json.JSONDecodeError as e: return _tool_result(ok=False, tool=name, error=f"Некорректный JSON аргументов: {e}") if not isinstance(args, dict): return _tool_result(ok=False, tool=name, error="Аргументы инструмента должны быть объектом JSON.") if name == "slim_sql": return run_slim_sql(app, args, chat_sql_mode=chat_sql_mode, max_rows=max_rows) if name in ("slim_extract", "slim_summary", "slim_xsum", "slim_sentiment", "slim_emotions"): source = str(args.get("source") or "") client_id = str(args.get("client_id") or "") raw_text = str(args.get("text") or "") text, w = _resolve_log_text( app, source=source, client_id=client_id, raw_text=raw_text, max_bytes=max_log_bytes, max_lines=max_log_lines, ) if not text and w: return _tool_result(ok=False, tool=name, error="; ".join(w)) if name == "slim_extract": feat = _extract_features(text) truncated = any("обрезан" in str(x) for x in w) return _tool_result( ok=True, tool=name, data=feat, warnings=w, truncated=truncated, ) if name == "slim_summary": summary = _summarize_text(text, strict_short=False) return _tool_result(ok=True, tool=name, data={"summary_ru": summary}, warnings=w) if name == "slim_xsum": summary = _summarize_text(text, strict_short=True) return _tool_result(ok=True, tool=name, data={"summary_ru": summary}, warnings=w) if name == "slim_sentiment": cl = _classify_operational(text) return _tool_result(ok=True, tool=name, data=cl, warnings=w) if name == "slim_emotions": em = _classify_emotions(text) return _tool_result(ok=True, tool=name, data=em, warnings=w) return _tool_result(ok=False, tool=name, error=f"Неизвестный инструмент: {name}")