johannes/sql-ai-gui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2104bfc74aa54d7bcb253873f19c1d1835f5d31e
sql-ai-gui/sql_ai_gui_local_ollama.py
T

1205 lines
44 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
JR SQL AI GUI (local Ollama, resilient version v9)
Improvements over v4:
- auto-prefill BASE_MODEL / EXPERT_MODEL / EXTRA_MODELS from .env, .env.example and Modelfile
- fills missing GUI fields before maintenance actions, so buttons can usually be used directly
- clearer validation messages with source hints
- keeps inference model separate from maintenance config
"""
from __future__ import annotations
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, List, Optional, Tuple
import requests
from PySide6.QtCore import Qt, QThread, Signal, QTimer
from PySide6.QtGui import QAction, QFont, QKeySequence
from PySide6.QtWidgets import (
QApplication,
QCheckBox,
QComboBox,
QFileDialog,
QGridLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QMainWindow,
QMessageBox,
QPushButton,
QPlainTextEdit,
QSplitter,
QStatusBar,
QVBoxLayout,
QWidget,
QTextBrowser,
)
ANSI_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
SQL_KW_RE = re.compile(
r"\b(select|from|where|join|group|order|having|insert|update|delete|create|alter|drop|with|merge)\b",
re.IGNORECASE,
)
FENCE_RE = re.compile(r"```(\w+)?\s*\n(.*?)\n```", re.DOTALL)
FROM_RE = re.compile(r"^\s*FROM\s+(.+?)\s*$", re.IGNORECASE | re.MULTILINE)
VAR_RE = re.compile(r"\$\{?BASE_MODEL\}?", re.IGNORECASE)
MENU_STYLE_SHEET = """
QMenuBar {
background-color: #2b2f34;
color: #f1f3f5;
border: 1px solid #3a3f46;
}
QMenuBar::item {
background: transparent;
color: #f1f3f5;
padding: 5px 10px;
margin: 1px 2px;
border-radius: 4px;
}
QMenuBar::item:selected {
background: #3a4250;
color: #ffffff;
}
QMenuBar::item:pressed {
background: #465062;
color: #ffffff;
}
QMenu {
background-color: #25292e;
color: #f1f3f5;
border: 1px solid #3a3f46;
padding: 4px;
}
QMenu::item {
background-color: transparent;
color: #f1f3f5;
padding: 6px 28px 6px 12px;
border-radius: 4px;
}
QMenu::item:selected {
background-color: #3a4250;
color: #ffffff;
}
QMenu::item:disabled {
color: #8a9099;
}
QMenu::separator {
height: 1px;
background: #454b54;
margin: 6px 8px;
}
"""
# -----------------------------
# Helpers
# -----------------------------
def strip_ansi(text: str) -> str:
return ANSI_RE.sub("", text or "")
def human_error(e: Exception) -> str:
return strip_ansi(f"{type(e).__name__}: {e}")
def read_env_file(env_path: Path) -> dict[str, str]:
data: dict[str, str] = {}
if not env_path.is_file():
return data
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip()
if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
value = value[1:-1]
data[key] = value
return data
def write_env_file(env_path: Path, updates: dict[str, str]) -> None:
current = read_env_file(env_path)
current.update({k: v for k, v in updates.items() if v is not None})
lines = [f"{k}={v}" for k, v in current.items()]
env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def extract_sql_blocks(markdown_text: str) -> List[str]:
blocks: List[str] = []
for m in FENCE_RE.finditer(markdown_text):
lang = (m.group(1) or "").strip().lower()
body = (m.group(2) or "").strip()
if not body:
continue
if lang in {"sql", "tsql", "t-sql", "mssql"}:
blocks.append(body)
elif SQL_KW_RE.search(body):
blocks.append(body)
return blocks
def build_sql_only_text(blocks: List[str]) -> str:
if not blocks:
return ""
return "\n\n-- ----------------------------------------\n\n".join(blocks) + "\n"
def looks_like_registry_pull_error(text: str) -> bool:
normalized = strip_ansi(text).lower()
return "pull model manifest" in normalized and "file does not exist" in normalized
def detect_project_root(start: Path) -> Path:
start = start.resolve()
candidates = [start, *start.parents]
for path in candidates:
if (path / ".env").exists() or (path / ".env.example").exists() or (path / "Modelfile").exists() or (path / "scripts").is_dir():
return path
return start
def resolve_path_from_value(root: Path, raw_value: str, default: Path) -> Path:
value = (raw_value or '').strip()
if not value:
return default.resolve()
p = Path(value).expanduser()
if not p.is_absolute():
p = (root / p)
return p.resolve()
def parse_modelfile_for_base(modelfile_path: Path) -> Tuple[str, str]:
"""
Returns: (base_model, source_hint)
source_hint is empty when nothing was found.
"""
if not modelfile_path.is_file():
return "", ""
try:
text = modelfile_path.read_text(encoding="utf-8")
except Exception:
return "", ""
m = FROM_RE.search(text)
if not m:
return "", ""
raw = m.group(1).strip().strip('"').strip("'")
if not raw:
return "", ""
if VAR_RE.search(raw):
return "", f"{modelfile_path} enthält nur einen BASE_MODEL-Platzhalter"
return raw, f"{modelfile_path} (FROM ...)"
@dataclass
class ResolvedDefaults:
ollama_url: str
ollama_bin: str
base_model: str
expert_model: str
extra_models: str
inference_model: str
modelfile_path: Path
source_notes: List[str]
def resolve_defaults(root: Path, explicit_modelfile: Optional[Path] = None, current_inference_model: str = "") -> ResolvedDefaults:
env_path = root / ".env"
env_example_path = root / ".env.example"
env_data = read_env_file(env_path)
env_example_data = read_env_file(env_example_path)
default_modelfile_path = explicit_modelfile.resolve() if explicit_modelfile else (root / "Modelfile").resolve()
notes: List[str] = []
def pick(key: str, default: str = "") -> str:
if key in env_data and env_data[key].strip():
notes.append(f"{key} aus .env")
return env_data[key].strip()
if key in env_example_data and env_example_data[key].strip():
notes.append(f"{key} aus .env.example")
return env_example_data[key].strip()
value = os.environ.get(key, "").strip()
if value:
notes.append(f"{key} aus Umgebung")
return value
return default
ollama_url = pick("OLLAMA_URL", os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434"))
ollama_bin = pick("OLLAMA_BIN", "ollama")
base_model = pick("BASE_MODEL", "")
expert_model = pick("EXPERT_MODEL", "jr-sql-expert")
extra_models = pick("EXTRA_MODELS", "")
modelfile_raw = ""
modelfile_source = ""
if explicit_modelfile and str(explicit_modelfile).strip():
modelfile_path = explicit_modelfile.resolve()
modelfile_source = "GUI"
elif env_data.get("MODELFILE_PATH", "").strip():
modelfile_raw = env_data.get("MODELFILE_PATH", "").strip()
modelfile_path = resolve_path_from_value(root, modelfile_raw, default_modelfile_path)
modelfile_source = ".env"
elif env_example_data.get("MODELFILE_PATH", "").strip():
modelfile_raw = env_example_data.get("MODELFILE_PATH", "").strip()
modelfile_path = resolve_path_from_value(root, modelfile_raw, default_modelfile_path)
modelfile_source = ".env.example"
elif os.environ.get("MODELFILE_PATH", "").strip():
modelfile_raw = os.environ.get("MODELFILE_PATH", "").strip()
modelfile_path = resolve_path_from_value(root, modelfile_raw, default_modelfile_path)
modelfile_source = "Umgebung"
else:
modelfile_path = default_modelfile_path
if modelfile_source:
notes.append(f"MODELFILE_PATH aus {modelfile_source}")
if not base_model:
modelfile_base, modelfile_note = parse_modelfile_for_base(modelfile_path)
if modelfile_base:
base_model = modelfile_base
notes.append(f"BASE_MODEL aus {modelfile_note}")
elif modelfile_note:
notes.append(modelfile_note)
inference_model = (
current_inference_model.strip()
or pick("DEFAULT_INFERENCE_MODEL", "")
or os.environ.get("OLLAMA_MODEL", "").strip()
or (f"{expert_model}:latest" if expert_model else "jr-sql-expert:latest")
)
return ResolvedDefaults(
ollama_url=ollama_url,
ollama_bin=ollama_bin,
base_model=base_model,
expert_model=expert_model,
extra_models=extra_models,
inference_model=inference_model,
modelfile_path=modelfile_path,
source_notes=notes,
)
@dataclass
class AppConfig:
project_root: Path
env_path: Path
modelfile_path: Path
ollama_url: str
ollama_bin: str
base_model: str
expert_model: str
extra_models: str
default_inference_model: str
@classmethod
def load(cls, root: Path) -> "AppConfig":
defaults = resolve_defaults(root)
return cls(
project_root=root,
env_path=root / ".env",
modelfile_path=defaults.modelfile_path,
ollama_url=defaults.ollama_url,
ollama_bin=defaults.ollama_bin,
base_model=defaults.base_model,
expert_model=defaults.expert_model,
extra_models=defaults.extra_models,
default_inference_model=defaults.inference_model,
)
def extra_model_list(self) -> List[str]:
return [item for item in shlex.split(self.extra_models) if item.strip()]
# -----------------------------
# Workers
# -----------------------------
@dataclass
class GenerateParams:
base_url: str
model: str
prompt: str
stream: bool = True
keep_alive: str = "30m"
class GenerateWorker(QThread):
chunk = Signal(str)
done = Signal(str)
error = Signal(str)
def __init__(self, params: GenerateParams):
super().__init__()
self.params = params
def run(self) -> None:
try:
url = self.params.base_url.rstrip("/") + "/api/generate"
payload = {
"model": self.params.model,
"prompt": self.params.prompt,
"stream": self.params.stream,
"keep_alive": self.params.keep_alive,
}
with requests.post(url, json=payload, stream=self.params.stream, timeout=(5, 3600)) as r:
r.raise_for_status()
if not self.params.stream:
data = r.json()
self.done.emit(data.get("response", ""))
return
full: list[str] = []
for line in r.iter_lines(decode_unicode=True):
if not line:
continue
obj = json.loads(line)
part = obj.get("response", "")
if part:
full.append(part)
self.chunk.emit(part)
if obj.get("done", False):
break
self.done.emit("".join(full))
except Exception as e:
self.error.emit(human_error(e))
@dataclass
class CommandStep:
title: str
argv: list[str]
cwd: Optional[str] = None
env: Optional[dict[str, str]] = None
class CommandWorker(QThread):
log = Signal(str)
done = Signal()
error = Signal(str)
def __init__(self, steps: List[CommandStep], cleanup: Optional[Callable[[], None]] = None):
super().__init__()
self.steps = steps
self.cleanup = cleanup
def _emit_line(self, text: str) -> None:
cleaned = strip_ansi(text.rstrip("\n\r"))
if cleaned:
self.log.emit(cleaned)
def run(self) -> None:
try:
for step in self.steps:
self._emit_line(f"==> {step.title}")
proc = subprocess.Popen(
step.argv,
cwd=step.cwd,
env=step.env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
assert proc.stdout is not None
for line in proc.stdout:
self._emit_line(line)
rc = proc.wait()
if rc != 0:
raise RuntimeError(f"Command failed ({rc}): {' '.join(step.argv)}")
self.done.emit()
except Exception as e:
self.error.emit(human_error(e))
finally:
if self.cleanup is not None:
try:
self.cleanup()
except Exception:
pass
# -----------------------------
# Main Window
# -----------------------------
class MainWindow(QMainWindow):
def __init__(self) -> None:
super().__init__()
self.setWindowTitle("JR SQL AI GUI (local Ollama)")
self._gen_worker: Optional[GenerateWorker] = None
self._cmd_worker: Optional[CommandWorker] = None
self._temp_modelfile: Optional[Path] = None
self._raw_markdown = ""
self._config_dirty = False
self._suspend_dirty_tracking = False
self.project_root = detect_project_root(Path(__file__).resolve().parent)
self.config = AppConfig.load(self.project_root)
self._render_timer = QTimer(self)
self._render_timer.setInterval(200)
self._render_timer.timeout.connect(self._render_markdown_throttled)
self._build_menu_bar()
self._apply_dark_menu_style()
root = QWidget()
self.setCentralWidget(root)
layout = QVBoxLayout(root)
config_group = QGroupBox("Konfiguration")
config_layout = QGridLayout(config_group)
layout.addWidget(config_group)
row = 0
config_layout.addWidget(QLabel("Projekt-Root:"), row, 0)
self.project_root_edit = QLineEdit(str(self.config.project_root))
config_layout.addWidget(self.project_root_edit, row, 1, 1, 3)
self.btn_browse_root = QPushButton("Wählen")
self.btn_browse_root.clicked.connect(self.on_browse_root)
config_layout.addWidget(self.btn_browse_root, row, 4)
row += 1
config_layout.addWidget(QLabel("Ollama URL:"), row, 0)
self.base_url = QLineEdit(self.config.ollama_url)
config_layout.addWidget(self.base_url, row, 1)
config_layout.addWidget(QLabel("Ollama Binärdatei:"), row, 2)
self.ollama_bin = QLineEdit(self.config.ollama_bin)
config_layout.addWidget(self.ollama_bin, row, 3)
self.btn_reload_config = QPushButton("Neu laden")
self.btn_reload_config.clicked.connect(self.reload_config_from_disk)
config_layout.addWidget(self.btn_reload_config, row, 4)
row += 1
config_layout.addWidget(QLabel("Base Model:"), row, 0)
self.base_model_edit = QLineEdit(self.config.base_model)
self.base_model_edit.setPlaceholderText("z. B. qwen2.5-coder:14b")
config_layout.addWidget(self.base_model_edit, row, 1)
config_layout.addWidget(QLabel("Expert Model:"), row, 2)
self.expert_model_edit = QLineEdit(self.config.expert_model)
config_layout.addWidget(self.expert_model_edit, row, 3)
self.btn_save_env = QPushButton(".env speichern")
self.btn_save_env.clicked.connect(self.save_env)
config_layout.addWidget(self.btn_save_env, row, 4)
row += 1
config_layout.addWidget(QLabel("Extra Models:"), row, 0)
self.extra_models_edit = QLineEdit(self.config.extra_models)
config_layout.addWidget(self.extra_models_edit, row, 1)
config_layout.addWidget(QLabel("Modelfile:"), row, 2)
self.modelfile_edit = QLineEdit(str(self.config.modelfile_path))
config_layout.addWidget(self.modelfile_edit, row, 3)
self.btn_browse_modelfile = QPushButton("Datei")
self.btn_browse_modelfile.clicked.connect(self.on_browse_modelfile)
config_layout.addWidget(self.btn_browse_modelfile, row, 4)
row += 1
self.lbl_defaults_hint = QLabel("")
self.lbl_defaults_hint.setWordWrap(True)
config_layout.addWidget(self.lbl_defaults_hint, row, 0, 1, 5)
row += 1
self.lbl_config_state = QLabel("")
self.lbl_config_state.setWordWrap(True)
config_layout.addWidget(self.lbl_config_state, row, 0, 1, 5)
top = QHBoxLayout()
layout.addLayout(top)
top.addWidget(QLabel("Antwortmodell:"))
self.model = QComboBox()
self.model.setEditable(True)
self.model.addItem(self.config.default_inference_model)
self.model.setCurrentText(self.config.default_inference_model)
self.model.setMinimumWidth(280)
top.addWidget(self.model, 2)
self.btn_refresh_models = QPushButton("Modelle laden")
self.btn_refresh_models.clicked.connect(self.refresh_models)
top.addWidget(self.btn_refresh_models)
self.chk_stream = QCheckBox("Streaming")
self.chk_stream.setChecked(True)
top.addWidget(self.chk_stream)
self.chk_warmup = QCheckBox("Warmup nach Build/Update")
self.chk_warmup.setChecked(False)
top.addWidget(self.chk_warmup)
splitter = QSplitter(Qt.Horizontal)
layout.addWidget(splitter, 1)
left = QWidget()
left_l = QVBoxLayout(left)
left_l.addWidget(QLabel("Prompt / Kontext"))
self.prompt = QPlainTextEdit()
self.prompt.setPlaceholderText("Prompt + Kontext hier einfügen …")
self.prompt.setFont(QFont("Monospace", 10))
left_l.addWidget(self.prompt, 1)
send_row = QHBoxLayout()
self.btn_send = QPushButton("An AI senden")
self.btn_send.clicked.connect(self.on_send)
send_row.addWidget(self.btn_send)
self.btn_clear = QPushButton("Leeren")
self.btn_clear.clicked.connect(lambda: self.prompt.setPlainText(""))
send_row.addWidget(self.btn_clear)
left_l.addLayout(send_row)
right = QWidget()
right_l = QVBoxLayout(right)
right_l.addWidget(QLabel("Antwort (Markdown gerendert)"))
self.response_view = QTextBrowser()
self.response_view.setOpenExternalLinks(True)
self.response_view.setFont(QFont("Monospace", 10))
right_l.addWidget(self.response_view, 1)
self.response_raw = QPlainTextEdit()
self.response_raw.setReadOnly(True)
self.response_raw.setFont(QFont("Monospace", 10))
self.response_raw.setPlaceholderText("Raw Antwort (für Copy/Debug).")
self.response_raw.setMaximumHeight(140)
right_l.addWidget(self.response_raw)
right_l.addWidget(QLabel("Wartungs-/Kommando-Log"))
self.command_log = QPlainTextEdit()
self.command_log.setReadOnly(True)
self.command_log.setFont(QFont("Monospace", 10))
self.command_log.setMaximumHeight(180)
right_l.addWidget(self.command_log)
right_btn_row = QHBoxLayout()
self.btn_copy = QPushButton("Antwort kopieren")
self.btn_copy.clicked.connect(self.copy_response)
right_btn_row.addWidget(self.btn_copy)
self.btn_copy_sql = QPushButton("Copy SQL only")
self.btn_copy_sql.clicked.connect(self.copy_sql_only)
right_btn_row.addWidget(self.btn_copy_sql)
self.btn_update_base = QPushButton("Basismodell aktualisieren")
self.btn_update_base.clicked.connect(self.on_update_base_model)
right_btn_row.addWidget(self.btn_update_base)
self.btn_rebuild_expert = QPushButton("Expert neu bauen")
self.btn_rebuild_expert.clicked.connect(self.on_rebuild_expert)
right_btn_row.addWidget(self.btn_rebuild_expert)
self.btn_update_all = QPushButton("Alles aktualisieren")
self.btn_update_all.clicked.connect(self.on_update_all)
right_btn_row.addWidget(self.btn_update_all)
self.btn_close = QPushButton("Beenden")
self.btn_close.clicked.connect(self.on_close_requested)
right_btn_row.addWidget(self.btn_close)
right_l.addLayout(right_btn_row)
splitter.addWidget(left)
splitter.addWidget(right)
splitter.setSizes([520, 840])
self.status = QStatusBar()
self.setStatusBar(self.status)
self.status.showMessage("Bereit.")
self._connect_config_change_tracking()
self.apply_default_hints()
self._set_config_dirty(False)
QTimer.singleShot(300, self.refresh_models)
def _connect_config_change_tracking(self) -> None:
fields = [
self.project_root_edit,
self.base_url,
self.ollama_bin,
self.base_model_edit,
self.expert_model_edit,
self.extra_models_edit,
self.modelfile_edit,
]
for field in fields:
field.textEdited.connect(self._on_config_field_edited)
self.model.lineEdit().textEdited.connect(self._on_config_field_edited)
self.model.currentTextChanged.connect(self._on_inference_model_changed)
def _set_config_dirty(self, dirty: bool, reason: str = "") -> None:
if self._suspend_dirty_tracking:
return
self._config_dirty = dirty
if dirty:
self.lbl_config_state.setText(
"Konfiguration geändert und noch nicht gespeichert. Klicke auf '.env speichern', um die Änderungen dauerhaft zu übernehmen."
)
self.lbl_config_state.setStyleSheet("color: #f6c177; font-weight: 600;")
self.btn_save_env.setText(".env speichern *")
self.setWindowTitle("JR SQL AI GUI (local Ollama) *")
if reason:
self.status.showMessage(reason, 5000)
else:
self.lbl_config_state.setText("Konfiguration ist gespeichert bzw. entspricht dem geladenen Stand.")
self.lbl_config_state.setStyleSheet("color: #9ccfd8;")
self.btn_save_env.setText(".env speichern")
self.setWindowTitle("JR SQL AI GUI (local Ollama)")
def _on_config_field_edited(self, _text: str) -> None:
self._set_config_dirty(True, "Ungespeicherte Konfigurationsänderungen vorhanden.")
def _on_inference_model_changed(self, text: str) -> None:
if self._suspend_dirty_tracking:
return
if text.strip():
self._set_config_dirty(True, "Antwortmodell geändert. Speichere die .env, wenn dies der neue Standard sein soll.")
def _apply_dark_menu_style(self) -> None:
menu_bar = self.menuBar()
menu_bar.setNativeMenuBar(False)
app = QApplication.instance()
if app is not None:
current = app.styleSheet() or ""
if MENU_STYLE_SHEET not in current:
app.setStyleSheet((current + "\n" + MENU_STYLE_SHEET).strip())
menu_bar.update()
def _build_menu_bar(self) -> None:
menu_bar = self.menuBar()
menu_bar.setNativeMenuBar(False)
file_menu = menu_bar.addMenu("&Datei")
act_reload = QAction("Konfiguration neu laden", self)
act_reload.setShortcut(QKeySequence.Refresh)
act_reload.triggered.connect(self.reload_config_from_disk)
file_menu.addAction(act_reload)
act_save_env = QAction(".env speichern", self)
act_save_env.setShortcut(QKeySequence.Save)
act_save_env.triggered.connect(self.save_env)
file_menu.addAction(act_save_env)
file_menu.addSeparator()
act_exit = QAction("Beenden", self)
act_exit.setShortcut(QKeySequence.Quit)
act_exit.triggered.connect(self.on_close_requested)
file_menu.addAction(act_exit)
actions_menu = menu_bar.addMenu("&Aktionen")
act_refresh_models = QAction("Modelle laden", self)
act_refresh_models.triggered.connect(self.refresh_models)
actions_menu.addAction(act_refresh_models)
actions_menu.addSeparator()
act_update_base = QAction("Basismodell aktualisieren", self)
act_update_base.triggered.connect(self.on_update_base_model)
actions_menu.addAction(act_update_base)
act_rebuild_expert = QAction("Expert neu bauen", self)
act_rebuild_expert.triggered.connect(self.on_rebuild_expert)
actions_menu.addAction(act_rebuild_expert)
act_update_all = QAction("Alles aktualisieren", self)
act_update_all.triggered.connect(self.on_update_all)
actions_menu.addAction(act_update_all)
help_menu = menu_bar.addMenu("&Hilfe")
act_about = QAction("Über", self)
act_about.triggered.connect(self.show_about_dialog)
help_menu.addAction(act_about)
# ----------- config helpers -----------
def current_config(self) -> AppConfig:
root = Path(self.project_root_edit.text().strip() or ".").resolve()
return AppConfig(
project_root=root,
env_path=root / ".env",
modelfile_path=Path(self.modelfile_edit.text().strip() or (root / "Modelfile")).resolve(),
ollama_url=self.base_url.text().strip() or "http://127.0.0.1:11434",
ollama_bin=self.ollama_bin.text().strip() or "ollama",
base_model=self.base_model_edit.text().strip(),
expert_model=self.expert_model_edit.text().strip() or "jr-sql-expert",
extra_models=self.extra_models_edit.text().strip(),
default_inference_model=self.model.currentText().strip() or "jr-sql-expert:latest",
)
def apply_default_hints(self) -> None:
cfg = self.current_config()
defaults = resolve_defaults(cfg.project_root, cfg.modelfile_path, cfg.default_inference_model)
self.lbl_defaults_hint.setText(
"Vorbelegung: " + (", ".join(defaults.source_notes) if defaults.source_notes else "keine Dateiwerte gefunden")
)
def autofill_missing_fields_from_disk(self) -> None:
cfg = self.current_config()
defaults = resolve_defaults(cfg.project_root, cfg.modelfile_path, cfg.default_inference_model)
self._suspend_dirty_tracking = True
try:
if not self.base_url.text().strip() and defaults.ollama_url:
self.base_url.setText(defaults.ollama_url)
if not self.ollama_bin.text().strip() and defaults.ollama_bin:
self.ollama_bin.setText(defaults.ollama_bin)
if not self.base_model_edit.text().strip() and defaults.base_model:
self.base_model_edit.setText(defaults.base_model)
if not self.expert_model_edit.text().strip() and defaults.expert_model:
self.expert_model_edit.setText(defaults.expert_model)
if not self.extra_models_edit.text().strip() and defaults.extra_models:
self.extra_models_edit.setText(defaults.extra_models)
if not self.modelfile_edit.text().strip() and defaults.modelfile_path:
self.modelfile_edit.setText(str(defaults.modelfile_path))
if not self.model.currentText().strip() and defaults.inference_model:
self.model.setCurrentText(defaults.inference_model)
finally:
self._suspend_dirty_tracking = False
self.apply_default_hints()
def validate_runtime(self, cfg: AppConfig) -> Optional[str]:
if shutil.which(cfg.ollama_bin) is None:
return f"Ollama-Binärdatei nicht gefunden: {cfg.ollama_bin}"
return None
def validate_rebuild(self, cfg: AppConfig) -> Optional[str]:
if not cfg.base_model:
return (
"BASE_MODEL konnte nicht vorbelegt werden. Bitte setze BASE_MODEL in .env, .env.example "
"oder trage es im GUI ein. Falls dein Modelfile nur FROM ${BASE_MODEL} nutzt, braucht die GUI "
"zusätzlich diesen Wert."
)
if not cfg.expert_model:
return "EXPERT_MODEL darf nicht leer sein."
if not cfg.modelfile_path.is_file():
return f"Modelfile nicht gefunden: {cfg.modelfile_path}"
return None
def validate_update_base(self, cfg: AppConfig) -> Optional[str]:
if not cfg.base_model:
return (
"BASE_MODEL konnte nicht vorbelegt werden. Bitte setze BASE_MODEL in .env, .env.example "
"oder trage es im GUI ein."
)
return None
def on_browse_root(self) -> None:
path = QFileDialog.getExistingDirectory(self, "Projekt-Root wählen", self.project_root_edit.text().strip() or str(Path.cwd()))
if not path:
return
self.project_root_edit.setText(path)
self.reload_config_from_disk()
def on_browse_modelfile(self) -> None:
path, _ = QFileDialog.getOpenFileName(self, "Modelfile wählen", self.project_root_edit.text().strip() or str(Path.cwd()))
if path:
self.modelfile_edit.setText(path)
self.autofill_missing_fields_from_disk()
self._set_config_dirty(True, "Modelfile-Pfad geändert. Speichere die .env, um den neuen Pfad dauerhaft zu übernehmen.")
def reload_config_from_disk(self) -> None:
root = Path(self.project_root_edit.text().strip() or self.project_root).resolve()
self.config = AppConfig.load(root)
self._suspend_dirty_tracking = True
try:
self.project_root_edit.setText(str(self.config.project_root))
self.base_url.setText(self.config.ollama_url)
self.ollama_bin.setText(self.config.ollama_bin)
self.base_model_edit.setText(self.config.base_model)
self.expert_model_edit.setText(self.config.expert_model)
self.extra_models_edit.setText(self.config.extra_models)
self.modelfile_edit.setText(str(self.config.modelfile_path))
self.model.setCurrentText(self.config.default_inference_model)
finally:
self._suspend_dirty_tracking = False
self.apply_default_hints()
self._set_config_dirty(False)
self.status.showMessage("Konfiguration neu geladen.", 2500)
self.refresh_models()
def save_env(self) -> None:
cfg = self.current_config()
cfg.project_root.mkdir(parents=True, exist_ok=True)
write_env_file(
cfg.env_path,
{
"OLLAMA_URL": cfg.ollama_url,
"OLLAMA_BIN": cfg.ollama_bin,
"BASE_MODEL": cfg.base_model,
"EXPERT_MODEL": cfg.expert_model,
"EXTRA_MODELS": cfg.extra_models,
"MODELFILE_PATH": str(cfg.modelfile_path),
"DEFAULT_INFERENCE_MODEL": cfg.default_inference_model,
},
)
self.apply_default_hints()
self._set_config_dirty(False)
self.status.showMessage(f".env gespeichert: {cfg.env_path}", 3500)
# ----------- UI helpers -----------
def ui_busy(self, busy: bool) -> None:
widgets = [
self.btn_send,
self.btn_refresh_models,
self.btn_copy_sql,
self.btn_update_base,
self.btn_rebuild_expert,
self.btn_update_all,
self.btn_browse_root,
self.btn_reload_config,
self.btn_save_env,
self.btn_browse_modelfile,
self.btn_clear,
self.btn_close,
]
for w in widgets:
w.setEnabled(not busy)
for w in [
self.prompt,
self.base_url,
self.model,
self.chk_stream,
self.chk_warmup,
self.project_root_edit,
self.base_model_edit,
self.expert_model_edit,
self.extra_models_edit,
self.modelfile_edit,
self.ollama_bin,
]:
w.setEnabled(not busy)
def msg_error(self, title: str, text: str) -> None:
QMessageBox.critical(self, title, strip_ansi(text))
def msg_info(self, title: str, text: str) -> None:
QMessageBox.information(self, title, strip_ansi(text))
def append_command_log(self, text: str) -> None:
self.command_log.appendPlainText(strip_ansi(text))
sb = self.command_log.verticalScrollBar()
sb.setValue(sb.maximum())
# ----------- Model list -----------
def refresh_models(self) -> None:
base = self.base_url.text().strip().rstrip("/")
if not base:
return
try:
r = requests.get(base + "/api/tags", timeout=(3, 20))
r.raise_for_status()
data = r.json()
models = [m.get("name") for m in data.get("models", []) if m.get("name")]
current = self.model.currentText().strip()
self.model.clear()
if models:
self.model.addItems(models)
if current in models:
self.model.setCurrentText(current)
elif self.config.default_inference_model in models:
self.model.setCurrentText(self.config.default_inference_model)
else:
self.model.setCurrentIndex(0)
self.status.showMessage(f"{len(models)} Modelle geladen.", 2500)
else:
if current:
self.model.addItem(current)
self.model.setCurrentText(current)
self.status.showMessage("Keine Modelle gefunden (api/tags leer).", 5000)
except Exception as e:
self.status.showMessage(f"Model-Liste konnte nicht geladen werden: {human_error(e)}", 8000)
# ----------- Inference -----------
def on_send(self) -> None:
prompt = self.prompt.toPlainText().strip()
if not prompt:
self.msg_info("Hinweis", "Bitte erst einen Prompt/Kontext eingeben.")
return
base = self.base_url.text().strip()
model = self.model.currentText().strip()
if not base or not model:
self.msg_info("Hinweis", "Bitte Ollama URL und Antwortmodell setzen.")
return
self._raw_markdown = ""
self.response_raw.setPlainText("")
self.response_view.setMarkdown("")
self.status.showMessage("Sende Anfrage …")
self.ui_busy(True)
self._gen_worker = GenerateWorker(
GenerateParams(base_url=base, model=model, prompt=prompt, stream=self.chk_stream.isChecked())
)
self._gen_worker.chunk.connect(self._on_chunk)
self._gen_worker.done.connect(self._on_done)
self._gen_worker.error.connect(self._on_gen_error)
self._gen_worker.start()
if self.chk_stream.isChecked():
self._render_timer.start()
def _on_chunk(self, s: str) -> None:
self._raw_markdown += s
self.response_raw.setPlainText(self._raw_markdown)
sb = self.response_raw.verticalScrollBar()
sb.setValue(sb.maximum())
def _render_markdown_throttled(self) -> None:
if self._raw_markdown:
self.response_view.setMarkdown(self._raw_markdown)
def _on_done(self, full: str) -> None:
self._render_timer.stop()
if not self.chk_stream.isChecked():
self._raw_markdown = full
self.response_raw.setPlainText(full)
self.response_view.setMarkdown(self._raw_markdown)
self.status.showMessage("Fertig.", 2500)
self.ui_busy(False)
def _on_gen_error(self, err: str) -> None:
self._render_timer.stop()
self.ui_busy(False)
self.status.showMessage("Fehler.", 5000)
self.msg_error("Ollama Fehler", err)
# ----------- Clipboard -----------
def copy_response(self) -> None:
QApplication.clipboard().setText(self._raw_markdown or self.response_raw.toPlainText())
self.status.showMessage("Antwort (Markdown) in Clipboard kopiert.", 2500)
def copy_sql_only(self) -> None:
md = self._raw_markdown or self.response_raw.toPlainText()
blocks = extract_sql_blocks(md)
sql_text = build_sql_only_text(blocks)
if not sql_text:
self.msg_info("Kein SQL gefunden", "Ich habe in der Antwort keine SQL-Codeblöcke gefunden.")
return
QApplication.clipboard().setText(sql_text)
self.status.showMessage(f"SQL kopiert ({len(blocks)} Block/Blöcke).", 3000)
# ----------- Command orchestration -----------
def start_command_worker(self, steps: List[CommandStep], title: str, cleanup: Optional[Callable[[], None]] = None) -> None:
self.command_log.setPlainText("")
self.ui_busy(True)
self.status.showMessage(title)
self._cmd_worker = CommandWorker(steps, cleanup=cleanup)
self._cmd_worker.log.connect(self.append_command_log)
self._cmd_worker.done.connect(self._on_cmd_done)
self._cmd_worker.error.connect(self._on_cmd_error)
self._cmd_worker.start()
def build_warmup_step(self, cfg: AppConfig) -> Optional[CommandStep]:
sqlai = cfg.project_root / "bin" / "sqlai"
if self.chk_warmup.isChecked() and sqlai.is_file() and os.access(sqlai, os.X_OK):
return CommandStep(
title="Warmup",
argv=[str(sqlai), "ask", "--text", "Warmup: reply with exactly 'OK'.", "--no-metrics"],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
return None
def on_update_base_model(self) -> None:
self.autofill_missing_fields_from_disk()
cfg = self.current_config()
msg = self.validate_runtime(cfg) or self.validate_update_base(cfg)
if msg:
self.msg_error("Konfiguration unvollständig", msg)
return
steps = [
CommandStep(
title=f"Base Model pull: {cfg.base_model}",
argv=[cfg.ollama_bin, "pull", cfg.base_model],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
]
for model in cfg.extra_model_list():
steps.append(
CommandStep(
title=f"Extra Model pull: {model}",
argv=[cfg.ollama_bin, "pull", model],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
)
warmup = self.build_warmup_step(cfg)
if warmup:
steps.append(warmup)
self.start_command_worker(steps, "Basismodell wird aktualisiert …")
def _prepare_temp_modelfile(self, cfg: AppConfig) -> Path:
text = cfg.modelfile_path.read_text(encoding="utf-8")
text = text.replace("${BASE_MODEL}", cfg.base_model)
text = text.replace("$BASE_MODEL", cfg.base_model)
fd, path = tempfile.mkstemp(prefix="jr-sql-expert-", suffix=".Modelfile")
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(text)
temp_path = Path(path)
self._temp_modelfile = temp_path
return temp_path
def _cleanup_temp_modelfile(self) -> None:
temp = self._temp_modelfile
self._temp_modelfile = None
if temp is not None:
try:
temp.unlink(missing_ok=True)
except Exception:
pass
def on_rebuild_expert(self) -> None:
self.autofill_missing_fields_from_disk()
cfg = self.current_config()
msg = self.validate_runtime(cfg) or self.validate_rebuild(cfg)
if msg:
self.msg_error("Konfiguration unvollständig", msg)
return
temp_modelfile = self._prepare_temp_modelfile(cfg)
steps = [
CommandStep(
title=f"Expert Model build: {cfg.expert_model}",
argv=[cfg.ollama_bin, "create", cfg.expert_model, "-f", str(temp_modelfile)],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
]
warmup = self.build_warmup_step(cfg)
if warmup:
steps.append(warmup)
self.start_command_worker(steps, "Expert-Modell wird neu gebaut …", cleanup=self._cleanup_temp_modelfile)
def on_update_all(self) -> None:
self.autofill_missing_fields_from_disk()
cfg = self.current_config()
msg = self.validate_runtime(cfg) or self.validate_rebuild(cfg)
if msg:
self.msg_error("Konfiguration unvollständig", msg)
return
temp_modelfile = self._prepare_temp_modelfile(cfg)
steps = [
CommandStep(
title=f"Base Model pull: {cfg.base_model}",
argv=[cfg.ollama_bin, "pull", cfg.base_model],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
]
for model in cfg.extra_model_list():
steps.append(
CommandStep(
title=f"Extra Model pull: {model}",
argv=[cfg.ollama_bin, "pull", model],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
)
steps.append(
CommandStep(
title=f"Expert Model build: {cfg.expert_model}",
argv=[cfg.ollama_bin, "create", cfg.expert_model, "-f", str(temp_modelfile)],
cwd=str(cfg.project_root),
env=os.environ.copy(),
)
)
warmup = self.build_warmup_step(cfg)
if warmup:
steps.append(warmup)
self.start_command_worker(steps, "Update + Rebuild läuft …", cleanup=self._cleanup_temp_modelfile)
def _on_cmd_done(self) -> None:
self._cleanup_temp_modelfile()
self.ui_busy(False)
self.refresh_models()
self.status.showMessage("Wartungsaktion abgeschlossen.", 4000)
def _on_cmd_error(self, err: str) -> None:
self._cleanup_temp_modelfile()
self.ui_busy(False)
self.status.showMessage("Wartungsaktion fehlgeschlagen.", 5000)
log_text = self.command_log.toPlainText()
err_clean = strip_ansi(err)
if looks_like_registry_pull_error(log_text) or looks_like_registry_pull_error(err_clean):
self.msg_error(
"Modell konnte nicht gezogen werden",
"Das gewählte Modell ist offenbar kein remote pullbares Registry-Modell. "
"Für dein lokales Expert-Modell nutze 'Expert neu bauen' oder 'Alles aktualisieren'.",
)
return
if "BASE_MODEL" in err_clean and "not set" in err_clean.lower():
self.msg_error(
"BASE_MODEL fehlt",
"Setze BASE_MODEL in der .env, .env.example oder direkt im GUI und speichere die Konfiguration.",
)
return
self.msg_error("Ollama Wartung fehlgeschlagen", err_clean)
def show_about_dialog(self) -> None:
self.msg_info(
"Über JR SQL AI GUI",
"JR SQL AI GUI\n\n"
"- lokale Ollama API für Antworten\n"
"- lokales ollama-Binary für Modellwartung\n"
"- getrennte Steuerung für Base-/Expert-Modell\n"
"- Menüleiste mit Datei- und Wartungsaktionen",
)
def on_close_requested(self) -> None:
self.close()
def closeEvent(self, event) -> None: # type: ignore[override]
if self._cmd_worker is not None and self._cmd_worker.isRunning():
self.msg_info("Bitte warten", "Es läuft gerade eine Wartungsaktion. Schließe das Fenster danach erneut.")
event.ignore()
return
if self._config_dirty:
answer = QMessageBox.question(
self,
"Ungespeicherte Konfiguration",
"Es gibt ungespeicherte Konfigurationsänderungen. Möchtest du sie vor dem Beenden in die .env schreiben?",
QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel,
QMessageBox.Yes,
)
if answer == QMessageBox.Cancel:
event.ignore()
return
if answer == QMessageBox.Yes:
self.save_env()
event.accept()
def main() -> int:
app = QApplication(sys.argv)
app.setApplicationName("JR SQL AI GUI")
w = MainWindow()
w.resize(1500, 940)
w.show()
return app.exec()
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink