import os import sys import json import urllib.parse import subprocess import threading import asyncio from aiohttp import web import folder_paths import struct try: from ..net_proxy import CONFIG_KEY as PROXY_KEY, configured as configured_proxy, normalize as normalize_proxy from ..model_sidecars import layout as _side_file_layout except ImportError: # loaded outside the package (tests) from net_proxy import CONFIG_KEY as PROXY_KEY, configured as configured_proxy, normalize as normalize_proxy from model_sidecars import layout as _side_file_layout async def api_get_config(request): script_dir = os.path.dirname(os.path.abspath(__file__)) config_path = os.path.join(script_dir, "config.json") has_key = False folder_types = [] if os.path.exists(config_path): try: with open(config_path, 'r', encoding='utf-8') as f: cfg = json.load(f) has_key = bool(cfg.get("CIVITAI_API_KEY", "").strip()) folder_types = cfg.get("folder_types_config", []) except: pass if not has_key: legacy_config_path = os.path.join(os.path.dirname(script_dir), "config.json") if os.path.exists(legacy_config_path): try: with open(legacy_config_path, 'r', encoding='utf-8') as f: legacy_cfg = json.load(f) has_key = bool(legacy_cfg.get("CIVITAI_API_KEY", "").strip()) except Exception: pass deepl_key = "" try: with open(config_path, 'r', encoding='utf-8') as f: deepl_key = json.load(f).get("DEEPL_API_KEY", "") except (OSError, ValueError, AttributeError): pass return web.json_response({"has_api_key": has_key, "has_deepl_key": bool(isinstance(deepl_key, str) and deepl_key.strip()), "proxy": configured_proxy(), "sidecar_layout": _side_file_layout(), "folder_types_config": folder_types}) async def api_save_config(request): try: data = await request.json() api_key = data.get("api_key") folder_types_config = data.get("folder_types_config") script_dir = os.path.dirname(os.path.abspath(__file__)) config_path = os.path.join(script_dir, "config.json") cfg = {} if os.path.exists(config_path): try: with open(config_path, 'r', encoding='utf-8') as f: cfg = json.load(f) except: pass deepl_key = data.get("deepl_key") if api_key is not None and not isinstance(api_key, str): return web.json_response({"status": "error", "message": "Invalid API key"}, status=400) if deepl_key is not None and not isinstance(deepl_key, str): return web.json_response({"status": "error", "message": "Invalid DeepL key"}, status=400) proxy = None if "proxy" in data: # Settings -> Network (net_proxy.py); "" = the system's proxy try: proxy = normalize_proxy(data.get("proxy")) except ValueError as error: return web.json_response({"status": "error", "message": "Invalid proxy", "code": str(error)}, status=400) if folder_types_config is not None and not isinstance(folder_types_config, list): return web.json_response({"status": "error", "message": "Invalid folder configuration"}, status=400) if "physical_folders_config" in data and not isinstance(data.get("physical_folders_config"), list): return web.json_response({"status": "error", "message": "Invalid physical folder configuration"}, status=400) if "folder_view_mode" in data and data.get("folder_view_mode") not in {"abstract", "physical"}: return web.json_response({"status": "error", "message": "Invalid folder view mode"}, status=400) if api_key is not None: cfg["CIVITAI_API_KEY"] = api_key.strip() if deepl_key is not None: # Settings -> Translation (translation_routes.py) cfg["DEEPL_API_KEY"] = deepl_key.strip() if proxy is not None: cfg[PROXY_KEY] = proxy if folder_types_config is not None: cfg["folder_types_config"] = folder_types_config if "physical_folders_config" in data: cfg["physical_folders_config"] = data.get("physical_folders_config") if "folder_view_mode" in data: cfg["folder_view_mode"] = data.get("folder_view_mode") temp_path = config_path + ".tmp" with open(temp_path, 'w', encoding='utf-8') as f: json.dump(cfg, f, indent=4) os.replace(temp_path, config_path) return web.json_response({"status": "ok"}) except Exception as e: return web.json_response({"status": "error", "message": str(e)})