diff --git a/smtp/.gitignore b/smtp/.gitignore new file mode 100644 index 0000000..ae52178 --- /dev/null +++ b/smtp/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +.env/ \ No newline at end of file diff --git a/smtp/README.md b/smtp/README.md new file mode 100644 index 0000000..a5e9f2a --- /dev/null +++ b/smtp/README.md @@ -0,0 +1,16 @@ +# SMTP API 手册 +## `/send_email` +```json +{ + "email": "接收验证码的邮箱" +} +``` + +## `/verify_email` +```json +{ + "email": "刚才接收验证码的邮箱", + "input_captcha": "输入的验证码" +} + +``` \ No newline at end of file diff --git a/smtp/log_client.py b/smtp/log_client.py new file mode 100644 index 0000000..5bcf8c4 --- /dev/null +++ b/smtp/log_client.py @@ -0,0 +1,156 @@ +""" +统一日志客户端 — 各服务共用。 + +设计要点: + - 使用 urllib.request(标准库,不引入 requests) + - 异步非阻塞:内存队列 + 后台线程批量推送,不拖慢业务 + - 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 threading +import queue +import time +import urllib.request +import urllib.error +import atexit + +# ── 配置 ────────────────────────────────────────────────────────────── + +DB_LOG_URL = 'http://127.0.0.1:8005/log/ingest' + +_BATCH_SIZE = 50 # 队列满 50 条时 flush +_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush +_TIMEOUT = 5 # HTTP 请求超时(秒) + + +class LogClient: + """异步批量日志推送客户端。""" + + def __init__(self, url=DB_LOG_URL, batch_size=_BATCH_SIZE, + flush_interval=_FLUSH_INTERVAL): + self._url = url + self._batch_size = batch_size + self._flush_interval = flush_interval + self._queue = queue.Queue() + self._stop = 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 = [] + 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) + + # ── 内部实现 ──────────────────────────────────────────────────────── + + def _flush_loop(self): + """后台线程:积攒到 batch_size 或超时 flush_interval 时发送。""" + batch = [] + last_flush = time.monotonic() + + while not self._stop.is_set(): + timeout = max(0, self._flush_interval - (time.monotonic() - last_flush)) + try: + item = self._queue.get(timeout=timeout) + batch.append(item) + # 继续非阻塞地取,直到满 batch_size + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + # 【核心修复 1】:如果没有新日志,必须重置上次刷新时间。 + # 否则下次 timeout 会变成 0,导致疯狂空转。 + last_flush = time.monotonic() + # 【核心修复 2】:跳出本次循环,避免重复执行下面的 `if should_flush` 判断 + continue + + # 满了或超时了就发 + should_flush = ( + len(batch) >= self._batch_size or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush: + self._send(batch) + batch = [] + last_flush = time.monotonic() + + # 退出前 flush 剩余 + if batch: + self._send(batch) + + def _send(self, batch): + """发送一批日志到 db 服务。失败时静默丢弃。""" + data = json.dumps({'logs': batch}).encode('utf-8') + req = urllib.request.Request( + self._url, data=data, + headers={'Content-Type': 'application/json'}, + method='POST' + ) + try: + with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: + resp.read() # 消费响应体 + except (urllib.error.URLError, urllib.error.HTTPError, OSError): + # fail-safe:日志服务不可用,静默丢弃,不影响业务 + pass + except Exception: + pass + + def _shutdown(self): + """进程退出时 flush 剩余日志。""" + self._stop.set() + self.flush() + + +# ── 全局单例 ────────────────────────────────────────────────────────── + +logs = LogClient() diff --git a/smtp/smtp.py b/smtp/smtp.py index c32697b..816accc 100644 --- a/smtp/smtp.py +++ b/smtp/smtp.py @@ -1,62 +1,84 @@ -from flask import Flask,app -from urllib import request +#!/usr/bin/env python3 + +from flask import Flask, request, jsonify from email.mime.text import MIMEText -from email.header import Header +from email import header +from log_client import logs +import waitress import smtplib import random -from waitress import serve +import time -verification_code = 0 +app = Flask(__name__) +captchas: dict[str, list[str, str]] = {} -@app.route('/send_email', methods=["POST"]) +@app.route("/send_email", methods=["POST"]) def send_email(): + def build_captcha() -> str: + captcha: str = "" + for i in range(0,4): + digit: int = random.randint(0,9) + captcha += str(digit) + return captcha + + SMTP_SERVER: str = "" + SMTP_PORT: str = 0 + SMTP_USER: str = "" + __PASSWORD: str = "" + data = request.get_json() - reciver_email = data.get('email') - SMTP_SERVER = '' - SMTP_PORT = 587 - SENDER_EMAIL = '' - __PASS = '' + recipient: str = data.get("email") + subject: str = "DNFS 邮箱验证" + captcha: str = build_captcha() + content: str = f"DNFS: 您的邮箱验证码为 {captcha},一分钟内有效。" - verification_code = random.randint(1000,9999) + captchas[recipient] = ["", ""] + captchas[recipient][0] = captcha + captchas[recipient][1] = str(int(time.time())+60) - subject = "DNFS 验证码" - content = f"DNFS: 您的验证码为{str(verification_code)}" - - email_message = MIMEText(content, 'plain', 'utf-8') - email_message['From'] = Header(SENDER_EMAIL, 'utf-8') - email_message['To'] = Header(reciver_email, 'utf-8') - email_message['Subject'] = Header(subject) + message = MIMEText(content, "plain", "UTF-8") + message["From"] = "dnfs@qsont.me" + message["To"] = recipient + message["Subject"] = subject try: - server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) - server.starttls() - server.login(SENDER_EMAIL, __PASS) - server.sendmail(SENDER_EMAIL, [reciver_email], email_message.as_string()) - return 200 + 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: - raise err - return 400 + logs.error("smtp", f"{recipient}: 验证码发送失败({err})") + return jsonify({"error": f"验证码发送失败:{err}"}), 400 finally: server.quit() -@app.route('/get_verification_code', method=["POST"]) -def get_verification_code(): - data=request.get_json() +@app.route("/verify_email", methods=["POST"]) - input_verification_code = data.get('verification_code') +def verify_email(): + data = request.get_json() - if verification_code == input_verification_code: - return 200 + 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("input_captcha") + + if captchas[email][0] == input_captcha: + logs.info("smtp", f"{email}: 验证通过") + return jsonify({"success": "验证通过"}), 200 else: - return 400 + logs.info("smtp", f"{email}: 验证失败") + return jsonify({"error": "验证失败"}), 400 + -ADDR = '127.0.0.1' -PORT = 8007 if __name__ == "__main__": - print(f'Listen at http://{ADDR}:{PORT}') - serve(app, host=ADDR, port=PORT) \ No newline at end of file + logs.info("smtp", "开始监听http://127.0.0.1:1234") + print("Listen at http://127.0.0.1:1234") + waitress.serve(app, host="127.0.0.1", port=1234) \ No newline at end of file