"""Merkezi konfigurasyon. .env dosyasindan ortam degiskenlerini yukler. Tum diger moduller (ssh_client, agent, vs.) konfigurasyonu buradan ceker; boylece SSH bilgileri / model adi tek noktada yonetilir. """ from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path from dotenv import load_dotenv PROJECT_ROOT = Path(__file__).resolve().parent load_dotenv(PROJECT_ROOT / ".env") def _get(name: str, default: str | None = None, *, required: bool = False) -> str: val = os.getenv(name, default) if required and (val is None or val == ""): raise RuntimeError( f"Eksik ortam degiskeni: {name}. .env dosyasini kontrol et " f"(.env.example'dan kopyalayabilirsin)." ) return val or "" @dataclass(frozen=True) class SSHConfig: host: str port: int username: str key_path: Path key_passphrase: str | None password: str | None command_timeout: int @classmethod def from_env(cls) -> "SSHConfig": key_path = Path(_get("KALI_SSH_KEY", required=True)).expanduser() passphrase = _get("KALI_SSH_KEY_PASSPHRASE", "") or None password = _get("KALI_SSH_PASSWORD", "") or None return cls( host=_get("KALI_HOST", required=True), port=int(_get("KALI_PORT", "22")), username=_get("KALI_USER", required=True), key_path=key_path, key_passphrase=passphrase, password=password, command_timeout=int(_get("SSH_COMMAND_TIMEOUT", "600")), ) def _normalize_keep_alive(raw: str) -> str: """Ollama keep_alive degerini Go time.ParseDuration ile uyumlu hale getir. Ollama daemon `time: missing unit in duration "1"` benzeri 400 doner. Birimsiz int gelirse: - "0" -> "0s" (hemen unload) - "-1" -> "-1m" (sonsuz tutmak icin negatif duration; daemon -1ns gibi anlam vermez ama negatif degerler "kalici" sayilir) - "N" -> "Nm" (birimsiz pozitifi dakika kabul ediyoruz) Birim zaten varsa (ms/s/m/h/...) oldugu gibi birakilir. """ s = (raw or "").strip() if not s: return "30m" try: n = int(s) except ValueError: return s if n == 0: return "0s" if n < 0: return "-1m" return f"{n}m" @dataclass(frozen=True) class OllamaConfig: host: str model: str # Modelfile'daki PARAMETER'lari override etmek icin opsiyonel runtime # ayarlari. None ise Modelfile'daki deger kullanilir. num_gpu: int | None num_ctx: int | None temperature: float | None # Modelin Ollama daemon belleginde tutulma suresi. # "30m" / "1h" / "-1" (sonsuz) / "0" (hemen unload). # Suresiz tutmak CUDA re-initialization hatalarini azaltir. keep_alive: str @classmethod def from_env(cls) -> "OllamaConfig": def _opt_int(name: str) -> int | None: raw = _get(name, "").strip() return int(raw) if raw else None def _opt_float(name: str) -> float | None: raw = _get(name, "").strip() return float(raw) if raw else None return cls( host=_get("OLLAMA_HOST", "http://127.0.0.1:11434"), model=_get("OLLAMA_MODEL", "kali-agent"), num_gpu=_opt_int("OLLAMA_NUM_GPU"), num_ctx=_opt_int("OLLAMA_NUM_CTX"), temperature=_opt_float("OLLAMA_TEMPERATURE"), keep_alive=_normalize_keep_alive(_get("OLLAMA_KEEP_ALIVE", "30m")), ) def runtime_options(self) -> dict[str, float | int]: """Sadece set edilmis alanlari iceren options dict'i don.""" opts: dict[str, float | int] = {} if self.num_gpu is not None: opts["num_gpu"] = self.num_gpu if self.num_ctx is not None: opts["num_ctx"] = self.num_ctx if self.temperature is not None: opts["temperature"] = self.temperature return opts def _opt_bool(name: str, default: bool) -> bool: raw = _get(name, "").strip().lower() if not raw: return default return raw in {"1", "true", "yes", "on", "evet", "y"} @dataclass(frozen=True) class VoiceConfig: """Voice (sesli komut) ozelligi icin konfigurasyon. `enabled=False` ise voice modulleri hic yuklenmez; kullanici `:v` komutunu denerse "kapali" der. `enabled=True` iken bile sounddevice veya faster-whisper kurulu degilse modul dinamik olarak devre disi kalir ve net bir mesaj gosterir. """ enabled: bool prewarm: bool # baslangicta modeli yukle (ilk :v gecikmesi olmasin) model_size: str # tiny / base / small / medium / large-v3 device: str # cpu / cuda / auto compute_type: str # int8 / int8_float16 / float16 / float32 language: str # "tr", "en", "auto" samplerate: int # 16000 onerilir silence_threshold: float # RMS esigi (0..1, ~0.008-0.02 normal) silence_seconds: float # bu kadar sessizlikten sonra dinleme biter max_seconds: float # sert ust sinir initial_timeout: float # konusma baslamazsa iptal input_device: str # bos => sistem default; aksi: index ya da isim parcasi confirm: bool # transkripti gondermeden once onay sor @classmethod def from_env(cls) -> "VoiceConfig": def _opt_float(name: str, default: float) -> float: raw = _get(name, "").strip() return float(raw) if raw else default def _opt_int(name: str, default: int) -> int: raw = _get(name, "").strip() return int(raw) if raw else default return cls( enabled=_opt_bool("VOICE_ENABLED", True), prewarm=_opt_bool("VOICE_PREWARM", True), model_size=_get("VOICE_MODEL", "small").strip() or "small", device=_get("VOICE_DEVICE", "cpu").strip() or "cpu", compute_type=_get("VOICE_COMPUTE_TYPE", "int8").strip() or "int8", language=_get("VOICE_LANGUAGE", "tr").strip() or "tr", samplerate=_opt_int("VOICE_SAMPLERATE", 16000), silence_threshold=_opt_float("VOICE_SILENCE_THRESHOLD", 0.012), silence_seconds=_opt_float("VOICE_SILENCE_SECONDS", 1.5), max_seconds=_opt_float("VOICE_MAX_SECONDS", 60.0), initial_timeout=_opt_float("VOICE_INITIAL_TIMEOUT", 8.0), input_device=_get("VOICE_INPUT_DEVICE", "").strip(), confirm=_opt_bool("VOICE_CONFIRM", True), ) @dataclass(frozen=True) class SafetyConfig: """safety.py icin konfigurasyon. mode: - "enforce": "block" siniflandirilmis kurallar komutu durdurur. - "audit" : kural eslese de komut yine de calistirilir, sadece loglanir. """ mode: str # "enforce" | "audit" @classmethod def from_env(cls) -> "SafetyConfig": mode = _get("SAFETY_MODE", "enforce").strip().lower() if mode not in {"enforce", "audit"}: raise RuntimeError( f"SAFETY_MODE gecersiz: {mode!r}. 'enforce' veya 'audit' olmali." ) return cls(mode=mode) LOGS_DIR = PROJECT_ROOT / "logs" LOGS_DIR.mkdir(exist_ok=True)