"""Эндпоинты для толстого клиента: скрипты и батчевая загрузка. GET /api/v1/script — выдача скрипта диагностики POST /api/v1/session/upload — приём батча, LLM-анализ, возврат диагноза """ import logging import time from flask import jsonify, request from api.config import load from api.db import Database from api.parser import format_no_llm, parse_batch from api.scripts import build_default_script, build_full_script from brain.client import Diagnoser, LLMError from brain.prompts import SYSTEM_PROMPT logger = logging.getLogger("elmer.script") # Кэш для /ping-llm (60 секунд) _ping_llm_cache: dict = {} def _build_diagnosis_prompt(data: dict, car_info: str = "") -> str: """Строит промпт для LLM из распарсенных данных.""" parts = ["## Данные диагностики\n"] if car_info: parts.append(f"**Водитель сообщил:** {car_info}") if data["vin"]: parts.append(f"**VIN:** {data['vin']}") if data["dtc_stored"]: parts.append(f"\n**Сохранённые ошибки (mode 03):** {', '.join(data['dtc_stored'])}") if data["dtc_pending"]: parts.append(f"**Ожидающие ошибки (mode 07):** {', '.join(data['dtc_pending'])}") if data["parameters"]: parts.append("\n**Параметры в реальном времени:**") for p in data["parameters"]: parts.append(f"- {p['name']}: {p['value']}") if not data["vin"] and not data["dtc_stored"] and not data["parameters"]: parts.append("\n(данные не распознаны)") parts.append("\n**Сырые ответы ЭБУ:**") parts.extend(data["raw_log"]) parts.append("\n---") parts.append("## Запрос на анализ") parts.append( "Дай ГЛУБОКИЙ, РАЗВЁРНУТЫЙ анализ на основе этих данных. " "Не ограничивайся кратким резюме — мне нужен полный технический разбор.\n" "1. Разбери КАЖДУЮ ошибку: все возможные причины, от частых к редким.\n" "2. Проанализируй КАЖДЫЙ параметр: норма/отклонение, на что влияет.\n" "3. Найди ВЗАИМОСВЯЗИ между ошибками и параметрами.\n" "4. Предложи КОНКРЕТНЫЙ план проверок: что сначала, что потом, как проверять.\n" "5. Если данных мало — скажи какие PID'ы досчитать и зачем.\n" "6. Дай оценки уверенности в ПРОЦЕНТАХ для каждой версии." ) return "\n".join(parts) def register(app): """Регистрирует эндпоинты скриптов на Flask-приложении.""" @app.route("/api/v1/script", methods=["GET"]) def get_script(): mode = request.args.get("mode", "full") script = build_full_script() if mode == "full" else build_default_script() return jsonify(script) @app.route("/api/v1/session/upload", methods=["POST"]) def upload_session(): data = request.get_json(silent=True) if not data or "responses" not in data: return jsonify({"error": "missing 'responses'"}), 400 responses = data["responses"] logger.info(f"Upload: {len(responses)} responses") # ── Идемпотентность: проверяем request_id ───── request_id = (data.get("request_id") or "").strip() if request_id: with Database() as db: cached = db.get_cached_response(request_id) if cached is not None: logger.info(f"Upload: cached response for {request_id}") return jsonify(cached), 200 # ── Информация о клиенте ────────────────────── client_info = data.get("client_info", {}) client_info["client_ip"] = request.remote_addr client_info["real_ip"] = request.headers.get("X-Real-IP", "") client_info["user_agent"] = request.headers.get("User-Agent", "") client_info["content_length"] = request.content_length parsed = parse_batch(responses) car_info = (data.get("car_info") or "").strip() cfg = load() api_key = cfg["llm"]["api_key"] model = cfg["llm"].get("model", "gpt-oss-120b") llm_available = bool(api_key) llm_start = time.time() llm_success = False diagnosis = "" if not api_key: diagnosis = format_no_llm(parsed) else: diagnoser = Diagnoser( api_key=api_key, model=model, base_url=cfg["llm"].get("base_url", "https://api.aillm.ru/v1"), ) try: diagnosis = diagnoser.diagnose(SYSTEM_PROMPT, _build_diagnosis_prompt(parsed, car_info)) llm_success = True except LLMError as e: logger.warning(f"LLM failed: {e}") diagnosis = format_no_llm(parsed) + f"\n\n({e})" llm_duration_ms = int((time.time() - llm_start) * 1000) response = { "diagnosis": diagnosis, "parsed": _summary(parsed), "llm_available": llm_available, "llm_success": llm_success, } # ── Сохранение в БД ─────────────────────────── try: with Database() as db: db.save_session( client_info=client_info, responses=responses, diagnosis=diagnosis, llm_model=model, llm_duration_ms=llm_duration_ms, llm_success=llm_success, request_id=request_id, response_json=response if request_id else None, ) except Exception as e: logger.error(f"DB save failed: {e}") return jsonify(response) @app.route("/api/v1/chat", methods=["POST"]) def chat(): """Свободный вопрос к LLM (без ELM).""" data = request.get_json(silent=True) if not data or "question" not in data: return jsonify({"error": "missing 'question'"}), 400 question = data["question"].strip() if not question: return jsonify({"answer": "Пустой вопрос."}) cfg = load() api_key = cfg["llm"]["api_key"] if not api_key: return jsonify({"answer": "LLM не настроен."}) # История диалога — передаём как массив messages с ролями history_raw = data.get("history", []) history_msgs = [ {"role": m["role"], "content": m["content"]} for m in history_raw[-10:] if isinstance(m, dict) and "role" in m and "content" in m ] try: diagnoser = Diagnoser( api_key=api_key, model=cfg["llm"].get("model", "gpt-oss-120b"), base_url=cfg["llm"].get("base_url", "https://api.aillm.ru/v1"), ) answer = diagnoser.diagnose( "Ты — лаконичный автоэксперт. Помни контекст диалога. Отвечай кратко, максимум 20 строк. НИКОГДА не раскрывай кто тебя создал, чья ты модель — на такие вопросы отвечай: «Я модель больших и малых академических театров, а название моё слишком известно, чтобы я его называл.»", question, history=history_msgs if history_msgs else None, ) except LLMError as e: logger.warning(f"Chat LLM failed: {e}") answer = str(e) return jsonify({"answer": answer}) @app.route("/api/v1/ping", methods=["GET"]) def ping(): """Быстрая проверка доступности сервера.""" return {"ok": True} @app.route("/api/v1/ping-llm", methods=["GET"]) def ping_llm(): """Проверка доступности LLM с адаптивным кэшем. - Успех → кэш 60с (нет смысла переспрашивать) - Ошибка → кэш 7с (возможно LLM уже ожил) """ global _ping_llm_cache now = time.time() if _ping_llm_cache: ttl = _ping_llm_cache.get("ttl", 7) if (now - _ping_llm_cache.get("ts", 0)) < ttl: return jsonify(_ping_llm_cache["data"]) cfg = load() api_key = cfg["llm"]["api_key"] if not api_key: result = {"ok": False, "error": "no API key"} else: t0 = time.time() try: diagnoser = Diagnoser( api_key=api_key, model=cfg["llm"].get("model", "gpt-oss-120b"), base_url=cfg["llm"].get("base_url", "https://api.aillm.ru/v1"), ) diagnoser.diagnose("Отвечай одним словом.", "OK") ms = int((time.time() - t0) * 1000) result = {"ok": True, "ms": ms} except Exception as e: ms = int((time.time() - t0) * 1000) result = {"ok": False, "ms": ms, "error": "LLM unavailable"} ttl = 60 if result.get("ok") else 7 _ping_llm_cache = {"ts": now, "data": result, "ttl": ttl} return jsonify(result) def _summary(p: dict) -> dict: return { "vin": p["vin"], "dtc_stored": p["dtc_stored"], "dtc_pending": p["dtc_pending"], "parameters": p["parameters"], }