refactor: smtp 改造为 pymsg TCP 服务并修复 NameError/finally/KeyError 等已知 bug

README.md

@@ -49,7 +49,7 @@
 | [db](db/) | Python(pymsg) | 8005 | 通用数据库服务，基于 pymsg 协议（TCP 长度前缀帧 + JSON）的 SQL-over-TCP，SQLAlchemy Core 参数化查询防注入；内置集中式日志收集（Logs 表 + 自动清理） |
 | [ai_review](ai_review/) | Python(Flask) | 8007 | AI 内容审核服务，DeepSeek LLM + ChromaDB RAG，支持人工反馈纠偏 |
 | [captcha](captcha/) | Python(pymsg) | 8009 | 验证码服务（pymsg/TCP），生成图形验证码（5 分钟过期、一次性），经 gateway 的 `GET /captcha/image` 代理给浏览器 |
-| [smtp](smtp/) | Python(Flask) | 8008 | 邮件服务，发送/校验邮箱验证码（/send_email、/verify_email），注册流程的邮件验证码后端，SMTP 账号配置在 .env |
+| [smtp](smtp/) | Python(pymsg) | 8008 | 邮件服务，发送/校验邮箱验证码（pymsg 路由 send_email/verify_email/health），注册流程的邮件验证码后端，SMTP 账号配置在 .env |
 | [aibox](aibox/) | Python | 8765 | 独立 AI 智能体系统（DeepSeek + WebSocket），自带技能系统/记忆/任务调度，由 core 拉起但与主干微服务无调用关系 |
 
 ### core（核心模块 · 进程托管服务）
@@ -135,9 +135,9 @@
 
 ### smtp（邮件服务）
 
-- 注册流程的邮箱验证码后端，基于 Flask + waitress，监听 127.0.0.1:8008。
+- 注册流程的邮箱验证码后端，基于 pymsg（MsgServer，TCP 长度前缀帧 + JSON），监听 127.0.0.1:8008。
-- 接口：`POST /send_email`（生成 6 位验证码并经 SMTP_SSL 发送至收件人，60 秒有效）、`POST /verify_email`（校验）。
+- pymsg 路由：`send_email`（生成 6 位验证码并经 SMTP_SSL 发送，60 秒有效，发送失败自动作废）、`verify_email`（校验，通过即一次性消费防重放，兼容旧字段 `input_captcha`）、`health`。
-- 验证码存于进程内存，SMTP 账号通过 `.env` 配置（`SMTP_SERVER` / `SMTP_PORT` / `SMTP_USER` / `__PASSWORD`）。
+- 验证码存于进程内存（锁保护 + 惰性过期清理），SMTP 账号通过 `.env` 配置（`SMTP_SERVER` / `SMTP_PORT` / `SMTP_USER` / `__PASSWORD`，`__PASSWORD` 兼容别名 `SMTP_PASSWORD`）。
 
 ### aibox（独立 AI 智能体系统）
 

smtp/requirements.txt

@@ -1,8 +1,3 @@
 ../pymsg
-nuitka
-flask
-flask-cors
-flask_sqlalchemy
-#gunicorn
-waitress
 python-dotenv
+nuitka

smtp/smtp.py

@@ -1,84 +1,221 @@
-#!/usr/bin/env python3
+"""
+smtp — 基于 pymsg 协议（MsgServer）的邮箱验证码服务。
 
-from flask import Flask, request, jsonify
+替代原 Flask HTTP 接口，通过 TCP 长度前缀帧收发 JSON。
-from email.mime.text import MIMEText
+协议：每条消息为完整 JSON，``_route`` 字段指定路由名。
-from email import header
+
-from log_client import logs
+请求示例::
-from dotenv import load_dotenv
+
-import waitress
+    {"_route": "send_email", "email": "user@example.com"}          → 生成并发送验证码
-import smtplib
+    {"_route": "verify_email", "email": "...", "code": "123456"}   → 校验（一次性消费）
+    {"_route": "health"}                                           → 健康检查
+
+响应格式（与 dbv2 / captcha 一致）::
+
+    {"success": true,  "data": {...}}
+    {"success": false, "error": "..."}
+
+SMTP 账号经 .env 配置：SMTP_SERVER / SMTP_PORT / SMTP_USER / __PASSWORD
+（__PASSWORD 兼容别名 SMTP_PASSWORD）。
+"""
+
+import json
+import os
 import random
+import smtplib
+import threading
 import time
-import sys, os
+from email.mime.text import MIMEText
 
-sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+try:
-load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
+    from dotenv import load_dotenv
+    load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
+except ImportError:
+    pass  # 无 dotenv 时直接读进程环境变量
 
-app = Flask(__name__)
+from log_client import logs
-captchas: dict[str, list[str, str]] = {}
 
-@app.route("/send_email", methods=["POST"])
+# pymsg 导入
+import sys as _sys
+_sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from pymsg import MsgServer  # noqa: E402
 
-def send_email():
+# ── 配置 ────────────────────────────────────────────────────────────────
-    def build_captcha() -> str:
-        captcha: str = ""
-        for i in range(0,6):
-            digit: int = random.randint(0,9)
-            captcha += str(digit)
-        return captcha
 
-    data = request.get_json()
+SMTP_SERVER = os.environ.get("SMTP_SERVER", "")
+SMTP_PORT = int(os.environ.get("SMTP_PORT", "465"))
+SMTP_USER = os.environ.get("SMTP_USER", "")
+# 历史 .env 使用 __PASSWORD，允许 SMTP_PASSWORD 作为别名
+SMTP_PASSWORD = os.environ.get("__PASSWORD") or os.environ.get("SMTP_PASSWORD", "")
 
-    recipient: str = data.get("email")
+CODE_LEN = 6
-    subject: str = "DNFS 邮箱验证"
+CODE_TTL = 60            # 验证码有效期（秒）
-    captcha: str = build_captcha()
+MAIL_FROM = "dnfs@qsont.me"
-    content: str = f"DNFS: 您的邮箱验证码为 {captcha}，一分钟内有效。"
+DEFAULT_SUBJECT = "DNFS 邮箱验证"
+DEFAULT_CONTENT = "DNFS: 您的邮箱验证码为 {code}，一分钟内有效。"
 
-    captchas[recipient] = ["", ""]
+code_store = {}          # email -> {'code': str, 'expires_at': float}
-    captchas[recipient][0] = captcha
+STORE_LOCK = threading.Lock()
-    captchas[recipient][1] = str(int(time.time())+60)
-
-    message = MIMEText(content, "plain", "UTF-8")
-    message["From"] = "dnfs@qsont.me"
-    message["To"] = recipient
-    message["Subject"] = subject
-
-    try:
-        server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10)
-        server.login(SMTP_USER, __PASSWORD)
-        server.sendmail(SMTP_USER, [recipient], message.as_string())
-        logs.info("smtp", f"{recipient}: 验证码发送成功({captcha})")
-        return jsonify({"success": "验证码发送成功"}), 200
-
-    except Exception as err:
-        logs.error("smtp", f"{recipient}: 验证码发送失败({err})")
-        return jsonify({"error": f"验证码发送失败：{err}"}), 400
-
-    finally:
-        server.quit()
 
 
-@app.route("/verify_email", methods=["POST"])
+# ═══════════════════════════════════════════════════════════════════════
+# SmtpServer — 主服务类
+# ═══════════════════════════════════════════════════════════════════════
 
-def verify_email():
+class SmtpServer:
-    data = request.get_json()
+    """基于 pymsg 的邮箱验证码服务，通过 TCP 长连接收发 JSON。"""
 
-    for i in list(captchas.keys()):
+    def __init__(self, addr="127.0.0.1", port=8008):
-        if int(captchas[i][1]) < int(time.time()):
+        self.addr = addr
-            del captchas[i]
+        self.port = port
+        self._server = MsgServer(addr, port)
 
-    email: str = data.get("email")
+        # 路由表
-    input_captcha: str = data.get("input_captcha")
+        self._routes = {
+            "send_email": self._handle_send_email,
+            "verify_email": self._handle_verify_email,
+            "health": self._handle_health,
+        }
 
-    if captchas[email][0] == input_captcha:
+    # ── 启动 / 停止 ──────────────────────────────────────────────────
-        logs.info("smtp", f"{email}: 验证通过")
+    def start(self) -> None:
-        return jsonify({"success": "验证通过"}), 200
+        """启动服务（阻塞当前线程）。"""
-    else:
+        if not all([SMTP_SERVER, SMTP_USER, SMTP_PASSWORD]):
+            logs.error("smtp", "SMTP 配置不完整（.env 缺 SMTP_SERVER/SMTP_USER/__PASSWORD），"
+                               "send_email 将始终失败")
+            print("[smtp] 警告: SMTP 配置不完整，send_email 将失败")
+        print(f"[smtp] 监听 tcp://{self.addr}:{self.port}")
+        logs.info("smtp", f"开始监听 tcp://{self.addr}:{self.port}")
+        self._server.start(self._dispatch)
+
+    def stop(self) -> None:
+        """从另一线程停止服务。"""
+        self._server.stop()
+
+    # ── 消息调度 -------------------------------------------------------
+    def _dispatch(self, sock, addr, text):
+        """解析 JSON 中的 _route，分发给对应的 handler，并回复结果。"""
+        try:
+            req = json.loads(text)
+        except json.JSONDecodeError as e:
+            self._reply(sock, False, error=f"JSON 解析失败: {e}")
+            return
+
+        route = req.pop("_route", "")
+        handler = self._routes.get(route)
+        if handler is None:
+            self._reply(sock, False, error=f"未知路由: {route}")
+            return
+
+        try:
+            result = handler(req)
+            self._reply(sock, True, data=result)
+        except Exception as e:
+            logs.error("smtp", f"路由 {route} 处理异常: {e}")
+            self._reply(sock, False, error=str(e))
+
+    def _reply(self, sock, success, data=None, error=None):
+        resp = {"success": success}
+        if data is not None:
+            resp["data"] = data
+        if error is not None:
+            resp["error"] = error
+        self._server.reply(sock, json.dumps(resp, ensure_ascii=False))
+
+    # ═══════════════════════════════════════════════════════════════════
+    # 路由处理
+    # ═══════════════════════════════════════════════════════════════════
+
+    def _handle_send_email(self, req):
+        """生成 6 位数字验证码，存入内存（60s），经 SMTP_SSL 发送邮件。
+
+        可选字段 subject / content 覆盖默认标题与正文，content 中 {code}
+        会被替换为验证码。
+        """
+        email = str(req.get("email", "")).strip()
+        if not email or "@" not in email:
+            return {"sent": False, "error": "invalid email"}
+
+        if not all([SMTP_SERVER, SMTP_USER, SMTP_PASSWORD]):
+            logs.error("smtp", f"{email}: SMTP 配置缺失，无法发送")
+            return {"sent": False, "error": "SMTP 服务未配置"}
+
+        code = "".join(str(random.randint(0, 9)) for _ in range(CODE_LEN))
+        subject = str(req.get("subject") or DEFAULT_SUBJECT)
+        content = str(req.get("content") or DEFAULT_CONTENT).replace("{code}", code)
+
+        with STORE_LOCK:
+            code_store[email] = {'code': code, 'expires_at': time.time() + CODE_TTL}
+            # 惰性清理过期项
+            now = time.time()
+            expired = [k for k, v in code_store.items() if v['expires_at'] < now]
+            for k in expired:
+                del code_store[k]
+
+        message = MIMEText(content, "plain", "UTF-8")
+        message["From"] = MAIL_FROM
+        message["To"] = email
+        message["Subject"] = subject
+
+        # server 作用域限定在 try 内；连接失败时不会误调 quit()
+        try:
+            with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) as server:
+                server.login(SMTP_USER, SMTP_PASSWORD)
+                server.sendmail(SMTP_USER, [email], message.as_string())
+        except Exception as err:
+            # 发送失败：作废刚写入的验证码，避免"码已存但邮件没到"的不可用状态
+            with STORE_LOCK:
+                code_store.pop(email, None)
+            logs.error("smtp", f"{email}: 验证码发送失败({err})")
+            return {"sent": False, "error": f"验证码发送失败：{err}"}
+
+        logs.info("smtp", f"{email}: 验证码发送成功({code})")
+        return {"sent": True, "ttl": CODE_TTL}
+
+    def _handle_verify_email(self, req):
+        """校验验证码，验证通过即一次性消费；不存在的 email 安全返回 invalid。"""
+        email = str(req.get("email", "")).strip()
+        # 兼容旧字段 input_captcha
+        code = str(req.get("code") or req.get("input_captcha") or "").strip()
+        if not email or not code:
+            return {"valid": False, "error": "email and code required"}
+
+        with STORE_LOCK:
+            entry = code_store.get(email)
+            if entry and entry['expires_at'] < time.time():
+                del code_store[email]
+                entry = None
+            if entry and entry['code'] == code:
+                # 验证通过：一次性消费，防重放
+                del code_store[email]
+                valid = True
+            else:
+                valid = False
+
+        if valid:
+            logs.info("smtp", f"{email}: 验证通过")
+            return {"valid": True}
         logs.info("smtp", f"{email}: 验证失败")
-        return jsonify({"error": "验证失败"}), 400
+        return {"valid": False, "error": "invalid or expired code"}
 
+    def _handle_health(self, _req):
+        return {"status": "ok", "service": "smtp"}
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# 入口
+# ═══════════════════════════════════════════════════════════════════════
 
 if __name__ == "__main__":
-    logs.info("smtp", "开始监听http://127.0.0.1:8008")
+    import argparse
-    print("Listen at http://127.0.0.1:8008")
+
-    waitress.serve(app, host="127.0.0.1", port=8008)
+    parser = argparse.ArgumentParser(description="smtp — pymsg 邮箱验证码服务")
+    parser.add_argument("--addr", default="127.0.0.1", help="监听地址")
+    parser.add_argument("--port", type=int, default=8008, help="监听端口")
+    args = parser.parse_args()
+
+    server = SmtpServer(args.addr, args.port)
+    try:
+        server.start()
+    except KeyboardInterrupt:
+        server.stop()