feat(captcha): 接入 confs 配置中心并迁移字体路径

- confs/configs/captcha.ini: 新增 captcha 配置 (server + captcha 段, 含 font_path)
- captcha/log_client.py: 新增日志客户端副本 (service_name=captcha)
- captcha/captcha_service.py: 接入统一的 confs 配置加载模式
  (confs 强依赖 retry_on_refused -> 环境变量覆盖 -> 完整性校验)
  - CAPTCHA_LEN/CAPTCHA_TTL 改为可配置
  - 字体路径迁移到 confs, 新增 CAPTCHA_FONT_PATH 环境变量覆盖与 PIL 回退兜底
  - 关键路径 (启动/生成/校验失败) 接入 log_client 日志上报

captcha/captcha_service.py

@@ -20,6 +20,7 @@ captcha — 基于 pymsg 协议（MsgServer）的图形验证码服务。
 """
 
 import base64
+import configparser
 import io
 import os
 import random
@@ -30,15 +31,96 @@ import time
 
 from PIL import Image, ImageDraw, ImageFont
 
-# pymsg 导入
+# pymsg / log_client 导入
 import sys as _sys
 _sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 from pymsg import RpcServer  # noqa: E402
+from log_client import logs  # noqa: E402
 
-# ── 配置 ────────────────────────────────────────────────────────────────
+# ── 配置加载（两层：confs 强依赖 → 环境变量覆盖）────────────────────────
+# confs 是 captcha 启动的强依赖：必须成功从 confs 拉取 captcha.ini，否则直接退出。
+# 环境变量优先级最高（Docker 兼容），可覆盖 confs 中的配置值。
+# confs 自身的连接地址通过引导配置确定：环境变量 CONFS_HOST/CONFS_PORT
+# 优先，缺省使用 127.0.0.1:8010（与其他服务一致的约定）。
 
-CAPTCHA_LEN = 4
-CAPTCHA_TTL = 300        # Token 有效期（秒）
+# 引导配置：confs 连接地址（不依赖 confs 本身）
+_confs_host = os.environ.get('CONFS_HOST', '127.0.0.1')
+_confs_port = int(os.environ.get('CONFS_PORT', '8010'))
+
+# 白名单结构：_merge_ini 仅覆盖此处列出的 section/key，防止意外字段
+_captcha_config = {
+    'server': {'host': '', 'port': '', 'workers': ''},
+    'captcha': {'len': '', 'ttl': '', 'font_path': ''},
+}
+
+
+def _merge_ini(text):
+    """把 ini 文本合并进 _captcha_config（仅覆盖已知的 section/key）。"""
+    ini = configparser.ConfigParser(interpolation=None)
+    ini.read_string(text)
+    for section in _captcha_config:
+        if ini.has_section(section):
+            for key, value in ini[section].items():
+                if key in _captcha_config[section]:
+                    _captcha_config[section][key] = value
+
+
+def _fetch_from_confs():
+    """从 confs 统一配置服务拉取 captcha.ini 全文。
+
+    使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次，
+    每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。
+    RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。
+    """
+    from pymsg import RpcClient, RpcError  # noqa: E402
+    client = RpcClient(_confs_host, _confs_port, 'captcha', timeout=5,
+                       retry_on_refused=True)
+    try:
+        data = client.call('get', {'name': 'captcha.ini'})
+    except RpcError as e:
+        print(f'[captcha] 致命错误: 从 confs 获取 captcha.ini 失败: {e}',
+              file=_sys.stderr)
+        _sys.exit(1)
+    return data.get('content')
+
+
+_confs_text = _fetch_from_confs()
+_merge_ini(_confs_text)
+_config_source = f'confs (captcha.ini @ {_confs_host}:{_confs_port})'
+print(f'[captcha] 配置来源: {_config_source}')
+
+# 环境变量覆盖（最高优先级，保持 Docker 兼容）
+if os.environ.get('LISTEN_ADDR'):
+    _captcha_config['server']['host'] = os.environ['LISTEN_ADDR']
+if os.environ.get('LISTEN_PORT'):
+    _captcha_config['server']['port'] = os.environ['LISTEN_PORT']
+if os.environ.get('WORKERS'):
+    _captcha_config['server']['workers'] = os.environ['WORKERS']
+if os.environ.get('CAPTCHA_FONT_PATH'):
+    _captcha_config['captcha']['font_path'] = os.environ['CAPTCHA_FONT_PATH']
+
+# 配置完整性校验
+_required = {
+    'server': ['host', 'port', 'workers'],
+    'captcha': ['len', 'ttl'],
+}
+for section, keys in _required.items():
+    for key in keys:
+        if not _captcha_config[section][key]:
+            print(f'[captcha] 致命错误: 配置项 [{section}] {key} 缺失（captcha.ini 中未配置且无环境变量覆盖）',
+                  file=_sys.stderr)
+            _sys.exit(1)
+
+# 解析为最终常量
+CAPTCHA_LEN = int(_captcha_config['captcha']['len'])
+CAPTCHA_TTL = int(_captcha_config['captcha']['ttl'])
+
+# 字体路径：可选配置项（缺失或用空值时不致命，回退到代码默认值；
+# 即使默认值文件不存在，PIL 在 truetype() 失败时会回退到 load_ Default()）。
+DEFAULT_FONT_PATH = '/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf'
+FONT_PATH = _captcha_config['captcha']['font_path'] or DEFAULT_FONT_PATH
+
+logs.info("captcha", f"配置来源: {_config_source}")
 
 CHARS = string.ascii_letters + string.digits
 CHARS = CHARS.translate(str.maketrans('', '', 'lI1O0o'))   # 剔除易混淆字符
@@ -64,7 +146,7 @@ def generate_image(text):
         y = random.randint(2, 12)
         x = 15 + i * 32 + random.randint(-3, 3)   # 4 位字符在 160px 宽度内均匀分布
         try:
-            font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
+            font = ImageFont.truetype(FONT_PATH,
                                       random.choice(range(20,40,2)))
         except (IOError, OSError):
             font = ImageFont.load_default()
@@ -108,6 +190,7 @@ class CaptchaServer:
     def start(self) -> None:
         """启动服务（阻塞当前线程）。"""
         print(f"[captcha] 监听 tcp://{self.addr}:{self.port}")
+        logs.info("captcha", f"开始监听 tcp://{self.addr}:{self.port}")
         self._server.start()
 
     def stop(self) -> None:
@@ -132,6 +215,7 @@ class CaptchaServer:
                 del captcha_store[k]
 
         png_bytes = generate_image(text)
+        logs.info("captcha", f"验证码已生成 token={token[:8]}...")
         return {
             "token": token,
             "image": base64.b64encode(png_bytes).decode('ascii'),
@@ -148,6 +232,7 @@ class CaptchaServer:
             entry = captcha_store.pop(token, None)
 
         if not entry:
+            logs.warning("captcha", f"验证码校验失败-无效Token token={token[:8]}...")
             return {"valid": False, "error": "invalid or expired token"}
 
         expected = entry['answer']
@@ -159,6 +244,9 @@ class CaptchaServer:
                 diff |= ord(a) ^ ord(b)
             valid = diff == 0
 
+        if not valid:
+            logs.warning("captcha", f"验证码校验失败-答案错误 token={token[:8]}...")
+
         return {"valid": valid}
 
     def _handle_health(self, _req):
@@ -172,14 +260,19 @@ class CaptchaServer:
 if __name__ == '__main__':
     import argparse
 
+    default_addr = _captcha_config['server']['host']
+    default_port = int(_captcha_config['server']['port'])
+    default_workers = int(_captcha_config['server']['workers'])
+
     parser = argparse.ArgumentParser(description="captcha — pymsg 验证码服务")
-    parser.add_argument("--addr", default="127.0.0.1", help="监听地址")
-    parser.add_argument("--port", type=int, default=8009, help="监听端口")
-    parser.add_argument("--workers", type=int, default=4,
+    parser.add_argument("--addr", default=default_addr, help="监听地址")
+    parser.add_argument("--port", type=int, default=default_port, help="监听端口")
+    parser.add_argument("--workers", type=int, default=default_workers,
                         help="回调并发线程数（0=同步串行模式）")
     args = parser.parse_args()
 
     server = CaptchaServer(args.addr, args.port, workers=args.workers)
+    print(f'[captcha] 配置来源: {_config_source}')
     try:
         server.start()
     except KeyboardInterrupt:

captcha/log_client.py

@@ -0,0 +1,158 @@
+"""
+统一日志客户端 — captcha 服务副本（基于 pymsg.AsyncRpcQueue）。
+
+设计要点：
+  - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
+  - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
+  - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+  - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
+  - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
+
+用法（与旧版完全兼容）：
+    from log_client import logs
+
+    logs.info("captcha", "验证码已生成 token=abc123")
+    logs.error("captcha", "图片生成失败", extra={"errno": 2002})
+    logs.push("captcha", "WARNING", "Token 校验失败",
+              trace_id="req-abc", extra={"ip": "1.2.3.4"})
+"""
+
+import json
+import os as _os
+import sys as _sys
+import atexit
+from datetime import datetime, timezone
+
+_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
+from pymsg import RpcClient, RpcError, AsyncRpcQueue  # 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
+_FLUSH_INTERVAL = 2.0
+_TIMEOUT = 5
+_QUEUE_MAXSIZE = 10000
+
+
+def _safe_json(obj):
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(obj)
+
+
+def _now_iso():
+    """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
+    return datetime.now(timezone.utc).isoformat()
+
+
+class LogClient:
+    """异步批量日志推送客户端（API 与旧版完全兼容）。"""
+
+    def __init__(self, service_name: str = "log",
+                 batch_size: int = _BATCH_SIZE,
+                 flush_interval: float = _FLUSH_INTERVAL):
+        self._service_name = service_name
+        self._client = RpcClient(
+            _DBV2_ADDR, _DBV2_PORT,
+            service_name + "-log",
+            timeout=_TIMEOUT,
+        )
+        self._queue = AsyncRpcQueue(
+            handler=self._send_batch,
+            batch_size=batch_size,
+            flush_interval=flush_interval,
+            max_queue_size=_QUEUE_MAXSIZE,
+            on_error=self._on_error,
+        )
+        self._queue.start()
+        atexit.register(self.close)
+
+    # ── 公开接口 ────────────────────────────────────────────────────────
+
+    def push(self, service: str, level: str, message: str,
+             trace_id: str = None, extra=None):
+        """推送一条日志（非阻塞，立即返回）。
+
+        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
+        """
+        entry = {
+            'ts': _now_iso(),
+            'service': service,
+            'level': level,
+            'message': message,
+            'trace_id': trace_id,
+            'extra': extra,
+        }
+        self._queue.put(entry)
+
+    def debug(self, service: str, message: str, **kw):
+        self.push(service, 'DEBUG', message, **kw)
+
+    def info(self, service: str, message: str, **kw):
+        self.push(service, 'INFO', message, **kw)
+
+    def warning(self, service: str, message: str, **kw):
+        self.push(service, 'WARNING', message, **kw)
+
+    def warn(self, service: str, message: str, **kw):
+        self.push(service, 'WARNING', message, **kw)
+
+    def error(self, service: str, message: str, **kw):
+        self.push(service, 'ERROR', message, **kw)
+
+    def critical(self, service: str, message: str, **kw):
+        self.push(service, 'CRITICAL', message, **kw)
+
+    def flush(self):
+        """手动 flush 队列中剩余的日志。"""
+        self._queue.flush()
+
+    def close(self):
+        """关闭客户端，flush 剩余日志。"""
+        self._queue.close()
+
+    # ── 内部实现 ────────────────────────────────────────────────────────
+
+    def _send_batch(self, batch: list):
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
+        for entry in batch:
+            extra_val = entry.get("extra")
+            try:
+                extra_json = _safe_json(extra_val) if extra_val is not None else None
+            except Exception:
+                extra_json = None
+            payload = {
+                "action": "insert",
+                "table": "Logs",
+                "values": {
+                    "ts": entry.get("ts", _now_iso()),
+                    "service": entry.get("service"),
+                    "level": entry.get("level"),
+                    "message": entry.get("message"),
+                    "trace_id": entry.get("trace_id"),
+                    "extra": extra_json,
+                },
+            }
+            try:
+                self._client.call("execute", payload, timeout=_TIMEOUT)
+            except RpcError as e:
+                print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
+            except Exception as e:
+                print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
+                break
+
+    def _on_error(self, e: Exception, batch: list):
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
+              file=_sys.stderr)
+
+
+# ── 全局单例 ──────────────────────────────────────────────────────────
+
+logs = LogClient(service_name="captcha")