294 lines
12 KiB
Python
294 lines
12 KiB
Python
"""Эндпоинты: скрипт, загрузка сессии, чат, пробинг ELM.
|
|
|
|
GET /api/v1/script — выдача скрипта диагностики
|
|
POST /api/v1/session/upload — приём батча + LLM
|
|
POST /api/v1/chat — свободный вопрос к LLM
|
|
POST /api/v1/elm/probe — пробинг ELM327, определение уровня
|
|
|
|
См. также: api/dtc.py (DTC), api/ping.py (ping)
|
|
"""
|
|
|
|
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, build_script_for_level, build_dynamic_script
|
|
from brain.client import Diagnoser, LLMError
|
|
from brain.prompts import SYSTEM_PROMPT, DYNAMIC_PROMPT
|
|
|
|
logger = logging.getLogger("elmer.script")
|
|
|
|
|
|
def _build_diagnosis_prompt(data: dict, car_info: str = "") -> str:
|
|
"""Строит промпт для LLM из распарсенных данных."""
|
|
parts = ["## Данные диагностики\n"]
|
|
|
|
if car_info:
|
|
parts.append(f"**Водитель сообщил:** {car_info}")
|
|
|
|
if data.get("vin"):
|
|
parts.append(f"**VIN:** {data['vin']}")
|
|
|
|
if data.get("dtc_stored"):
|
|
parts.append(f"\n**Сохранённые ошибки (mode 03):** {', '.join(data['dtc_stored'])}")
|
|
if data.get("dtc_pending"):
|
|
parts.append(f"**Ожидающие ошибки (mode 07):** {', '.join(data['dtc_pending'])}")
|
|
|
|
if data.get("parameters"):
|
|
parts.append("\n**Параметры в реальном времени:**")
|
|
for p in data["parameters"]:
|
|
parts.append(f"- {p['name']}: {p['value']}")
|
|
|
|
if not data.get("vin") and not data.get("dtc_stored") and not data.get("parameters"):
|
|
parts.append("\n(данные не распознаны)")
|
|
|
|
raw_log = data.get("raw_log") or []
|
|
parts.append("\n**Сырые ответы ЭБУ:**")
|
|
parts.extend(raw_log)
|
|
|
|
parts.append("\n---")
|
|
|
|
parts.append("\n---")
|
|
parts.append("## Запрос на анализ")
|
|
parts.append(
|
|
"Дай краткий диагноз: расшифруй КАЖДЫЙ код ошибки, назови ВОЗМОЖНЫЕ причины (от частых к редким), "
|
|
"и дай СТЕПЕНЬ уверенности. БЕЗ воды, без длинных отступлений, без таблиц. "
|
|
"Если данных мало — скажи что ещё проверить."
|
|
)
|
|
return "\n".join(parts)
|
|
|
|
|
|
# Загрузка справочника DTC
|
|
_DTC_CODES: dict[str, str] = {}
|
|
|
|
def register(app):
|
|
"""Регистрирует эндпоинты скриптов на Flask-приложении."""
|
|
|
|
@app.route("/api/v1/script", methods=["GET"])
|
|
def get_script():
|
|
mode = request.args.get("mode", "full")
|
|
level = request.args.get("level")
|
|
if level is not None:
|
|
try:
|
|
script = build_script_for_level(int(level))
|
|
except (ValueError, TypeError):
|
|
script = build_default_script()
|
|
elif mode == "dynamic":
|
|
script = build_dynamic_script()
|
|
elif mode == "full":
|
|
script = build_full_script()
|
|
else:
|
|
script = 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"]
|
|
|
|
# Динамический тест: если есть — добавляем к обычным ответам
|
|
dynamic_samples = data.get("dynamic_samples")
|
|
has_dynamic = bool(dynamic_samples)
|
|
if has_dynamic:
|
|
# Прикрепляем все сэмплы к responses для парсинга
|
|
all_samples = []
|
|
for sample in dynamic_samples:
|
|
all_samples.extend(sample)
|
|
logger.info(f"Upload: {len(responses)} обычных + {len(all_samples)} динамических ответов")
|
|
responses = responses + all_samples
|
|
else:
|
|
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()
|
|
client_info["car_info"] = car_info
|
|
|
|
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:
|
|
prompt = DYNAMIC_PROMPT if has_dynamic else SYSTEM_PROMPT
|
|
diagnosis = diagnoser.diagnose(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/elm/probe", methods=["POST"])
|
|
def probe_elm():
|
|
"""Пробинг ELM327: определение уровня устройства.
|
|
|
|
Принимает MAC и сырые ответы на команды пробинга от Android-клиента.
|
|
Клиент посылает команды из списка, сервер классифицирует ответы.
|
|
|
|
Body: {
|
|
"mac": "AA:BB:CC:...",
|
|
"responses": [
|
|
{"cmd": "ATE0", "raw": "OK"},
|
|
{"cmd": "ATL0", "raw": "OK"},
|
|
...
|
|
]
|
|
}
|
|
|
|
Returns: профиль устройства (level, supported, unsupported, ...)
|
|
"""
|
|
data = request.get_json(silent=True)
|
|
if not data or "mac" not in data or "responses" not in data:
|
|
return jsonify({"error": "missing 'mac' or 'responses'"}), 400
|
|
|
|
mac = data["mac"].strip()
|
|
responses = data["responses"]
|
|
|
|
if not mac:
|
|
return jsonify({"error": "empty mac"}), 400
|
|
|
|
# Классификация ответов через сервис
|
|
from obd.classifier import determine_level
|
|
|
|
# Собираем ответы в словарь cmd→raw
|
|
resp_map = {}
|
|
for r in responses:
|
|
cmd = (r.get("cmd") or "").strip().upper()
|
|
raw = (r.get("raw") or "").strip()
|
|
resp_map[cmd] = raw
|
|
|
|
result = determine_level(resp_map)
|
|
result["mac"] = mac
|
|
|
|
_save_profile(mac, result)
|
|
return jsonify(result)
|
|
|
|
|
|
@app.route("/api/v1/elm/profile/<mac>", methods=["GET"])
|
|
def get_elm_profile(mac: str):
|
|
"""Возвращает сохранённый профиль устройства по MAC."""
|
|
with Database() as db:
|
|
p = db.get_device_profile(mac)
|
|
if p is None:
|
|
return jsonify({"error": "not found"}), 404
|
|
return jsonify(p)
|
|
|
|
|
|
def _save_profile(mac: str, profile: dict):
|
|
"""Сохраняет профиль в БД (best-effort)."""
|
|
try:
|
|
with Database() as db:
|
|
db.save_device_profile(mac, profile)
|
|
except Exception as e:
|
|
logger.error(f"Failed to save device profile for {mac}: {e}")
|
|
|
|
|
|
def _summary(p: dict) -> dict:
|
|
return {
|
|
"vin": p["vin"],
|
|
"dtc_stored": p["dtc_stored"],
|
|
"dtc_pending": p["dtc_pending"],
|
|
"parameters": p["parameters"],
|
|
}
|