From 192610615e547e697e10018cfd3736d9df299dc0 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sun, 2 Aug 2026 01:09:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(captcha):=20=E6=8E=A5=E5=85=A5=20confs=20?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E4=B8=AD=E5=BF=83=E5=B9=B6=E8=BF=81=E7=A7=BB?= =?UTF-8?q?=E5=AD=97=E4=BD=93=E8=B7=AF=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 109 +++++++++++++++++++++++-- captcha/log_client.py | 158 +++++++++++++++++++++++++++++++++++++ confs/configs/captcha.ini | 9 +++ 3 files changed, 268 insertions(+), 8 deletions(-) create mode 100644 captcha/log_client.py create mode 100644 confs/configs/captcha.ini diff --git a/captcha/captcha_service.py b/captcha/captcha_service.py index c93331e..8df2d1b 100644 --- a/captcha/captcha_service.py +++ b/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: diff --git a/captcha/log_client.py b/captcha/log_client.py new file mode 100644 index 0000000..218ba3d --- /dev/null +++ b/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") diff --git a/confs/configs/captcha.ini b/confs/configs/captcha.ini new file mode 100644 index 0000000..daf4d78 --- /dev/null +++ b/confs/configs/captcha.ini @@ -0,0 +1,9 @@ +[server] +host = 127.0.0.1 +port = 8009 +workers = 4 + +[captcha] +len = 4 +ttl = 300 +font_path = /usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf