147 lines
6.0 KiB
Python
147 lines
6.0 KiB
Python
"""
|
||
test_routes.py — Тестовый слой (/test Blueprint).
|
||
|
||
Мост к БД извне для отладки и диагностики.
|
||
Не зависит от бизнес-слоёв, только от db.py.
|
||
|
||
Эндпоинты:
|
||
GET /test — статус БД + список доступных действий
|
||
POST /test status — только статус БД
|
||
POST /test createdb — создать базу данных (если не существует)
|
||
POST /test tables — список таблиц в public схеме
|
||
POST /test sql {...} — выполнить произвольный SQL
|
||
"""
|
||
|
||
import os
|
||
from psycopg2 import sql as psysql
|
||
from flask import Blueprint, jsonify, request
|
||
import db
|
||
import llm_client
|
||
|
||
test_bp = Blueprint("test", __name__)
|
||
|
||
|
||
@test_bp.route("/test", methods=["GET", "POST"])
|
||
def test():
|
||
"""
|
||
Тестовый слой: проверка БД и выполнение запросов.
|
||
GET — статус подключения
|
||
POST — выполнить действие (status/tables/sql/createdb)
|
||
"""
|
||
|
||
# ── GET — показать статус и список команд ────────────────────
|
||
if request.method == "GET":
|
||
conn, err = db.connect()
|
||
return jsonify({
|
||
"db": "ok" if conn else f"fail: {err}",
|
||
"actions": [
|
||
"GET /test — статус БД",
|
||
"POST /test status — статус БД (JSON)",
|
||
"POST /test createdb — создать БД (если нет)",
|
||
"POST /test tables — список таблиц",
|
||
"POST /test exec {...} — INSERT/UPDATE/DDL (без fetch)",
|
||
"POST /test llm {...} — тест LLM (промпт → ответ)",
|
||
"POST /test sql {...} — SELECT (с fetch)",
|
||
],
|
||
})
|
||
|
||
# ── POST — выполнить действие ────────────────────────────────
|
||
data = request.get_json(silent=True) or {}
|
||
action = data.get("action", "status")
|
||
|
||
if action == "status":
|
||
conn, err = db.connect()
|
||
return jsonify({"db": "ok" if conn else f"fail: {err}"})
|
||
|
||
if action == "tables":
|
||
result, err = db.query(
|
||
"SELECT table_name FROM information_schema.tables "
|
||
"WHERE table_schema = 'public' ORDER BY table_name"
|
||
)
|
||
if err:
|
||
return jsonify({"error": err}), 500
|
||
return jsonify({"tables": [r[0] for r in result["rows"]]})
|
||
|
||
if action == "createdb":
|
||
# Подключаемся к дефолтной БД 'postgres' и создаём целевую
|
||
try:
|
||
conn = db._pg_connect("postgres")
|
||
conn.autocommit = True
|
||
cur = conn.cursor()
|
||
db_name = data.get("name") or os.environ.get("DB_NAME", "contracts")
|
||
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (db_name,))
|
||
if cur.fetchone():
|
||
msg = f"DB '{db_name}' already exists"
|
||
else:
|
||
cur.execute(psysql.SQL("CREATE DATABASE {}").format(psysql.Identifier(db_name)))
|
||
msg = f"DB '{db_name}' created"
|
||
cur.close()
|
||
conn.close()
|
||
return jsonify({"result": msg})
|
||
except Exception as e:
|
||
return jsonify({"error": str(e)}), 500
|
||
|
||
if action == "exec":
|
||
# Выполнить INSERT/UPDATE/DELETE/DDL через db.execute()
|
||
# (не делает fetch, поэтому подходит для CREATE TABLE и т.д.)
|
||
sql_text = data.get("sql", "")
|
||
if not sql_text:
|
||
return jsonify({"error": "no sql"}), 400
|
||
rowcount, err = db.execute(sql_text)
|
||
if err:
|
||
return jsonify({"error": err}), 500
|
||
return jsonify({"rowcount": rowcount})
|
||
|
||
if action == "sql":
|
||
sql_text = data.get("sql", "")
|
||
if not sql_text:
|
||
return jsonify({"error": "no sql"}), 400
|
||
result, err = db.query(sql_text)
|
||
if err:
|
||
return jsonify({"error": err}), 500
|
||
return jsonify(result)
|
||
|
||
if action == "llm":
|
||
prompt = data.get("prompt", "Скажи 'Привет, мир!'")
|
||
model = data.get("model")
|
||
max_tokens = data.get("max_tokens", 100)
|
||
result = llm_client.ask(prompt, model=model, max_tokens=max_tokens)
|
||
# добавим отладочную информацию
|
||
result["_debug"] = {
|
||
"key_prefix": os.environ.get("LLM_API_KEY", "")[:6] + "...",
|
||
"key_len": len(os.environ.get("LLM_API_KEY", "")),
|
||
"url": os.environ.get("LLM_API_URL", llm_client.DEFAULT_URL),
|
||
}
|
||
if "error" in result:
|
||
return jsonify(result), 500
|
||
return jsonify(result)
|
||
|
||
if action == "debug":
|
||
# Показать переменные окружения (с маскировкой ключей)
|
||
import os as _os
|
||
import hashlib
|
||
vars_info = {}
|
||
for k, v in sorted(_os.environ.items()):
|
||
if any(x in k.upper() for x in ("KEY", "PASS", "SECRET", "TOKEN")):
|
||
if v:
|
||
h = hashlib.sha256(v.encode()).hexdigest()
|
||
vars_info[k] = f"{v[:4]}...{v[-4:]} (len={len(v)}, hash={h[:8]})"
|
||
else:
|
||
vars_info[k] = "(empty)"
|
||
elif any(x in k.upper() for x in ("DB_", "LLM_", "FLASK", "PORT")):
|
||
vars_info[k] = v
|
||
# добавим проверку h2 и httpx
|
||
try:
|
||
import h2
|
||
vars_info["h2_available"] = True
|
||
except ImportError:
|
||
vars_info["h2_available"] = False
|
||
try:
|
||
import httpx
|
||
vars_info["httpx_version"] = httpx.__version__
|
||
except ImportError:
|
||
vars_info["httpx_version"] = "not installed"
|
||
return jsonify({"env": vars_info, "python": _os.sys.version})
|
||
|
||
return jsonify({"error": f"unknown action: {action}"}), 400
|