from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path def load_env_file(path: Path) -> None: if not path.exists(): return for raw_line in path.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) os.environ.setdefault(key.strip(), value.strip()) @dataclass(slots=True) class ServerConfig: host: str port: int model: str default_provider: str fallback_providers: list[str] gigachat_model: str gigachat_auth_key: str gigachat_scope: str gigachat_api_base_url: str gigachat_oauth_url: str gigachat_ca_bundle: Path | None gigachat_insecure_skip_verify: bool yandexgpt_model: str workspace_root: Path session_dir: Path state_dir: Path system_prompt: str max_tool_rounds: int max_file_read_bytes: int max_command_output_bytes: int tool_policy: str approval_timeout_seconds: int jobs_retention_seconds: int approvals_retention_seconds: int tavily_api_key: str google_search_api_key: str google_search_engine_id: str @classmethod def load(cls) -> "ServerConfig": base_dir = Path(__file__).resolve().parent load_env_file(base_dir / ".env") workspace_root = Path( os.environ.get("NEW_QWEN_WORKSPACE_ROOT", str(Path.cwd())) ).expanduser() session_dir = Path( os.environ.get( "NEW_QWEN_SESSION_DIR", str(base_dir.parent / ".new-qwen" / "sessions"), ) ).expanduser() state_dir = Path( os.environ.get( "NEW_QWEN_STATE_DIR", str(base_dir.parent / ".new-qwen" / "state"), ) ).expanduser() return cls( host=os.environ.get("NEW_QWEN_HOST", "127.0.0.1"), port=int(os.environ.get("NEW_QWEN_PORT", "8080")), model=os.environ.get("NEW_QWEN_MODEL", "qwen3.6-plus"), default_provider=os.environ.get("NEW_QWEN_DEFAULT_PROVIDER", "qwen").strip() or "qwen", fallback_providers=[ item.strip() for item in os.environ.get("NEW_QWEN_FALLBACK_PROVIDERS", "").split(",") if item.strip() ], gigachat_model=os.environ.get("NEW_QWEN_GIGACHAT_MODEL", "GigaChat").strip(), gigachat_auth_key=os.environ.get("NEW_QWEN_GIGACHAT_AUTH_KEY", "").strip(), gigachat_scope=os.environ.get("NEW_QWEN_GIGACHAT_SCOPE", "GIGACHAT_API_PERS").strip(), gigachat_api_base_url=os.environ.get( "NEW_QWEN_GIGACHAT_API_BASE_URL", "https://gigachat.devices.sberbank.ru/api/v1", ).strip(), gigachat_oauth_url=os.environ.get( "NEW_QWEN_GIGACHAT_OAUTH_URL", "https://ngw.devices.sberbank.ru:9443/api/v2/oauth", ).strip(), gigachat_ca_bundle=( Path(os.environ["NEW_QWEN_GIGACHAT_CA_BUNDLE"]).expanduser().resolve() if os.environ.get("NEW_QWEN_GIGACHAT_CA_BUNDLE", "").strip() else None ), gigachat_insecure_skip_verify=os.environ.get( "NEW_QWEN_GIGACHAT_INSECURE_SKIP_VERIFY", "", ).strip().lower() in {"1", "true", "yes", "on"}, yandexgpt_model=os.environ.get("NEW_QWEN_YANDEXGPT_MODEL", "yandexgpt").strip(), workspace_root=workspace_root.resolve(), session_dir=session_dir.resolve(), state_dir=state_dir.resolve(), system_prompt=os.environ.get("NEW_QWEN_SYSTEM_PROMPT", "").strip(), max_tool_rounds=int(os.environ.get("NEW_QWEN_MAX_TOOL_ROUNDS", "8")), max_file_read_bytes=int( os.environ.get("NEW_QWEN_MAX_FILE_READ_BYTES", "200000") ), max_command_output_bytes=int( os.environ.get("NEW_QWEN_MAX_COMMAND_OUTPUT_BYTES", "12000") ), tool_policy=os.environ.get("NEW_QWEN_TOOL_POLICY", "full-access").strip(), approval_timeout_seconds=int( os.environ.get("NEW_QWEN_APPROVAL_TIMEOUT_SECONDS", "3600") ), jobs_retention_seconds=int( os.environ.get("NEW_QWEN_JOBS_RETENTION_SECONDS", str(7 * 24 * 3600)) ), approvals_retention_seconds=int( os.environ.get( "NEW_QWEN_APPROVALS_RETENTION_SECONDS", str(7 * 24 * 3600), ) ), tavily_api_key=os.environ.get("NEW_QWEN_TAVILY_API_KEY", "").strip(), google_search_api_key=os.environ.get( "NEW_QWEN_GOOGLE_SEARCH_API_KEY", "", ).strip(), google_search_engine_id=os.environ.get( "NEW_QWEN_GOOGLE_SEARCH_ENGINE_ID", "", ).strip(), )