统一配置模块

confs/.env

@@ -0,0 +1,4 @@
+SMTP_SERVER="smtp.163.com"
+SMTP_PORT=465
+SMTP_USER="18848414841@163.com"
+__PASSWORD="TAYUGjS79QZsNG37"

confs/_run.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+cd "$(dirname "$0")" || exit
+source ./venv/bin/activate
+#exec ./smtp.bin
+sleep 2
+exec python3 confs.py

confs/_set.sh

@@ -0,0 +1,7 @@
+#!/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

confs/confs.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+'''
+使用pymsg库的尝试
+'''
+
+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
+import json
+
+# 从文件加载到环境变量？
+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')
+
+
+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 SmtpServer:
+    '''SMTP服务器类'''
+
+    def __init__(self, addr="127.0.0.1", port=1234, workers=4):
+        self.addr = addr
+        self.port = port
+        self._server = MsgServer(addr, port, workers=workers)
+
+        self._route = {
+            "send_email": self._send_email,
+            "verify_email": self._verify_email
+        }
+
+    def start(self):
+        self._server.start(self._analysis)
+
+    def stop(self):
+        self._server.stop()
+
+    def _analysis(self, sock, addr, data):
+        print("""【recv】\n"""+str(addr)+'\n'+data) # 打印实际收到的消息
+
+
+        # 解包json
+        try:
+            request = json.loads(data)
+        except json.JSONDecodeError as err:
+            self._reply(sock, False, error=f"解析JSON失败: {err}")
+            return
+
+        # 路由
+        route = request.pop("_route", "")
+        func = self._route.get(route)
+        if func is None:
+            self._reply(sock, False, error=f"解析路由失败：没有找到路由{route}")
+            return
+
+        # 执行分支
+        try:
+            result = func(request)
+            self._reply(sock, True, data=result)
+        except Exception as err:
+            self._reply(sock, False, error=f"执行操作失败: {err}")
+
+    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": "验证通过"}
+        else:
+            logs.info("smtp", f"{email}: 验证失败")
+            return {"success": False, "data": "验证失败"}
+
+
+
+if __name__ == "__main__":
+    import argparse
+
+    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="监听端口")
+    parser.add_argument("--workers", type=int, default=4,
+                        help="回调并发线程数（0=同步串行模式）")
+    args = parser.parse_args()
+
+    server = SmtpServer(args.addr, args.port)
+    try:
+        server.start()
+    except KeyboardInterrupt:
+        server.stop()

confs/requirements.txt

@@ -0,0 +1,2 @@
+../pymsg
+nuitka