From 55fbe2f1f1a31fa028fd0f4e289ac626c9b2e7c7 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 02:15:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(confs):=20=E7=BB=9F=E4=B8=80=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E6=9C=8D=E5=8A=A1=EF=BC=88pymsg/TCP=20:8010=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - confs.py 重写为统一配置服务：get/list/reload/health 路由， 内存缓存 + mtime 自动重载，路径遍历防护，ini/json 按键提取 - 复制 log_client.py，新增 configs/example.ini 示例配置 - 注册到 core/init.txt（db 之后启动）与 _inst.sh 安装序列 --- _inst.sh | 1 + confs/_set.sh | 3 +- confs/configs/example.ini | 10 + confs/confs.py | 384 ++++++++++++++++++++++++++------------ confs/log_client.py | 173 +++++++++++++++++ core/init.txt | 1 + 6 files changed, 448 insertions(+), 124 deletions(-) create mode 100644 confs/configs/example.ini create mode 100644 confs/log_client.py diff --git a/_inst.sh b/_inst.sh index 5ef7711..94cd456 100755 --- a/_inst.sh +++ b/_inst.sh @@ -36,4 +36,5 @@ inst_module "dnfs" inst_module "db" inst_module "smtp" inst_module "captcha" +inst_module "confs" inst_module "core" # core last diff --git a/confs/_set.sh b/confs/_set.sh index 1eb7202..7a5e8a6 100755 --- a/confs/_set.sh +++ b/confs/_set.sh @@ -1,7 +1,6 @@ #!/bin/bash cd "$(dirname "$0")" || exit python3 -m venv venv -#sudo chmod +x .env/bin/activate source venv/bin/activate pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple -#nuitka --standalone smtp_v2.py +#nuitka --standalone confs.py diff --git a/confs/configs/example.ini b/confs/configs/example.ini new file mode 100644 index 0000000..cbb1dd4 --- /dev/null +++ b/confs/configs/example.ini @@ -0,0 +1,10 @@ +# 示例配置 —— 统一配置服务 confs 的配置文件存放目录 +# 各微服务的配置文件集中放在本目录，由 confs 通过 pymsg get 路由分发给内部服务。 +# 支持类型：.ini / .conf / .cfg / .json / .yaml / .yml / .toml / .txt + +[Server] +address = 127.0.0.1 +port = 8010 + +[Feature] +dark_mode = on diff --git a/confs/confs.py b/confs/confs.py index 4b1de6b..4e066cf 100644 --- a/confs/confs.py +++ b/confs/confs.py @@ -1,158 +1,298 @@ -#!/usr/bin/env python3 -''' -使用pymsg库的尝试 -''' +""" +confs — 统一配置服务（pymsg/TCP，默认 127.0.0.1:8010） -from pymsg import MsgServer -from email.mime.text import MIMEText -from email.header import Header -from log_client import logs -from dotenv import load_dotenv -import smtplib -import random -import time -import sys, os +集中存放各微服务的配置文件，通过 pymsg 协议分发给内部服务， +替代各服务本地散落的 ini/json/yaml 配置文件，实现配置的统一管理与下发。 + +pymsg 路由:: + + {"_route": "get", "name": "example.ini"} → 取配置文件全文 + {"_route": "get", "name": "example.ini", "key": "Server.port"} → 取单个键（ini/json） + {"_route": "list"} → 列出全部配置 + {"_route": "reload", "name": "example.ini"} → 重载指定配置（缺省清空全部缓存） + {"_route": "health"} → 健康检查 + +响应格式（与 db / smtp / captcha 一致）:: + + {"success": true, "data": {...}} + {"success": false, "error": "..."} + +配置文件存放于 confs/configs/ 目录（可用 --dir 覆盖），内存缓存 + +文件 mtime 变化自动重载；配置名校验防路径遍历。仅监听 127.0.0.1，不对外暴露。 +""" + +import configparser import json +import os +import sys +import threading +import time +from datetime import datetime, timezone -# 从文件加载到环境变量？ -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")) -SMTP_SERVER: str = os.getenv('SMTP_SERVER') -SMTP_PORT: int = int(os.getenv("SMTP_PORT")) -SMTP_USER: str = os.getenv('SMTP_USER') -SMTP_PASSWORD: str = os.getenv('__PASSWORD') +# pymsg / log_client 导入：生产环境 pymsg 由 venv（pip install ../pymsg）提供； +# 附加 ../pymsg 源码路径作为无 venv 时的开发兜底 +_here = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(_here, "..")) +sys.path.insert(0, os.path.join(_here, "..", "pymsg")) +from pymsg import RpcServer # noqa: E402 +from log_client import logs # noqa: E402 + +# ── 配置 ──────────────────────────────────────────────────────────────── + +CONFIG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "configs") +# 允许分发的配置文件类型（白名单） +ALLOWED_EXT = {".ini", ".conf", ".cfg", ".json", ".yaml", ".yml", ".toml", ".txt"} -print("【SMTP】") -print("server: ", SMTP_SERVER) -print("port: ", SMTP_PORT) -print("user: ", SMTP_USER) -print("password: ", SMTP_PASSWORD) +# ── 异常 ──────────────────────────────────────────────────────────────── -captchas: dict[str, list[int, int]] = {} +class ConfigError(Exception): + """配置服务业务错误（由 RpcServer 包装为 success=false 返回）。""" -class SmtpServer: - '''SMTP服务器类''' +class ConfigNotFound(ConfigError): + """配置文件不存在。""" - def __init__(self, addr="127.0.0.1", port=1234, workers=4): + +class ConfigNameInvalid(ConfigError): + """配置名非法（路径遍历或类型不允许）。""" + + +class KeyNotFound(ConfigError): + """配置键不存在或键格式错误。""" + + +def _iso(ts: float) -> str: + """时间戳转 ISO 8601（UTC，与 db 服务的日期序列化约定一致）。""" + return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() + + +# ═══════════════════════════════════════════════════════════════════════ +# ConfigStore — 配置文件仓库 +# ═══════════════════════════════════════════════════════════════════════ + +class ConfigStore: + """配置文件仓库：磁盘 configs/ 目录 + 内存缓存，mtime 变化自动重载。 + + 所有公开方法持锁操作（配置文件小、内部调用频率低，简单持锁即可）， + 保证 workers>0 并发回调下的缓存一致性。 + """ + + def __init__(self, root: str): + self.root = os.path.realpath(root) + os.makedirs(self.root, exist_ok=True) + self._cache = {} # name -> {"content": str, "mtime": float, "size": int} + self._lock = threading.Lock() + + def _resolve(self, name: str) -> str: + """校验配置名并返回绝对路径（防路径遍历 + 扩展名白名单）。""" + if not name or name != os.path.basename(name): + raise ConfigNameInvalid(f"非法配置名: {name!r}") + ext = os.path.splitext(name)[1].lower() + if ext not in ALLOWED_EXT: + raise ConfigNameInvalid(f"不支持的配置类型: {ext or '(无扩展名)'}") + path = os.path.realpath(os.path.join(self.root, name)) + if not path.startswith(self.root + os.sep): + raise ConfigNameInvalid(f"非法配置名: {name!r}") + if not os.path.isfile(path): + raise ConfigNotFound(f"配置不存在: {name}") + return path + + def get(self, name: str) -> dict: + """取配置内容（带缓存；文件 mtime 变化时自动重新读取）。""" + path = self._resolve(name) + with self._lock: + try: + mtime = os.path.getmtime(path) + except OSError: + self._cache.pop(name, None) + raise ConfigNotFound(f"配置不存在或不可读: {name}") from None + entry = self._cache.get(name) + if entry is not None and entry["mtime"] == mtime: + return entry + try: + with open(path, "r", encoding="utf-8") as f: + content = f.read() + except (OSError, UnicodeDecodeError) as e: + raise ConfigError(f"配置读取失败: {name}（{e}）") from None + entry = {"content": content, "mtime": mtime, + "size": len(content.encode("utf-8"))} + self._cache[name] = entry + logs.info("confs", f"配置已加载 {name}（{entry['size']}B）") + return entry + + def list(self) -> list: + """列出配置目录下全部允许类型的配置文件。""" + out = [] + for name in sorted(os.listdir(self.root)): + if os.path.splitext(name)[1].lower() not in ALLOWED_EXT: + continue + path = os.path.join(self.root, name) + if not os.path.isfile(path): + continue + st = os.stat(path) + out.append({"name": name, "size": st.st_size, + "mtime": _iso(st.st_mtime), + "cached": name in self._cache}) + return out + + def reload(self, name: "str | None" = None) -> list: + """重载配置：指定 name 立即重新读取；缺省清空全部缓存（下次访问惰性重载）。 + + Returns: + 受影响（重载/移出缓存）的配置名列表。 + """ + if name is not None: + self._resolve(name) # 校验，文件不存在则抛 ConfigNotFound + with self._lock: + self._cache.pop(name, None) + self.get(name) + return [name] + with self._lock: + dropped = list(self._cache.keys()) + self._cache.clear() + return dropped + + def cached_count(self) -> int: + with self._lock: + return len(self._cache) + + +# ── 按键提取（可选功能） ──────────────────────────────────────────────── + +def _extract_key(name: str, content: str, key: str): + """从配置内容中提取单个键。 + + - ini/conf/cfg：键格式 ``Section.option``，返回字符串值； + - json：键格式为点号路径（如 ``server.port``，列表可用数字下标）， + 返回原始 JSON 值； + - 其他类型不支持按键查询，需取全文后自行解析。 + """ + ext = os.path.splitext(name)[1].lower() + + if ext in (".ini", ".conf", ".cfg"): + parser = configparser.ConfigParser() + try: + parser.read_string(content) + except configparser.Error as e: + raise ConfigError(f"INI 解析失败: {name}（{e}）") from None + section, _, option = key.partition(".") + if not section or not option: + raise KeyNotFound(f"INI 键格式应为 Section.option: {key!r}") + if parser.has_option(section, option): + return parser.get(section, option) + raise KeyNotFound(f"键不存在: {name} -> {key}") + + if ext == ".json": + try: + node = json.loads(content) + except json.JSONDecodeError as e: + raise ConfigError(f"JSON 解析失败: {name}（{e}）") from None + for part in key.split("."): + if isinstance(node, dict) and part in node: + node = node[part] + elif isinstance(node, list) and part.isdigit() and int(part) < len(node): + node = node[int(part)] + else: + raise KeyNotFound(f"键不存在: {name} -> {key}") + return node + + raise ConfigError(f"{ext} 类型不支持按键查询，请取全文后自行解析") + + +# ═══════════════════════════════════════════════════════════════════════ +# ConfsServer — 主服务类 +# ═══════════════════════════════════════════════════════════════════════ + +class ConfsServer: + """统一配置服务：集中管理并向内部微服务分发配置文件。""" + + def __init__(self, addr="127.0.0.1", port=8010, workers=4, config_dir=CONFIG_DIR): self.addr = addr self.port = port - self._server = MsgServer(addr, port, workers=workers) + self._started_at = time.time() + self.store = ConfigStore(config_dir) + self._server = RpcServer( + addr, port, workers=workers, + on_error=lambda route, e: logs.error( + "confs", f"路由 {route} 处理异常: {e}"), + ) + self._server.add_route("get", self._handle_get) + self._server.add_route("list", self._handle_list) + self._server.add_route("reload", self._handle_reload) + self._server.add_route("health", self._handle_health) - self._route = { - "send_email": self._send_email, - "verify_email": self._verify_email - } + # ── 启动 / 停止 ────────────────────────────────────────────────── - def start(self): - self._server.start(self._analysis) + def start(self) -> None: + """启动服务（阻塞当前线程）。""" + print(f"[confs] 监听 tcp://{self.addr}:{self.port}，配置目录 {self.store.root}") + logs.info("confs", f"开始监听 tcp://{self.addr}:{self.port}" + f"（配置目录 {self.store.root}）") + self._server.start() - def stop(self): + def stop(self) -> None: + """从另一线程停止服务。""" self._server.stop() - def _analysis(self, sock, addr, data): - print("""【recv】\n"""+str(addr)+'\n'+data) # 打印实际收到的消息 + # ── 路由处理 ───────────────────────────────────────────────────── + def _handle_get(self, req): + """取配置文件全文；带 key 时返回单个键的值（仅 ini/json）。""" + name = str(req.get("name", "")).strip() + if not name: + raise ConfigError("缺少 name 字段") + entry = self.store.get(name) - # 解包json - try: - request = json.loads(data) - except json.JSONDecodeError as err: - self._reply(sock, False, error=f"解析JSON失败: {err}") - return + key = req.get("key") + if key is None: + logs.info("confs", f"分发配置 {name}（{entry['size']}B）") + return {"name": name, "content": entry["content"], + "size": entry["size"], "mtime": _iso(entry["mtime"])} - # 路由 - route = request.pop("_route", "") - func = self._route.get(route) - if func is None: - self._reply(sock, False, error=f"解析路由失败：没有找到路由{route}") - return + key = str(key).strip() + value = _extract_key(name, entry["content"], key) + logs.info("confs", f"分发配置键 {name} -> {key}") + return {"name": name, "key": key, "value": value} - # 执行分支 - try: - result = func(request) - self._reply(sock, True, data=result) - except Exception as err: - self._reply(sock, False, error=f"执行操作失败: {err}") + def _handle_list(self, _req): + """列出配置目录下全部配置文件及其元信息。""" + return {"config_dir": self.store.root, "configs": self.store.list()} - def _reply(self, sock, success: bool, data: str = None, error: str = None): - print("""【send】\n"""+str(data)+'\n'+str(error)) # 打印实际输出 - response = {"success": success} - if data is not None: - response["data"] = data - if error is not None: - response["error"] = error - self._server.reply(sock, json.dumps(response, ensure_ascii=False)) - - - def _send_email(self, data): - def build_captcha() -> str: - captcha: str = "" - for i in range(0,6): - digit: int = random.randint(0,9) - captcha += str(digit) - return captcha - - recipient: str = data.get("email") - subject: str = "DNFS 邮箱验证" - captcha: str = build_captcha() - content: str = f"DNFS: 您的邮箱验证码为 {captcha}，一分钟内有效。" - - captchas[recipient] = ["", ""] - captchas[recipient][0] = captcha - captchas[recipient][1] = str(int(time.time())+60) - - message = MIMEText(content, "plain", "UTF-8") - message["From"] = SMTP_USER - message["To"] = recipient - message["Subject"] = subject - - # 这一行不用写到try里，因为如果它失败就没有这个类不用quit() - # 反而因为写进了try，和finally不在一个作用域 - server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) - try: - server.login(SMTP_USER, SMTP_PASSWORD) - server.sendmail(SMTP_USER, recipient, message.as_string()) - logs.info("smtp", f"{recipient}: 验证码发送成功({captcha})") - return {"success": True, "data": "发送成功"} - - except Exception as err: - logs.error("smtp", f"{recipient}: 验证码发送失败({err})") - return {"success": False, "error": f"发送失败: {err}"} - - finally: - server.quit() - - - def _verify_email(data): - for i in list(captchas.keys()): - if int(captchas[i][1]) < int(time.time()): - del captchas[i] - - email: str = data.get("email") - input_captcha: str = data.get("code") - - if captchas[email][0] == input_captcha: - logs.info("smtp", f"{email}: 验证通过") - return {"success": True, "data": "验证通过"} + def _handle_reload(self, req): + """重载指定配置；缺省 name 时清空全部缓存（下次访问惰性重载）。""" + name = req.get("name") + names = self.store.reload(str(name).strip() if name else None) + if name: + logs.info("confs", f"重载配置 {names[0]}") else: - logs.info("smtp", f"{email}: 验证失败") - return {"success": False, "data": "验证失败"} + logs.info("confs", f"清空配置缓存（{len(names)} 项）") + return {"reloaded": names} + + def _handle_health(self, _req): + return {"status": "ok", "service": "confs", + "config_dir": self.store.root, + "cached": self.store.cached_count(), + "uptime": int(time.time() - self._started_at)} +# ═══════════════════════════════════════════════════════════════════════ +# 入口 +# ═══════════════════════════════════════════════════════════════════════ if __name__ == "__main__": import argparse - parser = argparse.ArgumentParser(description="smtp--pymsg邮箱验证码服务") + parser = argparse.ArgumentParser(description="confs — 统一配置服务（pymsg/TCP）") parser.add_argument("--addr", default="127.0.0.1", help="监听地址") - parser.add_argument("--port", type=int, default=8008, help="监听端口") + parser.add_argument("--port", type=int, default=8010, help="监听端口") parser.add_argument("--workers", type=int, default=4, help="回调并发线程数（0=同步串行模式）") + parser.add_argument("--dir", default=CONFIG_DIR, help="配置文件目录") args = parser.parse_args() - server = SmtpServer(args.addr, args.port) + server = ConfsServer(args.addr, args.port, + workers=args.workers, config_dir=args.dir) try: server.start() except KeyboardInterrupt: diff --git a/confs/log_client.py b/confs/log_client.py new file mode 100644 index 0000000..60a426d --- /dev/null +++ b/confs/log_client.py @@ -0,0 +1,173 @@ +"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。 + +设计要点： + - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 + - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 + - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + +用法： + from log_client import logs + + logs.info("dnfs", "节点新增成功 id=42") + logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) + logs.push("gateway", "WARNING", "登录失败次数超限", + trace_id="req-abc", extra={"ip": "1.2.3.4"}) +""" + +import json +import os as _os +import sys as _sys +import threading +import queue +import time +import atexit +from datetime import datetime, timezone + +_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) +from pymsg import RpcClient, RpcError # noqa: E402 + +# ── 配置 ────────────────────────────────────────────────────────────── + +_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) + +_BATCH_SIZE = 50 # 队列满 50 条时 flush +_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush +_TIMEOUT = 5 # socket 超时（秒） + + +class LogClient: + """异步批量日志推送客户端。""" + + def __init__(self, service_name="log", + batch_size=_BATCH_SIZE, + flush_interval=_FLUSH_INTERVAL): + self._service_name = service_name + self._batch_size = batch_size + self._flush_interval = flush_interval + self._queue = queue.Queue() + self._stop = threading.Event() + self._flush_signal = threading.Event() + self._thread = threading.Thread(target=self._flush_loop, + daemon=True, name='log-client') + self._thread.start() + atexit.register(self._shutdown) + + # ── 公开接口 ──────────────────────────────────────────────────────── + + def push(self, service, level, message, trace_id=None, extra=None): + """推送一条日志（非阻塞，立即返回）。""" + self._queue.put({ + 'service': service, + 'level': level, + 'message': message, + 'trace_id': trace_id, + 'extra': extra, + }) + + def debug(self, service, message, **kw): + self.push(service, 'DEBUG', message, **kw) + + def info(self, service, message, **kw): + self.push(service, 'INFO', message, **kw) + + def warning(self, service, message, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service, message, **kw): + self.push(service, 'ERROR', message, **kw) + + def critical(self, service, message, **kw): + self.push(service, 'CRITICAL', message, **kw) + + def flush(self): + """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" + self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch + time.sleep(0.6) # 等待后台线程响应 + batch = [] + while True: + try: + batch.append(self._queue.get_nowait()) + if len(batch) >= self._batch_size: + self._send(batch) + batch = [] + except queue.Empty: + break + if batch: + self._send(batch) + self._flush_signal.clear() + + # ── 内部实现 ──────────────────────────────────────────────────────── + + def _flush_loop(self): + """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。""" + batch = [] + last_flush = time.monotonic() + + while not self._stop.is_set(): + timeout = self._flush_interval - (time.monotonic() - last_flush) + chk_timeout = max(0.1, min(timeout, 0.5)) + try: + item = self._queue.get(timeout=chk_timeout) + batch.append(item) + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + pass + + should_flush = ( + len(batch) >= self._batch_size or + self._flush_signal.is_set() or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush and batch: + self._send(batch) + batch = [] + last_flush = time.monotonic() + + # 退出前 flush 剩余 + if batch: + self._send(batch) + + def _send(self, batch): + """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。""" + now = datetime.now(timezone.utc).isoformat() + client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") + for entry in batch: + extra_val = entry.get("extra") + payload = { + "action": "insert", + "table": "Logs", + "values": { + "ts": now, + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": json.dumps(extra_val, ensure_ascii=False) + if extra_val is not None else None, + }, + } + try: + client.call("execute", payload, timeout=_TIMEOUT) + except RpcError as e: + # 业务错误（如表结构不符）：打印后继续下一条 + print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) + except Exception as e: + # 网络错误：db 不可用，放弃本批剩余（fail-safe） + print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + break + + def _shutdown(self): + """进程退出时 flush 剩余日志。""" + self._stop.set() + self.flush() + + +# ── 全局单例 ────────────────────────────────────────────────────────── + +logs = LogClient() diff --git a/core/init.txt b/core/init.txt index 9343f3d..a3e4926 100644 --- a/core/init.txt +++ b/core/init.txt @@ -1,4 +1,5 @@ bash /opt/micro_servers/db/_run.sh +bash /opt/micro_servers/confs/_run.sh bash /opt/micro_servers/ai_review/_run.sh bash /opt/micro_servers/dnfs/_run.sh bash /opt/micro_servers/smtp/_run.sh