Text-to-SQL / t2s /settings.py
SevdanurGenc's picture
Upload 466 files
a72d1cf verified
Raw History Blame Contribute Delete
8.51 kB
"""Uygulama ayarları: tek bir JSON dosyası (ayarlar/tablo2sql-ayarlar.json).
Dosyada dil, tema, model bağlantıları (API anahtarları dahil, düz metin), senkronize edilen model listeleri,
Ollama hesaplama aygıtı, seçili modeller ve ajan seçenekleri bulunur. Dosyayı başka bir bilgisayardaki uygulama
klasörüne kopyalamak bütün ayarları taşır. Anahtarlar arayüze hiçbir zaman tam olarak geri gönderilmez;
yalnızca maskelenmiş hâli (sk-…F1cA) gösterilir.
"""
from __future__ import annotations
import copy
import json
import os
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
SETTINGS_DIR = ROOT / "ayarlar"
SETTINGS_FILE = SETTINGS_DIR / "tablo2sql-ayarlar.json"
LEGACY_PREFS = ROOT / "outputs" / ".arayuz_ayarlari.json"
PROVIDERS = ("ollama", "openai", "claude")
CLOUD = ("openai", "claude")
KEY_URLS = {"openai": "https://platform.openai.com/api-keys", "claude": "https://console.anthropic.com/settings/keys"}
ENV_KEYS = {"openai": "OPENAI_API_KEY", "claude": "ANTHROPIC_API_KEY"}
RULE = "kural" # model seçimi: "kural" ya da "sağlayıcı|model"
DEFAULTS = {
"surum": 1,
"dil": "tr",
"tema": "light",
"baglantilar": {
"ollama": {"url": "http://localhost:11434", "cihaz": "auto", "modeller": [], "son_senkron": None,
"hata": None},
"openai": {"anahtar": "", "url": "", "modeller": [], "son_senkron": None, "hata": None},
"claude": {"anahtar": "", "url": "", "modeller": [], "son_senkron": None, "hata": None},
},
"secim": {"genel": RULE, "sekmeler": {}},
"ajan": {"kendini_duzelt": True, "deneme": 3, "hibrit": True},
}
def _merge(base: dict, extra: dict) -> dict:
out = copy.deepcopy(base)
for k, v in (extra or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _merge(out[k], v)
else:
out[k] = v
return out
def mask_key(key: str | None) -> str:
"""'sk-proj-abc…F1cA' gibi: önek + son 4 karakter. Tam anahtar arayüze gönderilmez."""
if not key:
return ""
k = key.strip()
if len(k) <= 8:
return "…" + k[-2:]
pre = k.split("-")[0] + "-" if "-" in k[:6] else k[:2]
return f"{pre}…{k[-4:]}"
def now_iso() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%S")
class Settings:
def __init__(self, path: Path | str = SETTINGS_FILE):
self.path = Path(path)
self.data = self._load()
# ------------------------------------------------------------------ dosya
def _load(self) -> dict:
if self.path.exists():
try:
return _merge(DEFAULTS, json.loads(self.path.read_text(encoding="utf-8")))
except Exception: # noqa: BLE001 - bozuk dosya: varsayılanlarla aç, dosyayı ezme
d = copy.deepcopy(DEFAULTS)
d["_bozuk"] = True
return d
d = copy.deepcopy(DEFAULTS)
if LEGACY_PREFS.exists(): # eski arayüz tercihlerini bir kez taşı
try:
old = json.loads(LEGACY_PREFS.read_text(encoding="utf-8"))
d["dil"] = old.get("lang", d["dil"])
d["tema"] = old.get("theme", d["tema"])
if old.get("base_openai") and "api.openai.com" not in old["base_openai"]:
d["baglantilar"]["openai"]["url"] = old["base_openai"]
if old.get("base_ollama"):
d["baglantilar"]["ollama"]["url"] = old["base_ollama"]
prov, mdl = old.get("provider"), (old.get("model") or {}).get(old.get("provider") or "")
if prov in PROVIDERS and mdl:
d["secim"]["genel"] = f"{prov}|{mdl}"
except Exception: # noqa: BLE001
pass
env_ollama = os.getenv("OLLAMA_HOST") or os.getenv("OLLAMA_BASE_URL")
if env_ollama:
d["baglantilar"]["ollama"]["url"] = env_ollama if env_ollama.startswith("http") else "http://" + env_ollama
return d
def save(self):
self.data.pop("_bozuk", None)
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(".tmp")
tmp.write_text(json.dumps(self.data, ensure_ascii=False, indent=1), encoding="utf-8")
os.replace(tmp, self.path)
try:
os.chmod(self.path, 0o600)
except OSError:
pass
def to_bytes(self) -> bytes:
d = copy.deepcopy(self.data)
d.pop("_bozuk", None)
return json.dumps(d, ensure_ascii=False, indent=1).encode("utf-8")
def load_bytes(self, raw: bytes) -> None:
"""İndirilmiş bir ayar dosyasını yükler. Geçersizse ValueError."""
try:
d = json.loads(raw.decode("utf-8-sig"))
except Exception as e: # noqa: BLE001
raise ValueError(f"JSON okunamadı: {e}") from e
if not isinstance(d, dict) or not isinstance(d.get("baglantilar", {}), dict):
raise ValueError("Bu bir Tablo2SQL ayar dosyası değil.")
unknown = set(d.get("baglantilar", {})) - set(PROVIDERS)
if unknown:
raise ValueError(f"Bilinmeyen sağlayıcı: {', '.join(sorted(unknown))}")
self.data = _merge(DEFAULTS, d)
self.save()
# ------------------------------------------------------------------ erişim
def conn(self, prov: str) -> dict:
return self.data["baglantilar"][prov]
def key(self, prov: str) -> str:
"""Kayıtlı anahtar; yoksa ortam değişkeni (.env)."""
if prov not in CLOUD:
return ""
return self.conn(prov).get("anahtar") or os.getenv(ENV_KEYS[prov], "")
def key_source(self, prov: str) -> str:
if self.conn(prov).get("anahtar"):
return "dosya"
return "ortam" if os.getenv(ENV_KEYS.get(prov, "-")) else ""
def base(self, prov: str) -> str | None:
c = self.conn(prov)
if prov == "ollama":
return (c.get("url") or "http://localhost:11434").rstrip("/")
if prov == "openai":
return c.get("url") or os.getenv("OPENAI_BASE_URL") or None
return c.get("url") or None
def is_configured(self, prov: str) -> bool:
return True if prov == "ollama" else bool(self.key(prov))
def models(self, prov: str) -> list[dict]:
return self.conn(prov).get("modeller") or []
def set_models(self, prov: str, models: list[dict], error: str | None = None):
c = self.conn(prov)
if error is None:
c["modeller"] = models
c["son_senkron"] = now_iso()
c["hata"] = error
self.save()
def set_key(self, prov: str, key: str):
self.conn(prov)["anahtar"] = key.strip()
self.save()
def disconnect(self, prov: str):
c = self.conn(prov)
if prov in CLOUD:
c["anahtar"] = ""
c.update({"modeller": [], "son_senkron": None, "hata": None})
# bu sağlayıcıya ait seçimleri kural tabanlıya döndür
sec = self.data["secim"]
if str(sec.get("genel", "")).startswith(prov + "|"):
sec["genel"] = RULE
for k, v in list(sec.get("sekmeler", {}).items()):
if isinstance(v, str) and v.startswith(prov + "|"):
sec["sekmeler"][k] = None
elif isinstance(v, list):
sec["sekmeler"][k] = [x for x in v if not str(x).startswith(prov + "|")]
self.save()
def choices(self) -> list[str]:
"""Seçilebilir modeller: 'kural' + bağlı sağlayıcıların senkronize modelleri."""
out = [RULE]
for p in ("claude", "openai", "ollama"):
if self.is_configured(p) and not (p != "ollama" and not self.key(p)):
out += [f"{p}|{m['id']}" for m in self.models(p)]
return out
def get(self, *path, default=None):
d = self.data
for k in path:
if not isinstance(d, dict) or k not in d:
return default
d = d[k]
return d
def set(self, *path_and_value):
*path, value = path_and_value
d = self.data
for k in path[:-1]:
d = d.setdefault(k, {})
if d.get(path[-1]) != value:
d[path[-1]] = value
self.save()
def split_choice(choice: str | None) -> tuple[str, str | None]:
if not choice or choice == RULE or "|" not in choice:
return RULE, None
p, m = choice.split("|", 1)
return p, m