| """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
|
|
|
|
|
| num_gpu: int | None
|
| num_ctx: int | None
|
| temperature: float | None
|
|
|
|
|
|
|
| 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
|
| model_size: str
|
| device: str
|
| compute_type: str
|
| language: str
|
| samplerate: int
|
| silence_threshold: float
|
| silence_seconds: float
|
| max_seconds: float
|
| initial_timeout: float
|
| input_device: str
|
| confirm: bool
|
|
|
| @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
|
|
|
| @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)
|
|
|
