#!/usr/bin/env python3 """ JR SQL AI GUI (local Ollama) - Uses the local `ollama` CLI for model/runtime management. - Uses the local Ollama HTTP API for fast streaming generation. - Friendly to Arch Linux + Hyprland / Wayland. """ from __future__ import annotations import json import os import re import subprocess import sys import time from dataclasses import dataclass from pathlib import Path from typing import Callable, Iterable, List, Optional import requests from PySide6.QtCore import Qt, QThread, Signal, QTimer from PySide6.QtGui import QFont from PySide6.QtWidgets import ( QApplication, QCheckBox, QComboBox, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QSplitter, QStatusBar, QTextBrowser, QVBoxLayout, QWidget, ) DEFAULT_OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434") DEFAULT_OLLAMA_BIN = os.environ.get("OLLAMA_BIN", "ollama") DEFAULT_MODEL = os.environ.get("OLLAMA_MODEL", "jr-sql-expert:latest") DEFAULT_KEEP_ALIVE = os.environ.get("OLLAMA_KEEP_ALIVE", "10m") DEFAULT_PROJECT_ROOT = Path(os.environ.get("JR_SQL_AI_ROOT", Path(__file__).resolve().parent)) DEFAULT_MODEFILE = Path(os.environ.get("MODEFILE", DEFAULT_PROJECT_ROOT / "Modelfile")) DEFAULT_EXPERT_MODEL = os.environ.get("EXPERT_MODEL", "jr-sql-expert") DEFAULT_BASE_MODEL = os.environ.get("BASE_MODEL", "") def load_env_file(env_file: Path) -> None: if not env_file.is_file(): return for raw_line in env_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) load_env_file(DEFAULT_PROJECT_ROOT / ".env") DEFAULT_EXPERT_MODEL = os.environ.get("EXPERT_MODEL", DEFAULT_EXPERT_MODEL) DEFAULT_BASE_MODEL = os.environ.get("BASE_MODEL", DEFAULT_BASE_MODEL) DEFAULT_OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", DEFAULT_OLLAMA_BASE_URL) DEFAULT_OLLAMA_BIN = os.environ.get("OLLAMA_BIN", DEFAULT_OLLAMA_BIN) DEFAULT_MODEL = os.environ.get("OLLAMA_MODEL", DEFAULT_MODEL) 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) def human_error(e: Exception) -> str: return f"{type(e).__name__}: {e}" def extract_sql_blocks(markdown_text: str) -> List[str]: blocks: List[str] = [] for match in FENCE_RE.finditer(markdown_text): lang = (match.group(1) or "").strip().lower() body = (match.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 cli_env(base_url: str) -> dict[str, str]: env = os.environ.copy() env["OLLAMA_HOST"] = base_url.strip() return env def run_cli( ollama_bin: str, base_url: str, args: List[str], timeout: int = 60, ) -> subprocess.CompletedProcess[str]: return subprocess.run( [ollama_bin, *args], env=cli_env(base_url), text=True, capture_output=True, timeout=timeout, check=False, ) def list_models_via_cli(ollama_bin: str, base_url: str) -> List[str]: proc = run_cli(ollama_bin, base_url, ["ls"], timeout=20) if proc.returncode != 0: raise RuntimeError((proc.stderr or proc.stdout or "ollama ls failed").strip()) models: List[str] = [] for raw in (proc.stdout or "").splitlines(): line = raw.strip() if not line or line.lower().startswith("name ") or line.lower() == "name": continue first = line.split()[0] if first and first != "NAME": models.append(first) if not models: raise RuntimeError("Keine Modelle aus 'ollama ls' geparst.") return models def check_api_ready(base_url: str, timeout: float = 2.5) -> bool: try: r = requests.get(base_url.rstrip("/") + "/api/tags", timeout=(timeout, timeout)) r.raise_for_status() data = r.json() return isinstance(data, dict) except Exception: return False def maybe_start_runtime(ollama_bin: str, base_url: str) -> bool: if check_api_ready(base_url): return True service = os.environ.get("OLLAMA_SYSTEMD_SERVICE", "ollama.service") for cmd in (["systemctl", "--user", "start", service], ["systemctl", "start", service]): try: subprocess.run(cmd, capture_output=True, text=True, timeout=15, check=False) for _ in range(15): if check_api_ready(base_url): return True time.sleep(1) except Exception: pass try: subprocess.Popen( [ollama_bin, "serve"], env=cli_env(base_url), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) for _ in range(20): if check_api_ready(base_url): return True time.sleep(1) except Exception: pass return check_api_ready(base_url) def render_modelfile(template_path: Path, base_model: str) -> str: if not template_path.is_file(): raise FileNotFoundError(f"Modelfile nicht gefunden: {template_path}") if not base_model.strip(): raise ValueError("BASE_MODEL ist leer.") return template_path.read_text(encoding="utf-8").replace("${BASE_MODEL}", base_model.strip()) @dataclass class GenerateParams: base_url: str model: str prompt: str stream: bool = True keep_alive: str = DEFAULT_KEEP_ALIVE class GenerateWorker(QThread): chunk = Signal(str) done = Signal(str, dict) error = Signal(str) def __init__(self, params: GenerateParams): super().__init__() self.params = params def run(self) -> None: try: session = requests.Session() 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 session.post(url, json=payload, stream=self.params.stream, timeout=(5, 1800)) as response: response.raise_for_status() if not self.params.stream: data = response.json() self.done.emit(data.get("response", ""), data) return full: List[str] = [] last_obj: dict = {} for raw_line in response.iter_lines(decode_unicode=True): if not raw_line: continue try: obj = json.loads(raw_line) except json.JSONDecodeError as exc: raise RuntimeError(f"Ungültiger Streaming-JSON-Chunk: {exc}") from exc last_obj = obj part = obj.get("response", "") or "" if part: full.append(part) self.chunk.emit(part) if obj.get("done", False): break self.done.emit("".join(full), last_obj) except Exception as exc: self.error.emit(human_error(exc)) class CommandWorker(QThread): status = Signal(str) done = Signal(str) error = Signal(str) def __init__( self, program: str, args: List[str], env: dict[str, str], cwd: Optional[str] = None, label: str = "command", ): super().__init__() self.program = program self.args = args self.env = env self.cwd = cwd self.label = label def run(self) -> None: try: proc = subprocess.Popen( [self.program, *self.args], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, cwd=self.cwd, env=self.env, ) assert proc.stdout is not None lines: List[str] = [] for raw_line in proc.stdout: line = raw_line.strip() if line: lines.append(line) self.status.emit(line) rc = proc.wait() joined = "\n".join(lines) if rc != 0: raise RuntimeError(joined or f"{self.label} failed with exit code {rc}") self.done.emit(joined) except Exception as exc: self.error.emit(human_error(exc)) class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("JR SQL AI GUI (local Ollama)") self._gen_worker: Optional[GenerateWorker] = None self._cmd_worker: Optional[CommandWorker] = None self._raw_markdown = "" self._last_metrics: dict = {} self._render_timer = QTimer(self) self._render_timer.setInterval(200) self._render_timer.timeout.connect(self._render_markdown_throttled) root = QWidget() self.setCentralWidget(root) layout = QVBoxLayout(root) top = QHBoxLayout() layout.addLayout(top) top.addWidget(QLabel("Ollama URL:")) self.base_url = QLineEdit(DEFAULT_OLLAMA_BASE_URL) self.base_url.setMinimumWidth(260) top.addWidget(self.base_url, 2) top.addWidget(QLabel("Ollama Bin:")) self.ollama_bin = QLineEdit(DEFAULT_OLLAMA_BIN) self.ollama_bin.setMinimumWidth(140) top.addWidget(self.ollama_bin, 1) top.addWidget(QLabel("Model:")) self.model = QComboBox() self.model.setEditable(True) self.model.addItem(DEFAULT_MODEL) self.model.setCurrentText(DEFAULT_MODEL) self.model.setMinimumWidth(220) top.addWidget(self.model, 1) self.btn_runtime = QPushButton("Runtime prüfen") self.btn_runtime.clicked.connect(self.ensure_runtime) top.addWidget(self.btn_runtime) 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) 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) btn_row = QHBoxLayout() self.btn_send = QPushButton("An AI senden") self.btn_send.clicked.connect(self.on_send) btn_row.addWidget(self.btn_send) self.btn_clear = QPushButton("Leeren") self.btn_clear.clicked.connect(lambda: self.prompt.setPlainText("")) btn_row.addWidget(self.btn_clear) left_l.addLayout(btn_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(180) right_l.addWidget(self.response_raw) 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_model_pull = QPushButton("Modell aktualisieren") self.btn_model_pull.clicked.connect(self.on_pull_model) right_btn_row.addWidget(self.btn_model_pull) 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_close = QPushButton("Close") self.btn_close.clicked.connect(self.on_close) right_btn_row.addWidget(self.btn_close) right_l.addLayout(right_btn_row) splitter.addWidget(left) splitter.addWidget(right) splitter.setSizes([520, 820]) self.status = QStatusBar() self.setStatusBar(self.status) self.status.showMessage("Bereit.") QTimer.singleShot(250, self.ensure_runtime) QTimer.singleShot(600, self.refresh_models) def msg_error(self, title: str, text: str) -> None: QMessageBox.critical(self, title, text) def msg_info(self, title: str, text: str) -> None: QMessageBox.information(self, title, text) def ui_busy(self, busy: bool) -> None: for widget in [ self.btn_send, self.btn_model_pull, self.btn_rebuild_expert, self.btn_refresh_models, self.btn_runtime, self.btn_copy_sql, ]: widget.setEnabled(not busy) self.prompt.setEnabled(not busy) self.base_url.setEnabled(not busy) self.ollama_bin.setEnabled(not busy) self.model.setEnabled(not busy) self.chk_stream.setEnabled(not busy) self.btn_close.setEnabled(True) def current_base_url(self) -> str: return self.base_url.text().strip().rstrip("/") def current_ollama_bin(self) -> str: return self.ollama_bin.text().strip() or DEFAULT_OLLAMA_BIN def ensure_runtime(self) -> None: base = self.current_base_url() ollama_bin = self.current_ollama_bin() if maybe_start_runtime(ollama_bin, base): self.status.showMessage("Lokale Ollama Runtime bereit.", 3000) else: self.status.showMessage("Ollama Runtime nicht erreichbar.", 6000) def refresh_models(self) -> None: base = self.current_base_url() ollama_bin = self.current_ollama_bin() if not base: return try: self.ensure_runtime() models = list_models_via_cli(ollama_bin, base) current = self.model.currentText().strip() self.model.clear() self.model.addItems(models) if current and current in models: self.model.setCurrentText(current) elif DEFAULT_MODEL in models: self.model.setCurrentText(DEFAULT_MODEL) else: self.model.setCurrentIndex(0) self.status.showMessage(f"{len(models)} Modelle geladen (CLI).", 2500) except Exception as exc: self.status.showMessage(f"Modelle konnten nicht geladen werden: {human_error(exc)}", 8000) def on_send(self) -> None: prompt = self.prompt.toPlainText().strip() base = self.current_base_url() model = self.model.currentText().strip() if not prompt: self.msg_info("Hinweis", "Bitte erst einen Prompt/Kontext eingeben.") return if not base or not model: self.msg_info("Hinweis", "Bitte Ollama URL und Modell setzen.") return if not check_api_ready(base): self.ensure_runtime() if not check_api_ready(base): self.msg_error("Ollama nicht erreichbar", "Die lokale Ollama API ist nicht erreichbar.") return self._raw_markdown = "" self._last_metrics = {} self._temp_modelfile: Optional[Path] = None 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(), keep_alive=DEFAULT_KEEP_ALIVE, ) ) 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, chunk: str) -> None: self._raw_markdown += chunk self.response_raw.setPlainText(self._raw_markdown) self.response_raw.verticalScrollBar().setValue(self.response_raw.verticalScrollBar().maximum()) def _render_markdown_throttled(self) -> None: if self._raw_markdown: self.response_view.setMarkdown(self._raw_markdown) def _on_done(self, full: str, meta: dict) -> 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._last_metrics = meta or {} self.ui_busy(False) eval_count = meta.get("eval_count") eval_duration = meta.get("eval_duration") load_duration = meta.get("load_duration") if eval_count and eval_duration: tps = eval_count / max(eval_duration / 1_000_000_000, 1e-9) load_ms = (load_duration or 0) / 1_000_000 self.status.showMessage(f"Fertig. {eval_count} Tokens, {tps:.1f} tok/s, Load {load_ms:.0f} ms", 6000) else: self.status.showMessage("Fertig.", 2500) 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) def copy_response(self) -> None: QApplication.clipboard().setText(self._raw_markdown or self.response_raw.toPlainText()) self.status.showMessage("Antwort (Markdown) in die Zwischenablage 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", "In der Antwort wurden keine SQL-Codeblöcke gefunden.") return QApplication.clipboard().setText(sql_text) self.status.showMessage(f"SQL kopiert ({len(blocks)} Block/Blöcke).", 3000) def run_command(self, args: List[str], label: str, done_message: str, on_done: Optional[Callable[[str], None]] = None) -> None: base = self.current_base_url() ollama_bin = self.current_ollama_bin() if not maybe_start_runtime(ollama_bin, base): self.msg_error("Ollama nicht erreichbar", "Die lokale Ollama Runtime konnte nicht gestartet werden.") return self.ui_busy(True) self.status.showMessage(f"{label} …") self._cmd_worker = CommandWorker( program=ollama_bin, args=args, env=cli_env(base), cwd=str(DEFAULT_PROJECT_ROOT), label=label, ) self._cmd_worker.status.connect(lambda s: self.status.showMessage(f"{label}: {s[-140:]}")) self._cmd_worker.done.connect(lambda out: self._on_cmd_done(done_message, out, on_done)) self._cmd_worker.error.connect(self._on_cmd_error) self._cmd_worker.start() def _on_cmd_done(self, done_message: str, output: str, on_done: Optional[Callable[[str], None]]) -> None: self.ui_busy(False) self.status.showMessage(done_message, 5000) if on_done is not None: on_done(output) def _on_cmd_error(self, err: str) -> None: self.ui_busy(False) if self._temp_modelfile is not None: try: self._temp_modelfile.unlink(missing_ok=True) finally: self._temp_modelfile = None self.msg_error("Ollama Kommando fehlgeschlagen", err) def on_pull_model(self) -> None: model = self.model.currentText().strip() if not model: self.msg_info("Hinweis", "Bitte ein Modell setzen.") return self.run_command(["pull", model], f"pull {model}", "Modell aktualisiert.", lambda _: self.refresh_models()) def on_rebuild_expert(self) -> None: base_model = os.environ.get("BASE_MODEL", DEFAULT_BASE_MODEL).strip() expert_model = os.environ.get("EXPERT_MODEL", DEFAULT_EXPERT_MODEL).strip() if not base_model: self.msg_error("BASE_MODEL fehlt", "Setze BASE_MODEL in deiner Umgebung oder .env, bevor du das Expert-Modell neu baust.") return try: modelfile_text = render_modelfile(DEFAULT_MODEFILE, base_model) except Exception as exc: self.msg_error("Modelfile Fehler", human_error(exc)) return tmp_path = DEFAULT_PROJECT_ROOT / ".tmp.gui.Modelfile" tmp_path.write_text(modelfile_text, encoding="utf-8") self._temp_modelfile = tmp_path def finalize(_: str) -> None: try: tmp_path.unlink(missing_ok=True) finally: self._temp_modelfile = None self.refresh_models() self.run_command( ["create", expert_model, "-f", str(tmp_path)], f"create {expert_model}", f"Expert-Modell {expert_model} neu gebaut.", finalize, ) def on_close(self) -> None: if self._gen_worker and self._gen_worker.isRunning(): reply = QMessageBox.question( self, "Beenden", "Es läuft noch eine Generierung. Trotzdem beenden?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if reply != QMessageBox.Yes: return self.close() def main() -> int: app = QApplication(sys.argv) app.setApplicationName("JR SQL AI GUI") window = MainWindow() window.resize(1360, 820) window.show() return app.exec() if __name__ == "__main__": raise SystemExit(main())