odboy
932db9fbdc
- 移除加密相关代码（服务端和所有客户端） - 修复 JWT 验证未检查签名算法的问题 - 修复重置密码使用不安全随机数的问题 - 实现 Logout 服务端失效机制 - 修复历史版本对比 diff 颜色标记错误 - 修复占位符解析循环引用风险 - 添加用户禁用状态检查 - 更新 bug 清单文档
135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
import logging
|
|
import threading
|
|
import time
|
|
|
|
import requests
|
|
|
|
from kenaito_conf.cache import Cache
|
|
from kenaito_conf.config import Config
|
|
from kenaito_conf.listener import Listener
|
|
|
|
logger = logging.getLogger("kenaito_conf")
|
|
|
|
|
|
class ConfigClient:
|
|
def __init__(self, config: Config):
|
|
self._config = config
|
|
self._cache = Cache(config.cache_dir)
|
|
self._listener = Listener()
|
|
self._configs: dict[str, str] = {}
|
|
self._last_time: int = 0
|
|
self._lock = threading.RLock()
|
|
self._stop_event = threading.Event()
|
|
self._started = False
|
|
|
|
def on_change(self, handler) -> None:
|
|
self._listener.on_change(handler)
|
|
|
|
def get(self, key: str) -> str | None:
|
|
with self._lock:
|
|
return self._configs.get(key)
|
|
|
|
def get_all(self) -> dict[str, str]:
|
|
with self._lock:
|
|
return dict(self._configs)
|
|
|
|
def start(self) -> None:
|
|
with self._lock:
|
|
if self._started:
|
|
return
|
|
self._started = True
|
|
|
|
try:
|
|
self._fetch_all()
|
|
except Exception as e:
|
|
logger.warning("[ConfigClient] fetch all configs error: %s, trying cache", e)
|
|
cached = self._cache.read()
|
|
if cached is not None:
|
|
self._apply_configs(cached)
|
|
logger.info("[ConfigClient] loaded %d configs from cache", len(self._configs))
|
|
else:
|
|
raise RuntimeError(f"fetch configs failed and no cache: {e}") from e
|
|
|
|
t = threading.Thread(target=self._sync_loop, daemon=True)
|
|
t.start()
|
|
|
|
def stop(self) -> None:
|
|
self._stop_event.set()
|
|
|
|
def _fetch_all(self) -> None:
|
|
url = f"{self._config.base_url}/api/fetchAllConfigs"
|
|
payload = {
|
|
"appName": self._config.app_name,
|
|
"env": self._config.env,
|
|
"version": self._config.version,
|
|
"ignoreGlobal": self._config.ignore_global,
|
|
"token": self._config.token,
|
|
}
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
if data.get("code", 0) != 0:
|
|
raise RuntimeError(f"server error: {data.get('msg', 'unknown')}")
|
|
|
|
ts = data.get("latest.fetch.timestamp")
|
|
if ts:
|
|
self._last_time = int(ts)
|
|
|
|
self._cache.write(data)
|
|
self._apply_configs(data)
|
|
logger.info("[ConfigClient] fetched %d configs", len(self._configs))
|
|
|
|
def _fetch_changed(self) -> None:
|
|
url = f"{self._config.base_url}/api/fetchChangedConfigs"
|
|
payload = {
|
|
"appName": self._config.app_name,
|
|
"env": self._config.env,
|
|
"version": self._config.version,
|
|
"lastTime": self._last_time,
|
|
"ignoreGlobal": self._config.ignore_global,
|
|
"token": self._config.token,
|
|
}
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
|
|
if result.get("code", -1) != 0:
|
|
raise RuntimeError(f"server error code: {result.get('code')}")
|
|
|
|
changed_data = result.get("data", {})
|
|
|
|
ts = changed_data.pop("latest.fetch.timestamp", None)
|
|
if ts:
|
|
self._last_time = int(ts)
|
|
|
|
if changed_data:
|
|
changed = {}
|
|
for k, v in changed_data.items():
|
|
changed[k] = v
|
|
|
|
with self._lock:
|
|
self._configs.update(changed)
|
|
|
|
self._listener.notify(changed)
|
|
logger.info("[ConfigClient] %d configs changed", len(changed))
|
|
|
|
def _apply_configs(self, data: dict) -> None:
|
|
with self._lock:
|
|
for k, v in data.items():
|
|
if k.startswith("kenaito.") or k == "latest.fetch.timestamp":
|
|
continue
|
|
if not isinstance(v, str):
|
|
continue
|
|
self._configs[k] = v
|
|
|
|
def _sync_loop(self) -> None:
|
|
while not self._stop_event.is_set():
|
|
self._stop_event.wait(self._config.sync_interval)
|
|
if self._stop_event.is_set():
|
|
break
|
|
try:
|
|
self._fetch_changed()
|
|
except Exception:
|
|
logger.exception("[ConfigClient] sync error")
|