From 015a63043d6ffc931ad7db3561e087b126c2f081 Mon Sep 17 00:00:00 2001 From: juanjuanjuanidea Date: Fri, 24 Jul 2026 13:57:34 +0800 Subject: [PATCH] feat: complete ni-mail rewrite with Graph API, IMAP, OTP extractor, Web UI and Docker deployment --- .dockerignore | 6 + .env.example | 5 + .gitignore | 6 + Dockerfile | 28 + app/__init__.py | 30 + app/api/accounts.py | 37 + app/api/emails.py | 53 + app/db.py | 3 + app/models.py | 19 + app/services/graph.py | 422 +++++++ app/services/http.py | 12 + app/services/imap.py | 780 ++++++++++++ app/services/verification.py | 598 +++++++++ app/templates/index.html | 2225 ++++++++++++++++++++++++++++++++++ docker-compose.yml | 29 + requirements.txt | 5 + scripts/entrypoint.sh | 7 + 17 files changed, 4265 insertions(+) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 app/__init__.py create mode 100644 app/api/accounts.py create mode 100644 app/api/emails.py create mode 100644 app/db.py create mode 100644 app/models.py create mode 100644 app/services/graph.py create mode 100644 app/services/http.py create mode 100644 app/services/imap.py create mode 100644 app/services/verification.py create mode 100644 app/templates/index.html create mode 100644 docker-compose.yml create mode 100644 requirements.txt create mode 100644 scripts/entrypoint.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..fe54b04 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +*.db +.env +.git/ +data/ diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..c864d03 --- /dev/null +++ b/.env.example @@ -0,0 +1,5 @@ +CLIENT_ID=your-azure-app-client-id +SECRET_KEY=ni-mail-secret-key-12345 +PROXY_URL=http://host.docker.internal:7890 +TZ=Asia/Shanghai +PORT=8080 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..29b208f --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +*.db +.env +data/ +venv/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..b8ba2e4 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-slim + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates tini wget && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt + +COPY app/ ./app/ +COPY scripts/entrypoint.sh ./entrypoint.sh +RUN chmod +x ./entrypoint.sh + +RUN groupadd -r nimail && useradd -r -g nimail -d /app/data nimail && mkdir -p /app/data && chown nimail:nimail /app/data + +USER nimail + +EXPOSE 8080 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 CMD wget -q -O /dev/null "http://127.0.0.1:8080/health" || exit 1 + +ENV FLASK_APP=app:create_app() +ENV DATA_DIR=/app/data +ENV PORT=8080 + +STOPSIGNAL SIGTERM +ENTRYPOINT ["/usr/bin/tini", "--"] +CMD ["./entrypoint.sh"] diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..f26aaaa --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,30 @@ +import os +from flask import Flask, render_template, jsonify +from app.db import db +from app.api import accounts, emails + +def create_app(): + app = Flask(__name__) + + db_path = os.getenv("DATABASE_URL", f"sqlite:///{os.path.join(os.getcwd(), 'ni_mail.db')}") + app.config["SQLALCHEMY_DATABASE_URI"] = db_path + app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False + app.config["SECRET_KEY"] = os.getenv("SECRET_KEY", "ni-mail-secret-key") + + db.init_app(app) + + app.register_blueprint(accounts.bp) + app.register_blueprint(emails.bp) + + @app.route("/") + def index(): + return render_template("index.html") + + @app.route("/health") + def health(): + return jsonify({"status": "ok", "app": "ni-mail"}) + + with app.app_context(): + db.create_all() + + return app diff --git a/app/api/accounts.py b/app/api/accounts.py new file mode 100644 index 0000000..d52510c --- /dev/null +++ b/app/api/accounts.py @@ -0,0 +1,37 @@ +from flask import Blueprint, jsonify, request +from app.db import db +from app.models import Account + +bp = Blueprint("accounts", __name__, url_prefix="/api/accounts") + +@bp.get("/") +def list_accounts(): + accounts = Account.query.all() + return jsonify([a.to_dict() for a in accounts]) + +@bp.post("/") +def add_account(): + data = request.json or {} + email = data.get("email") + if not email: + return jsonify({"error": "email is required"}), 400 + + acct = Account.query.filter_by(email=email).first() + if not acct: + acct = Account(email=email) + + acct.method = data.get("method", "graph") + acct.client_id = data.get("client_id") + acct.refresh_token = data.get("refresh_token") + acct.password = data.get("password") + + db.session.add(acct) + db.session.commit() + return jsonify(acct.to_dict()), 201 + +@bp.delete("/") +def delete_account(acct_id): + acct = Account.query.get_or_404(acct_id) + db.session.delete(acct) + db.session.commit() + return jsonify({"success": True}) diff --git a/app/api/emails.py b/app/api/emails.py new file mode 100644 index 0000000..df1f314 --- /dev/null +++ b/app/api/emails.py @@ -0,0 +1,53 @@ +from flask import Blueprint, jsonify, request +from app.models import Account +from app.services import graph, imap, verification +import os + +bp = Blueprint("emails", __name__, url_prefix="/api/emails") + +@bp.get("/") +def get_emails(): + email = request.args.get("email") + if not email: + return jsonify({"error": "email query param required"}), 400 + + acct = Account.query.filter_by(email=email).first() + if not acct: + return jsonify({"error": "account not found"}), 404 + + proxy_url = os.getenv("PROXY_URL") + messages = [] + + if acct.method == "graph": + token_res = graph.get_access_token_graph_result( + client_id=acct.client_id, + refresh_token=acct.refresh_token, + proxy_url=proxy_url + ) + if isinstance(token_res, dict) and "access_token" in token_res: + token = token_res["access_token"] + messages = graph.fetch_messages(token, top=15, proxy_url=proxy_url) + else: + # Fallback to IMAP + messages = imap.fetch_inbox_messages( + email=acct.email, + password=acct.password or acct.refresh_token, + limit=15, + proxy_url=proxy_url + ) + + results = [] + for msg in messages: + body = msg.get("body", {}).get("content", "") or msg.get("body_text", "") or "" + code, link = verification.extract_code_and_link(body) + results.append({ + "id": msg.get("id"), + "subject": msg.get("subject"), + "from": msg.get("from", {}).get("emailAddress", {}).get("address", ""), + "received_at": msg.get("receivedDateTime"), + "otp_code": code, + "link": link, + "body_preview": body[:300] + }) + + return jsonify(results) diff --git a/app/db.py b/app/db.py new file mode 100644 index 0000000..f0b13d6 --- /dev/null +++ b/app/db.py @@ -0,0 +1,3 @@ +from flask_sqlalchemy import SQLAlchemy + +db = SQLAlchemy() diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..e2504fa --- /dev/null +++ b/app/models.py @@ -0,0 +1,19 @@ +from app.db import db + +class Account(db.Model): + id = db.Column(db.Integer, primary_key=True) + email = db.Column(db.String(128), unique=True, nullable=False) + method = db.Column(db.String(10), default="graph") # graph | imap + client_id = db.Column(db.String(64)) + refresh_token = db.Column(db.Text) + password = db.Column(db.String(128)) + created_at = db.Column(db.DateTime, default=db.func.now()) + + def to_dict(self): + return { + "id": self.id, + "email": self.email, + "method": self.method, + "client_id": self.client_id, + "created_at": str(self.created_at) if self.created_at else None + } diff --git a/app/services/graph.py b/app/services/graph.py new file mode 100644 index 0000000..12224a3 --- /dev/null +++ b/app/services/graph.py @@ -0,0 +1,422 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import requests + +# from app.errors import build_error_payload +from app.services.http import get_response_details + +# Token 端点 +TOKEN_URL_TEMPLATE = "https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token" +TOKEN_URL_GRAPH = TOKEN_URL_TEMPLATE.format(tenant="common") +DEFAULT_GRAPH_SCOPE = "https://graph.microsoft.com/.default" +GRAPH_MAIL_READ_SCOPES = ("Mail.Read", "Mail.ReadWrite") + +# Graph API 返回 401 时表示账号授权失效（与 token endpoint 失败不同） +GRAPH_AUTH_EXPIRED_STATUS = 401 + + +def build_proxies(proxy_url: str) -> Optional[Dict[str, str]]: + """构建 requests 的 proxies 参数""" + if not proxy_url: + return None + return {"http": proxy_url, "https": proxy_url} + + +def build_token_url(tenant: str | None = None) -> str: + """按 tenant 生成 Microsoft OAuth token endpoint。""" + normalized_tenant = (tenant or "common").strip() or "common" + return TOKEN_URL_TEMPLATE.format(tenant=normalized_tenant) + + +def get_access_token_graph_result(client_id: str, refresh_token: str, proxy_url: str = None) -> Dict[str, Any]: + """获取 Graph API access_token（包含错误详情）""" + try: + proxies = build_proxies(proxy_url) + res = requests.post( + TOKEN_URL_GRAPH, + data={ + "client_id": client_id, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "scope": DEFAULT_GRAPH_SCOPE, + }, + timeout=30, + proxies=proxies, + ) + + if res.status_code != 200: + details = get_response_details(res) + return { + "success": False, + "error": build_error_payload( + "GRAPH_TOKEN_FAILED", + "获取访问令牌失败", + "GraphAPIError", + res.status_code, + details, + ), + } + + payload = res.json() + access_token = payload.get("access_token") + if not access_token: + return { + "success": False, + "error": build_error_payload( + "GRAPH_TOKEN_MISSING", + "获取访问令牌失败", + "GraphAPIError", + res.status_code, + payload, + ), + } + + # 根据 Microsoft Learn 文档：refresh token 可能会在每次使用时"自我替换"，应保存新的 refresh_token（如有）。 + new_refresh_token = payload.get("refresh_token") + return { + "success": True, + "access_token": access_token, + "refresh_token": new_refresh_token, + "new_refresh_token": new_refresh_token, + "scope": payload.get("scope", ""), + } + except Exception as exc: + return { + "success": False, + "error": build_error_payload( + "GRAPH_TOKEN_EXCEPTION", + "获取访问令牌失败", + type(exc).__name__, + 500, + str(exc), + ), + } + + +def has_mail_read_permission(scope: Any) -> bool: + scope_str = str(scope or "") + return any(mail_scope in scope_str for mail_scope in GRAPH_MAIL_READ_SCOPES) + + +def get_access_token_graph(client_id: str, refresh_token: str, proxy_url: str = None) -> Optional[str]: + """获取 Graph API access_token""" + result = get_access_token_graph_result(client_id, refresh_token, proxy_url) + if result.get("success"): + return result.get("access_token") + return None + + +def get_emails_graph( + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 20, + proxy_url: str = None, +) -> Dict[str, Any]: + """使用 Graph API 获取邮件列表（支持分页和文件夹选择）""" + token_result = get_access_token_graph_result(client_id, refresh_token, proxy_url) + if not token_result.get("success"): + return {"success": False, "error": token_result.get("error")} + + access_token = token_result.get("access_token") + scope = token_result.get("scope", "") + if not has_mail_read_permission(scope): + return { + "success": False, + "auth_expired": True, + "no_mail_permission": True, + "error": build_error_payload( + "NO_MAIL_PERMISSION", + "此账号未授予邮件读取权限 (scope 中不含 Mail.Read)", + "PermissionError", + 403, + f"scope={scope}", + ), + } + + try: + folder_map = { + "inbox": "inbox", + "junkemail": "junkemail", + "deleteditems": "deleteditems", + "trash": "deleteditems", + } + folder_name = folder_map.get((folder or "").lower(), "inbox") + + url = f"https://graph.microsoft.com/v1.0/me/mailFolders/{folder_name}/messages" + params = { + "$top": top, + "$skip": skip, + "$select": "id,subject,from,receivedDateTime,isRead,hasAttachments,bodyPreview", + "$orderby": "receivedDateTime desc", + } + headers = { + "Authorization": f"Bearer {access_token}", + "Prefer": "outlook.body-content-type='text'", + } + + proxies = build_proxies(proxy_url) + res = requests.get(url, headers=headers, params=params, timeout=30, proxies=proxies) + + if res.status_code != 200: + details = get_response_details(res) + return { + "success": False, + "auth_expired": res.status_code == GRAPH_AUTH_EXPIRED_STATUS, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "获取邮件失败，请检查账号配置", + "GraphAPIError", + res.status_code, + details, + ), + } + + return { + "success": True, + "emails": res.json().get("value", []), + "new_refresh_token": token_result.get("refresh_token"), + } + except Exception as exc: + return { + "success": False, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "获取邮件失败，请检查账号配置", + type(exc).__name__, + 500, + str(exc), + ), + } + + +def get_email_detail_graph( + client_id: str, + refresh_token: str, + message_id: str, + proxy_url: str = None, +) -> Optional[Dict]: + """使用 Graph API 获取邮件详情""" + access_token = get_access_token_graph(client_id, refresh_token, proxy_url) + if not access_token: + return None + + try: + url = f"https://graph.microsoft.com/v1.0/me/messages/{message_id}" + params = { + "$select": "id,subject,from,toRecipients,ccRecipients,receivedDateTime,isRead,hasAttachments,body,bodyPreview" + } + headers = { + "Authorization": f"Bearer {access_token}", + "Prefer": "outlook.body-content-type='html'", + } + + proxies = build_proxies(proxy_url) + res = requests.get(url, headers=headers, params=params, timeout=30, proxies=proxies) + + if res.status_code != 200: + return None + + return res.json() + except Exception: + return None + + +def get_email_raw_graph( + client_id: str, + refresh_token: str, + message_id: str, + proxy_url: str = None, +) -> Optional[str]: + """使用 Graph API 获取邮件 MIME RAW 内容。""" + access_token = get_access_token_graph(client_id, refresh_token, proxy_url) + if not access_token: + return None + + try: + url = f"https://graph.microsoft.com/v1.0/me/messages/{message_id}/$value" + headers = { + "Authorization": f"Bearer {access_token}", + } + + proxies = build_proxies(proxy_url) + res = requests.get(url, headers=headers, timeout=30, proxies=proxies) + + if res.status_code != 200: + return None + + res.encoding = res.encoding or "utf-8" + return res.text + except Exception: + return None + + +def test_refresh_token(client_id: str, refresh_token: str, proxy_url: str = None) -> tuple[bool, str | None]: + """测试 refresh token 是否有效，返回 (是否成功, 错误信息)""" + ok, err, _new_refresh_token = test_refresh_token_with_rotation(client_id, refresh_token, proxy_url) + return ok, err + + +def test_refresh_token_with_rotation( + client_id: str, + refresh_token: str, + proxy_url: str = None, + *, + tenant: str = "common", + scope: str = DEFAULT_GRAPH_SCOPE, + max_retries: int = 3, +) -> tuple[bool, str | None, str | None]: + """测试 refresh token 是否有效；如服务端返回新的 refresh_token，则一并返回（用于滚动更新）。 + 支持指数退避重试，遇到 429 时优先读取 Retry-After 头。""" + import time + + proxies = build_proxies(proxy_url) + resolved_scope = (scope or DEFAULT_GRAPH_SCOPE).strip() or DEFAULT_GRAPH_SCOPE + url = build_token_url(tenant) + data = { + "client_id": client_id, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "scope": resolved_scope, + } + + last_error_msg = None + for attempt in range(max_retries + 1): + try: + res = requests.post(url, data=data, timeout=15, proxies=proxies) + + if res.status_code == 200: + try: + payload = res.json() + except Exception: + payload = {} + new_refresh_token = payload.get("refresh_token") + return True, None, new_refresh_token + + # 429 限流：读取 Retry-After 并退避 + if res.status_code == 429: + retry_after = None + try: + retry_after = int(res.headers.get("Retry-After", 0)) + except Exception: + retry_after = None + wait = retry_after if retry_after else (2**attempt) + last_error_msg = f"请求被限流 (429)，{wait}s 后重试" + if attempt < max_retries: + time.sleep(wait) + continue + + try: + error_data = res.json() + except Exception: + error_data = {} + error_msg = None + if isinstance(error_data, dict): + error_msg = error_data.get("error_description") or error_data.get("error") + if not error_msg: + details = get_response_details(res) + error_msg = str(details)[:800] if details is not None else "未知错误" + last_error_msg = str(error_msg) + # 非 429 的明确错误响应（如 400/401/403）不需要重试，直接返回 + return False, last_error_msg, None + except Exception as e: + last_error_msg = f"请求异常: {str(e)}" + if attempt < max_retries: + time.sleep(2**attempt) + continue + return False, last_error_msg, None + + return False, last_error_msg or "请求失败", None + + +def delete_emails_graph( + client_id: str, + refresh_token: str, + message_ids: List[str], + proxy_url: str = None, +) -> Dict[str, Any]: + """通过 Graph API 批量删除邮件（永久删除）""" + token_result = get_access_token_graph_result(client_id, refresh_token, proxy_url) + if not token_result.get("success"): + return {"success": False, "error": token_result.get("error")} + + access_token = token_result.get("access_token") + if not access_token: + return { + "success": False, + "error": build_error_payload( + "GRAPH_TOKEN_FAILED", + "获取访问令牌失败", + "GraphAPIError", + 500, + "empty_access_token", + ), + } + + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + # Graph API batch 请求每次最多 20 + batch_size = 20 + success_count = 0 + failed_count = 0 + errors: List[str] = [] + + for i in range(0, len(message_ids), batch_size): + batch = message_ids[i : i + batch_size] + + batch_requests = [] + for idx, msg_id in enumerate(batch): + batch_requests.append({"id": str(idx), "method": "DELETE", "url": f"/me/messages/{msg_id}"}) + + try: + proxies = build_proxies(proxy_url) + response = requests.post( + "https://graph.microsoft.com/v1.0/$batch", + headers=headers, + json={"requests": batch_requests}, + timeout=30, + proxies=proxies, + ) + + if response.status_code == 200: + results = response.json().get("responses", []) + for res in results: + if res.get("status") in [200, 204]: + success_count += 1 + else: + failed_count += 1 + try: + errors.append(f"Msg ID: {batch[int(res['id'])]}, Status: {res.get('status')}") + except Exception: + errors.append(f"Status: {res.get('status')}") + else: + failed_count += len(batch) + errors.append(f"Batch request failed: {response.text}") + except Exception as e: + failed_count += len(batch) + errors.append(f"Network error: {str(e)}") + + result: Dict[str, Any] = { + "success": success_count > 0, + "partial_success": success_count > 0 and failed_count > 0, + "success_count": success_count, + "failed_count": failed_count, + "errors": errors, + } + + if not result["success"]: + result["error"] = build_error_payload( + "EMAIL_DELETE_FAILED", + "删除邮件失败", + "GraphAPIError", + 502, + {"failed_count": failed_count, "errors": errors[:10]}, + ) + + return result diff --git a/app/services/http.py b/app/services/http.py new file mode 100644 index 0000000..8b80316 --- /dev/null +++ b/app/services/http.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from typing import Any + +import requests + + +def get_response_details(response: requests.Response) -> Any: + try: + return response.json() + except Exception: + return response.text or response.reason diff --git a/app/services/imap.py b/app/services/imap.py new file mode 100644 index 0000000..b4ccfa9 --- /dev/null +++ b/app/services/imap.py @@ -0,0 +1,780 @@ +from __future__ import annotations + +import email +import hashlib +import imaplib +import logging +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from email.header import decode_header +from typing import Any, Dict, List, Optional + +import requests + +# from app.errors import build_error_payload +from outlook_web.services.graph import get_access_token_graph +from app.services.http import get_response_details + +_LOGGER = logging.getLogger(__name__) + +# Token 端点 +TOKEN_URL_IMAP = "https://login.microsoftonline.com/consumers/oauth2/v2.0/token" + +# IMAP 服务器配置 +IMAP_SERVER_NEW = "outlook.live.com" +IMAP_PORT = 993 + +_token_cache: Dict[str, tuple] = {} +_token_cache_lock = threading.Lock() + + +def decode_header_value(header_value: str) -> str: + """解码邮件头字段""" + if not header_value: + return "" + try: + decoded_parts = decode_header(str(header_value)) + decoded_string = "" + for part, charset in decoded_parts: + if isinstance(part, bytes): + try: + decoded_string += part.decode(charset if charset else "utf-8", "replace") + except (LookupError, UnicodeDecodeError): + decoded_string += part.decode("utf-8", "replace") + else: + decoded_string += str(part) + return decoded_string + except Exception: + return str(header_value) if header_value else "" + + +def get_email_body(msg) -> str: + """提取邮件正文 + + 优先返回 text/plain，但如果内容太短（<20字符），则回退到 text/html。 + 这解决了 Figma 等服务的邮件问题：它们的 text/plain 部分可能几乎为空， + 而正文内容都在 text/html 部分。 + """ + plain_text = "" + html_text = "" + + if msg.is_multipart(): + for part in msg.walk(): + content_type = part.get_content_type() + content_disposition = str(part.get("Content-Disposition", "")) + + if "attachment" in content_disposition: + continue + + if content_type == "text/plain" and not plain_text: + try: + payload = part.get_payload(decode=True) + charset = part.get_content_charset() or "utf-8" + plain_text = payload.decode(charset, errors="replace") + except Exception: + continue + elif content_type == "text/html" and not html_text: + try: + payload = part.get_payload(decode=True) + charset = part.get_content_charset() or "utf-8" + html_text = payload.decode(charset, errors="replace") + except Exception: + continue + + if plain_text and html_text: + break + else: + try: + payload = msg.get_payload(decode=True) + charset = msg.get_content_charset() or "utf-8" + content = payload.decode(charset, errors="replace") + if msg.get_content_type() == "text/html": + html_text = content + else: + plain_text = content + except Exception: + plain_text = str(msg.get_payload()) + + # 如果 text/plain 太短（<20字符），回退到 text/html + # 这解决了 Figma 等服务的邮件问题 + if len(plain_text.strip()) >= 20: + return plain_text + return html_text or plain_text + + +def get_email_body_and_type(msg) -> tuple: + """提取邮件正文和类型（用于需要区分 HTML/Text 的场景） + + 返回 (body, body_type) 元组： + - body: 邮件正文内容 + - body_type: "html" 或 "text" + """ + plain_text = "" + html_text = "" + + if msg.is_multipart(): + for part in msg.walk(): + content_type = part.get_content_type() + content_disposition = str(part.get("Content-Disposition", "")) + + if "attachment" in content_disposition: + continue + + if content_type == "text/plain" and not plain_text: + try: + payload = part.get_payload(decode=True) + charset = part.get_content_charset() or "utf-8" + plain_text = payload.decode(charset, errors="replace") + except Exception: + continue + elif content_type == "text/html" and not html_text: + try: + payload = part.get_payload(decode=True) + charset = part.get_content_charset() or "utf-8" + html_text = payload.decode(charset, errors="replace") + except Exception: + continue + + if plain_text and html_text: + break + else: + try: + payload = msg.get_payload(decode=True) + charset = msg.get_content_charset() or "utf-8" + content = payload.decode(charset, errors="replace") + if msg.get_content_type() == "text/html": + html_text = content + else: + plain_text = content + except Exception: + plain_text = str(msg.get_payload()) + + # 如果 text/plain 太短（<20字符），回退到 text/html + if len(plain_text.strip()) >= 20: + return plain_text, "text" + return html_text or plain_text, "html" if html_text else "text" + + +def _select_folder(connection, folder: str) -> Optional[str]: + folder_map = { + "inbox": ["INBOX"], + "junk": ["Junk", "Junk Email", "Spam", "垃圾邮件"], + "junkemail": ["Junk", "Junk Email", "Spam", "垃圾邮件"], + "deleteditems": ["Deleted", "Deleted Items", "Trash", "已删除邮件"], + "trash": ["Deleted", "Deleted Items", "Trash", "已删除邮件"], + } + candidates = folder_map.get((folder or "").lower(), [folder or "INBOX"]) + for candidate in candidates: + for select_target in (f'"{candidate}"', candidate): + try: + status, _ = connection.select(select_target, readonly=True) + if status == "OK": + return candidate + except Exception: + continue + return None + + +def _get_html_body(msg) -> str: + if msg.is_multipart(): + for part in msg.walk(): + if part.get_content_type() == "text/html": + payload = part.get_payload(decode=True) + if payload: + charset = part.get_content_charset() or "utf-8" + return payload.decode(charset, errors="replace") + else: + if msg.get_content_type() == "text/html": + payload = msg.get_payload(decode=True) + if payload: + return payload.decode(msg.get_content_charset() or "utf-8", errors="replace") + return "" + + +def _parse_batch_fetch_response(all_data: list) -> List[tuple]: + results = [] + for item in all_data: + header = None + raw_email = None + + if isinstance(item, tuple) and len(item) == 2: + first, second = item + if isinstance(first, (bytes, bytearray)) and isinstance(second, (bytes, bytearray)): + header = bytes(first) + raw_email = bytes(second) + elif isinstance(first, tuple) and len(first) == 2: + nested_header, nested_raw = first + if isinstance(nested_header, (bytes, bytearray)) and isinstance(nested_raw, (bytes, bytearray)): + header = bytes(nested_header) + raw_email = bytes(nested_raw) + + if not isinstance(header, (bytes, bytearray)) or not isinstance(raw_email, (bytes, bytearray)): + continue + + msg_id_str = header.split(b" ", 1)[0].decode("ascii", errors="ignore").strip() + if not msg_id_str: + continue + results.append((msg_id_str, raw_email)) + return results + + +def _make_cache_key(client_id: str, refresh_token: str) -> str: + rt_hash = hashlib.sha256(refresh_token.encode("utf-8")).hexdigest()[:16] + return f"{client_id}:{rt_hash}" + + +def clear_imap_token_cache(client_id: str = None) -> None: + with _token_cache_lock: + if client_id is None: + _token_cache.clear() + else: + keys_to_remove = [k for k in _token_cache if k.startswith(f"{client_id}:")] + for key in keys_to_remove: + del _token_cache[key] + + +def get_access_token_imap_result(client_id: str, refresh_token: str) -> Dict[str, Any]: + """获取 IMAP access_token（包含错误详情）""" + cache_key = _make_cache_key(client_id, refresh_token) + with _token_cache_lock: + cached = _token_cache.get(cache_key) + if cached: + access_token, expires_at = cached + if time.monotonic() < expires_at: + return {"success": True, "access_token": access_token} + + try: + res = requests.post( + TOKEN_URL_IMAP, + data={ + "client_id": client_id, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "scope": "https://outlook.office.com/IMAP.AccessAsUser.All offline_access", + }, + timeout=30, + ) + + if res.status_code != 200: + details = get_response_details(res) + return { + "success": False, + "error": build_error_payload( + "IMAP_TOKEN_FAILED", + "获取访问令牌失败", + "IMAPError", + res.status_code, + details, + ), + } + + payload = res.json() + access_token = payload.get("access_token") + if not access_token: + return { + "success": False, + "error": build_error_payload( + "IMAP_TOKEN_MISSING", + "获取访问令牌失败", + "IMAPError", + res.status_code, + payload, + ), + } + + expires_in = int(payload.get("expires_in", 3599)) + ttl = max(0, expires_in - 60) + with _token_cache_lock: + _token_cache[cache_key] = (access_token, time.monotonic() + ttl) + + return {"success": True, "access_token": access_token} + except Exception as exc: + return { + "success": False, + "error": build_error_payload( + "IMAP_TOKEN_EXCEPTION", + "获取访问令牌失败", + type(exc).__name__, + 500, + str(exc), + ), + } + + +def get_access_token_imap(client_id: str, refresh_token: str) -> Optional[str]: + """获取 IMAP access_token""" + result = get_access_token_imap_result(client_id, refresh_token) + if result.get("success"): + return result.get("access_token") + return None + + +def get_emails_imap( + account: str, + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 20, +) -> Dict[str, Any]: + """使用 IMAP 获取邮件列表（支持分页和文件夹选择）- 默认使用新版服务器""" + return get_emails_imap_with_server(account, client_id, refresh_token, folder, skip, top, IMAP_SERVER_NEW) + + +def get_emails_imap_with_server( + account: str, + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 20, + server: str = IMAP_SERVER_NEW, +) -> Dict[str, Any]: + """使用 IMAP 获取邮件列表（支持分页、文件夹选择和服务器选择）""" + token_result = get_access_token_imap_result(client_id, refresh_token) + if not token_result.get("success"): + return {"success": False, "error": token_result.get("error")} + + access_token = token_result.get("access_token") + + connection = None + try: + connection = imaplib.IMAP4_SSL(server, IMAP_PORT) + auth_string = f"user={account}\1auth=Bearer {access_token}\1\1".encode("utf-8") + connection.authenticate("XOAUTH2", lambda x: auth_string) + + selected_folder = _select_folder(connection, folder) + + if not selected_folder: + try: + status, folder_list = connection.list() + available_folders = [] + if status == "OK" and folder_list: + for folder_item in folder_list: + if isinstance(folder_item, bytes): + available_folders.append(folder_item.decode("utf-8", errors="ignore")) + else: + available_folders.append(str(folder_item)) + + error_details = { + "last_error": "select folder failed", + "tried_folder": folder, + "available_folders": available_folders[:10], + } + except Exception: + error_details = { + "last_error": "select folder failed", + "tried_folder": folder, + } + + return { + "success": False, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "无法访问文件夹，请检查账号配置", + "IMAPSelectError", + 500, + error_details, + ), + } + + status, messages = connection.search(None, "ALL") + if status != "OK": + _LOGGER.debug( + "[PERF] imap_search | account=%s | server=%s | folder=%s | status=%s (非OK)", + account, + server, + selected_folder, + status, + ) + return { + "success": False, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "获取邮件失败，请检查账号配置", + "IMAPSearchError", + 500, + f"search status={status}", + ), + } + if not messages or not messages[0]: + _LOGGER.debug( + "[PERF] imap_search | account=%s | server=%s | folder=%s | total=0 (空信箱)", + account, + server, + selected_folder, + ) + return {"success": True, "emails": []} + + message_ids = messages[0].split() + total = len(message_ids) + start_idx = max(0, total - skip - top) + end_idx = total - skip + + _LOGGER.debug( + "[PERF] imap_search | account=%s | server=%s | folder=%s | total=%d | skip=%d | top=%d | slice=[%d:%d]", + account, + server, + selected_folder, + total, + skip, + top, + start_idx, + end_idx, + ) + + if start_idx >= end_idx: + return {"success": True, "emails": []} + + paged_ids = message_ids[start_idx:end_idx][::-1] + emails_data = [] + + ids_str = b",".join(paged_ids) + status, all_data = connection.fetch(ids_str, "(RFC822)") + if status != "OK": + _LOGGER.debug( + "[PERF] imap_fetch | account=%s | batch fetch失败 status=%s", + account, + status, + ) + return {"success": True, "emails": emails_data} + + for msg_id_str, raw_email in _parse_batch_fetch_response(all_data or []): + try: + msg = email.message_from_bytes(raw_email) + body_preview = get_email_body(msg) + emails_data.append( + { + "id": msg_id_str, + "subject": decode_header_value(msg.get("Subject", "无主题")), + "from": decode_header_value(msg.get("From", "未知发件人")), + "date": msg.get("Date", "未知时间"), + "body_preview": (body_preview[:200] + "..." if len(body_preview) > 200 else body_preview), + } + ) + except Exception as fetch_err: + _LOGGER.debug( + "[PERF] imap_fetch | account=%s | msg_id=%s | 解析失败: %s", + account, + msg_id_str, + fetch_err, + ) + continue + + _LOGGER.debug( + "[PERF] imap_result | account=%s | server=%s | fetched=%d / requested=%d", + account, + server, + len(emails_data), + len(paged_ids), + ) + return {"success": True, "emails": emails_data} + except Exception as exc: + return { + "success": False, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "获取邮件失败，请检查账号配置", + type(exc).__name__, + 500, + str(exc), + ), + } + finally: + if connection: + try: + connection.logout() + except Exception: + pass + + +def fetch_and_detail_imap_with_server( + account: str, + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 1, + server: str = IMAP_SERVER_NEW, +) -> Dict[str, Any]: + """一次 IMAP 连接完成邮件列表 + 最新一封详情。""" + token_result = get_access_token_imap_result(client_id, refresh_token) + if not token_result.get("success"): + return { + "success": False, + "error": token_result.get("error"), + "emails": [], + "detail": None, + } + + access_token = token_result["access_token"] + connection = None + + try: + connection = imaplib.IMAP4_SSL(server, IMAP_PORT) + auth_string = f"user={account}\x01auth=Bearer {access_token}\x01\x01".encode("utf-8") + connection.authenticate("XOAUTH2", lambda x: auth_string) + + selected = _select_folder(connection, folder) + if not selected: + return { + "success": False, + "error": build_error_payload("FOLDER_NOT_FOUND", "文件夹选择失败", "IMAPError", 500, ""), + "emails": [], + "detail": None, + } + + status, messages = connection.search(None, "ALL") + if status != "OK" or not messages or not messages[0]: + return {"success": True, "emails": [], "detail": None} + + message_ids = messages[0].split() + total = len(message_ids) + start_idx = max(0, total - skip - top) + end_idx = total - skip + if start_idx >= end_idx: + return {"success": True, "emails": [], "detail": None} + + paged_ids = message_ids[start_idx:end_idx][::-1] + emails_data: List[Dict[str, Any]] = [] + detail = None + + ids_str = b",".join(paged_ids) + status, all_data = connection.fetch(ids_str, "(RFC822)") + if status != "OK": + return {"success": True, "emails": [], "detail": None} + + raw_by_id = {msg_id_str: raw_email for msg_id_str, raw_email in _parse_batch_fetch_response(all_data or [])} + + for i, msg_id in enumerate(paged_ids): + msg_id_str = msg_id.decode("ascii", errors="ignore").strip() + raw_email = raw_by_id.get(msg_id_str) + if raw_email is None: + continue + + msg = email.message_from_bytes(raw_email) + body_preview = get_email_body(msg) + email_item = { + "id": msg_id_str, + "subject": decode_header_value(msg.get("Subject", "无主题")), + "from": decode_header_value(msg.get("From", "未知发件人")), + "date": msg.get("Date", "未知时间"), + "body_preview": body_preview[:200] + "..." if len(body_preview) > 200 else body_preview, + } + emails_data.append(email_item) + + if i == 0: + raw_text = raw_email.decode("utf-8", errors="replace") if isinstance(raw_email, (bytes, bytearray)) else "" + detail = { + "id": email_item["id"], + "subject": email_item["subject"], + "from": email_item["from"], + "to": decode_header_value(msg.get("To", "")), + "cc": decode_header_value(msg.get("Cc", "")), + "date": email_item["date"], + "body": get_email_body(msg), + "body_html": _get_html_body(msg), + "raw_content": raw_text, + } + + return {"success": True, "emails": emails_data, "detail": detail} + except imaplib.IMAP4.error as exc: + return { + "success": False, + "error": build_error_payload("AUTH_FAILED", "IMAP认证失败", "IMAP4Error", 401, str(exc)), + "emails": [], + "detail": None, + } + except Exception as exc: + return { + "success": False, + "error": build_error_payload( + "EMAIL_FETCH_FAILED", + "获取邮件失败", + type(exc).__name__, + 500, + str(exc), + ), + "emails": [], + "detail": None, + } + finally: + if connection: + try: + connection.logout() + except Exception: + pass + + +def get_emails_imap_concurrent( + account: str, + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 20, + servers: tuple = (IMAP_SERVER_NEW, "outlook.office365.com"), +) -> Dict[str, Any]: + """并发连接多台 IMAP 服务器，返回第一个成功结果。""" + if len(servers) <= 1: + return get_emails_imap_with_server( + account, + client_id, + refresh_token, + folder, + skip, + top, + servers[0] if servers else IMAP_SERVER_NEW, + ) + + last_error = None + with ThreadPoolExecutor(max_workers=len(servers)) as executor: + futures = { + executor.submit( + get_emails_imap_with_server, + account, + client_id, + refresh_token, + folder, + skip, + top, + server, + ): server + for server in servers + } + for future in as_completed(futures): + result = future.result() + if result.get("success"): + return result + last_error = result + + return last_error or { + "success": False, + "error": {"code": "ALL_SERVERS_FAILED", "message": "所有服务器连接失败"}, + } + + +def get_email_detail_imap( + account: str, + client_id: str, + refresh_token: str, + message_id: str, + folder: str = "inbox", +) -> Optional[Dict]: + """使用 IMAP 获取邮件详情（默认使用新版服务器）。""" + return get_email_detail_imap_with_server(account, client_id, refresh_token, message_id, folder, IMAP_SERVER_NEW) + + +def get_email_detail_imap_with_server( + account: str, + client_id: str, + refresh_token: str, + message_id: str, + folder: str = "inbox", + server: str = IMAP_SERVER_NEW, +) -> Optional[Dict]: + """使用 IMAP 获取邮件详情（支持指定服务器）。""" + access_token = get_access_token_imap(client_id, refresh_token) + if not access_token: + return None + + connection = None + try: + connection = imaplib.IMAP4_SSL(server, IMAP_PORT) + auth_string = f"user={account}\1auth=Bearer {access_token}\1\1".encode("utf-8") + connection.authenticate("XOAUTH2", lambda x: auth_string) + + folder_map = { + "inbox": ['"INBOX"', "INBOX"], + "junkemail": ['"Junk"', '"Junk Email"', "Junk", '"垃圾邮件"'], + "deleteditems": [ + '"Deleted"', + '"Deleted Items"', + '"Trash"', + "Deleted", + '"已删除邮件"', + ], + "trash": [ + '"Deleted"', + '"Deleted Items"', + '"Trash"', + "Deleted", + '"已删除邮件"', + ], + } + possible_folders = folder_map.get((folder or "").lower(), ['"INBOX"']) + + selected_folder = None + for imap_folder in possible_folders: + try: + status, response = connection.select(imap_folder, readonly=True) + if status == "OK": + selected_folder = imap_folder + break + except Exception: + continue + + if not selected_folder: + return None + + fetch_id = message_id.encode() if isinstance(message_id, str) else message_id + status, msg_data = connection.fetch(fetch_id, "(RFC822)") + if status != "OK" or not msg_data or not msg_data[0]: + return None + + raw_email = msg_data[0][1] + msg = email.message_from_bytes(raw_email) + + raw_text = "" + try: + raw_text = raw_email.decode("utf-8", errors="replace") if isinstance(raw_email, (bytes, bytearray)) else "" + except Exception: + raw_text = "" + + body, body_type = get_email_body_and_type(msg) + return { + "id": message_id, + "subject": decode_header_value(msg.get("Subject", "无主题")), + "from": decode_header_value(msg.get("From", "未知发件人")), + "to": decode_header_value(msg.get("To", "")), + "cc": decode_header_value(msg.get("Cc", "")), + "date": msg.get("Date", "未知时间"), + "body": body, + "body_type": body_type, + "raw_content": raw_text, + } + except Exception: + return None + finally: + if connection: + try: + connection.logout() + except Exception: + pass + + +def delete_emails_imap( + email_addr: str, + client_id: str, + refresh_token: str, + message_ids: List[str], + server: str, +) -> Dict[str, Any]: + """通过 IMAP 删除邮件（永久删除）""" + access_token = get_access_token_graph(client_id, refresh_token) + if not access_token: + return {"success": False, "error": "获取 Access Token 失败"} + + try: + auth_string = "user=%s\x01auth=Bearer %s\x01\x01" % (email_addr, access_token) + + imap = imaplib.IMAP4_SSL(server, IMAP_PORT) + imap.authenticate("XOAUTH2", lambda x: auth_string.encode("utf-8")) + + imap.select("INBOX") + + # Graph message id 与 IMAP UID 不兼容：保留原行为（暂不支持） + return {"success": False, "error": "IMAP 删除暂不支持 (ID 格式不兼容)"} + except Exception as e: + return {"success": False, "error": str(e)} diff --git a/app/services/verification.py b/app/services/verification.py new file mode 100644 index 0000000..0c36a60 --- /dev/null +++ b/app/services/verification.py @@ -0,0 +1,598 @@ +""" +统一验证码提取模块（ZER-90） + +将验证码候选生成、评分、门控收敛为单一服务，供 Web API、External API、 +简洁模式摘要及旧版 verification_extractor 兼容层共用。 +""" + +from __future__ import annotations + +import html +import re +from dataclasses import dataclass, field +from html.parser import HTMLParser +from typing import Any, Dict, List, Optional + +# 验证码关键词列表（支持中英文） +VERIFICATION_KEYWORDS = [ + "验证码", + "code", + "验证", + "verification", + "OTP", + "动态码", + "校验码", + "verify code", + "confirmation code", + "security code", + "验证码是", + "your code", + "code is", + "激活码", + "短信验证码", +] + +VERIFICATION_PATTERN = r"\b[A-Z0-9]{4,8}\b" + +# 带连字符字母数字验证码，例如 x.ai 的 84A-KMN +HYPHENATED_VERIFICATION_PATTERN = r"(?"{}|\\^`\[\]]+' + +DEFAULT_LINK_KEYWORDS = [ + "verify", + "confirmation", + "confirm", + "activate", + "validation", +] + +LINK_CONTEXT_PHRASES = [ + "verify your email", + "verify your account", + "verify your address", + "confirm your email", + "confirm your account", + "confirm your address", + "activate your email", + "activate your account", + "email verification", + "account verification", + "验证您的邮箱", + "验证你的邮箱", + "验证您的账户", + "验证你的账户", + "验证您的账号", + "验证你的账号", + "确认您的邮箱", + "确认你的邮箱", + "确认您的账户", + "确认你的账户", + "激活您的账户", + "激活你的账户", + "激活您的邮箱", + "激活你的邮箱", + "邮箱验证", + "账号验证", + "账户验证", +] + + +@dataclass +class VerificationPolicy: + """验证码提取策略。""" + + code_regex: str | None = None + code_length: str | None = None + code_source: str = "all" + prefer_link_keywords: List[str] = field(default_factory=lambda: list(DEFAULT_LINK_KEYWORDS)) + enforce_mutual_exclusion: bool = True + apply_confidence_gate: bool = False + expected_field: str | None = None # code | link | any + + +@dataclass +class VerificationInput: + """统一邮件输入：subject / 正文 / HTML 分离，避免直接扫描原始 HTML 样式。""" + + subject: str = "" + body: str = "" + body_preview: str = "" + body_html: str = "" + html_content: str = "" + body_content: str = "" + body_content_type: str = "" + + @classmethod + def from_email_dict(cls, email: Dict[str, Any]) -> VerificationInput: + payload = email or {} + return cls( + subject=str(payload.get("subject") or "").strip(), + body=str(payload.get("body") or "").strip(), + body_preview=str(payload.get("body_preview") or "").strip(), + body_html=str(payload.get("body_html") or payload.get("html_content") or "").strip(), + html_content=str(payload.get("html_content") or "").strip(), + body_content=str(payload.get("bodyContent") or "").strip(), + body_content_type=str(payload.get("bodyContentType") or "").strip(), + ) + + def as_legacy_email_dict(self) -> Dict[str, Any]: + return { + "subject": self.subject, + "body": self.body, + "body_preview": self.body_preview, + "body_html": self.body_html or self.html_content, + "html_content": self.html_content, + "bodyContent": self.body_content, + "bodyContentType": self.body_content_type, + } + + +class HTMLTextExtractor(HTMLParser): + """HTML 转纯文本提取器（跳过 style/script 等不可见节点）。""" + + def __init__(self) -> None: + super().__init__() + self.text_parts: List[str] = [] + self._skip_tags = {"style", "script", "head", "meta", "link"} + self._current_skip = False + + def handle_starttag(self, tag: str, attrs: Any) -> None: + if tag.lower() in self._skip_tags: + self._current_skip = True + + def handle_endtag(self, tag: str) -> None: + if tag.lower() in self._skip_tags: + self._current_skip = False + + def handle_data(self, data: str) -> None: + if not self._current_skip and data.strip(): + self.text_parts.append(data.strip()) + + def get_text(self) -> str: + return " ".join(self.text_parts) + + +def html_to_visible_text(html_content: str) -> str: + if not html_content: + return "" + parser = HTMLTextExtractor() + try: + parser.feed(html_content) + return html.unescape(parser.get_text() or "").strip() + except Exception: + return html_content.strip() + + +def extract_content_text_without_subject(email_input: VerificationInput) -> str: + if email_input.body: + return email_input.body + + html_raw = email_input.body_html or email_input.html_content + if html_raw: + return html_to_visible_text(html_raw) + + if email_input.body_content: + if email_input.body_content_type.lower() == "html": + return html_to_visible_text(email_input.body_content) + return email_input.body_content + + if email_input.body_preview: + return email_input.body_preview + + return "" + + +def extract_email_text(email: Dict[str, Any]) -> str: + email_input = VerificationInput.from_email_dict(email) + content = extract_content_text_without_subject(email_input) + if content: + return content + if email_input.subject: + return email_input.subject + return "" + + +def _parse_code_length(code_length: str) -> tuple[int, int]: + m = re.match(r"^(\d+)-(\d+)$", str(code_length or "").strip()) + if not m: + raise ValueError("code_length 参数无效") + min_len = int(m.group(1)) + max_len = int(m.group(2)) + if min_len <= 0 or max_len <= 0 or min_len > max_len: + raise ValueError("code_length 参数无效") + return min_len, max_len + + +def build_code_regex(*, code_regex: str | None, code_length: str | None) -> re.Pattern[str]: + if code_regex: + try: + return re.compile(code_regex) + except re.error as exc: + raise ValueError("code_regex 参数无效") from exc + + if code_length: + min_len, max_len = _parse_code_length(code_length) + return re.compile(rf"\b[A-Za-z0-9]{{{min_len},{max_len}}}\b") + + return re.compile(r"\b\d{4,8}\b") + + +def _is_valid_hyphenated_code(code: str) -> bool: + if not code or "-" not in code: + return False + + parts = code.split("-") + if len(parts) != 2: + return False + if not all(part.isalnum() for part in parts): + return False + + alnum = "".join(parts) + if not (4 <= len(alnum) <= 10): + return False + if any(c.isdigit() for c in alnum): + return True + return len(alnum) >= 6 and all(len(part) >= 3 for part in parts) and alnum.isalpha() + + +def _has_code_context(email_content: str) -> bool: + content_lower = email_content.lower() + return any(phrase.lower() in content_lower for phrase in CODE_CONTEXT_PHRASES) + + +def _find_hyphenated_code_in_text(text: str) -> Optional[str]: + if not text: + return None + + for match in re.finditer(HYPHENATED_VERIFICATION_PATTERN, text, re.IGNORECASE): + code = match.group(1) + if _is_valid_hyphenated_code(code): + return code + return None + + +def smart_extract_hyphenated_verification_code(email_content: str) -> Optional[str]: + if not email_content: + return None + + content_lower = email_content.lower() + for keyword in VERIFICATION_KEYWORDS: + keyword_lower = keyword.lower() + pos = content_lower.find(keyword_lower) + if pos == -1: + continue + + start = max(0, pos - 50) + end = min(len(email_content), pos + len(keyword) + 50) + code = _find_hyphenated_code_in_text(email_content[start:end]) + if code: + return code + return None + + +def fallback_extract_hyphenated_verification_code(email_content: str) -> Optional[str]: + if not email_content or not _has_code_context(email_content): + return None + return _find_hyphenated_code_in_text(email_content) + + +def smart_extract_verification_code(email_content: str) -> Optional[str]: + if not email_content: + return None + + content_lower = email_content.lower() + for keyword in VERIFICATION_KEYWORDS: + keyword_lower = keyword.lower() + pos = content_lower.find(keyword_lower) + if pos == -1: + continue + + start = max(0, pos - 50) + end = min(len(email_content), pos + len(keyword) + 50) + context = email_content[start:end] + matches = re.findall(VERIFICATION_PATTERN, context, re.IGNORECASE) + for match in matches: + if any(c.isdigit() for c in match): + return match + + return smart_extract_hyphenated_verification_code(email_content) + + +def fallback_extract_verification_code(email_content: str) -> Optional[str]: + if not email_content: + return None + + filtered: List[str] = [] + for match in re.findall(VERIFICATION_PATTERN, email_content, re.IGNORECASE): + if not any(c.isdigit() for c in match): + continue + + if match.isdigit() and len(match) == 4: + year = int(match) + if 1900 <= year <= 2100: + continue + hour = int(match[:2]) + minute = int(match[2:]) + if 0 <= hour <= 23 and 0 <= minute <= 59: + continue + if 2020 <= year <= 2030: + continue + + filtered.append(match) + + if filtered: + return filtered[0] + + return fallback_extract_hyphenated_verification_code(email_content) + + +def smart_extract_code_by_keywords(email_content: str, code_re: re.Pattern[str]) -> Optional[str]: + if not email_content: + return None + + content_lower = email_content.lower() + for keyword in VERIFICATION_KEYWORDS: + keyword_lower = keyword.lower() + pos = content_lower.find(keyword_lower) + if pos == -1: + continue + + start = max(0, pos - 50) + end = min(len(email_content), pos + len(keyword) + 50) + context = email_content[start:end] + for match in code_re.finditer(context): + value = match.group(0) + if value and any(c.isdigit() for c in value): + return value + return None + + +def fallback_extract_code(email_content: str, code_re: re.Pattern[str]) -> Optional[str]: + if not email_content: + return None + + candidates: List[str] = [] + for match in code_re.finditer(email_content): + value = match.group(0) or "" + if not value or not any(c.isdigit() for c in value): + continue + + if value.isdigit() and len(value) == 4: + year = int(value) + if 1900 <= year <= 2100: + continue + hour = int(value[:2]) + minute = int(value[2:]) + if 0 <= hour <= 23 and 0 <= minute <= 59: + continue + if 2020 <= year <= 2030: + continue + + candidates.append(value) + + return candidates[0] if candidates else None + + +def extract_links(email_content: str) -> List[str]: + if not email_content: + return [] + + cleaned_links = [link.rstrip(".,;:!?)>'\"") for link in re.findall(LINK_PATTERN, email_content, re.IGNORECASE)] + seen: set[str] = set() + unique_links: List[str] = [] + for link in cleaned_links: + if link not in seen: + seen.add(link) + unique_links.append(link) + return unique_links + + +def pick_preferred_link(links: List[str], prefer_link_keywords: List[str]) -> Optional[str]: + if not links: + return None + + keywords = [keyword.lower() for keyword in (prefer_link_keywords or []) if keyword] + if keywords: + for keyword in keywords: + for link in links: + if keyword in (link or "").lower(): + return link + return links[0] + + +def build_source_text(email_input: VerificationInput, *, code_source: str) -> tuple[str, str]: + subject = email_input.subject + content = extract_content_text_without_subject(email_input) + html_raw = email_input.body_html or email_input.html_content + + source = str(code_source or "all").strip().lower() + if source == "subject": + return subject, "subject" + if source == "content": + return content, "content" + if source == "html": + return (html_to_visible_text(html_raw) if html_raw else ""), "html" + return f"{subject} {content}".strip(), "all" + + +def extract_verification_code_from_text( + source_text: str, + *, + code_regex: str | None, + code_length: str | None, +) -> tuple[Optional[str], str]: + code_re = build_code_regex(code_regex=code_regex, code_length=code_length) + caller_directed_code = bool(code_regex) + + verification_code = smart_extract_code_by_keywords(source_text, code_re) + code_confidence = "high" if verification_code else "low" + + if not verification_code: + verification_code = fallback_extract_code(source_text, code_re) + if verification_code and caller_directed_code: + code_confidence = "high" + + if not verification_code: + verification_code = smart_extract_hyphenated_verification_code(source_text) + if verification_code: + code_confidence = "high" + + if not verification_code: + verification_code = fallback_extract_hyphenated_verification_code(source_text) + if verification_code: + code_confidence = "high" + + return verification_code, code_confidence + + +def extract_verification( + email_input: VerificationInput, + policy: VerificationPolicy | None = None, +) -> Dict[str, Any]: + """ + 统一验证码/链接提取入口。 + + 返回字段与 extract_verification_info_with_options 兼容。 + """ + active_policy = policy or VerificationPolicy() + source_text, match_source = build_source_text(email_input, code_source=active_policy.code_source) + subject = email_input.subject + content = extract_content_text_without_subject(email_input) + html_raw = email_input.body_html or email_input.html_content + + verification_code, code_confidence = extract_verification_code_from_text( + source_text, + code_regex=active_policy.code_regex, + code_length=active_policy.code_length, + ) + + links = extract_links(f"{subject} {content} {html_raw}".strip()) + prefer_keywords = active_policy.prefer_link_keywords or DEFAULT_LINK_KEYWORDS + + verification_link = None + link_confidence = "low" + should_pick_link = (not active_policy.enforce_mutual_exclusion) or (not verification_code) + if should_pick_link: + verification_link = pick_preferred_link(links, prefer_keywords) + if verification_link: + for keyword in prefer_keywords: + if keyword and keyword.lower() in verification_link.lower(): + link_confidence = "high" + break + if link_confidence != "high": + full_text_lower = f"{subject} {content}".lower() + for phrase in LINK_CONTEXT_PHRASES: + if phrase.lower() in full_text_lower: + link_confidence = "high" + break + + confidence = "high" if code_confidence == "high" or link_confidence == "high" else "low" + + parts: List[str] = [] + if verification_code: + parts.append(verification_code) + if verification_link: + parts.append(verification_link) + formatted = " ".join(parts) if parts else None + + result = { + "verification_code": verification_code, + "verification_link": verification_link, + "links": links, + "formatted": formatted, + "match_source": match_source, + "confidence": confidence, + "code_confidence": code_confidence, + "link_confidence": link_confidence, + } + + if active_policy.apply_confidence_gate: + result = apply_confidence_gate(result, enforce_mutual_exclusion=active_policy.enforce_mutual_exclusion) + + expected_field = str(active_policy.expected_field or "").strip().lower() + if expected_field == "code": + result["verification_link"] = None + result["link_confidence"] = "low" + result["formatted"] = result.get("verification_code") or None + elif expected_field == "link": + result["verification_code"] = None + result["code_confidence"] = "low" + result["formatted"] = result.get("verification_link") or None + + return result + + +def apply_confidence_gate(extracted: Dict[str, Any], *, enforce_mutual_exclusion: bool = True) -> Dict[str, Any]: + result = dict(extracted) + + if result.get("code_confidence") != "high": + result["verification_code"] = None + if result.get("link_confidence") != "high": + result["verification_link"] = None + + if enforce_mutual_exclusion and result.get("verification_code"): + result["verification_link"] = None + result["link_confidence"] = "low" + + parts = [value for value in (result.get("verification_code"), result.get("verification_link")) if value] + result["formatted"] = " ".join(parts) if parts else None + result["confidence"] = ( + "high" if result.get("code_confidence") == "high" or result.get("link_confidence") == "high" else "low" + ) + return result + + +def policy_from_resolved( + resolved: Dict[str, Any] | None, + *, + code_source: str = "all", + enforce_mutual_exclusion: bool = True, + apply_confidence_gate: bool = False, + expected_field: str | None = None, +) -> VerificationPolicy: + payload = resolved or {} + return VerificationPolicy( + code_regex=payload.get("code_regex"), + code_length=payload.get("code_length"), + code_source=code_source, + enforce_mutual_exclusion=enforce_mutual_exclusion, + apply_confidence_gate=apply_confidence_gate, + expected_field=expected_field, + ) + + +def extract_verification_from_email_dict( + email: Dict[str, Any], + *, + code_regex: str | None = None, + code_length: str | None = None, + code_source: str = "all", + prefer_link_keywords: list[str] | None = None, + enforce_mutual_exclusion: bool = True, + apply_confidence_gate_after: bool = False, +) -> Dict[str, Any]: + """兼容旧 extract_verification_info_with_options 签名的薄封装。""" + policy = VerificationPolicy( + code_regex=code_regex, + code_length=code_length, + code_source=code_source, + prefer_link_keywords=list(prefer_link_keywords or DEFAULT_LINK_KEYWORDS), + enforce_mutual_exclusion=enforce_mutual_exclusion, + apply_confidence_gate=apply_confidence_gate_after, + ) + return extract_verification(VerificationInput.from_email_dict(email), policy) diff --git a/app/templates/index.html b/app/templates/index.html new file mode 100644 index 0000000..c32ce95 --- /dev/null +++ b/app/templates/index.html @@ -0,0 +1,2225 @@ + + + + + + + + + + + + + Outlook 邮件管理 + + + + + + + + + + + + + + + +
+ + + + + + + +
+ +
+ + + + + + + + + + + + + + + +
+ + + +
+ +
+ + + +
+ +
数据概览
+ +
运营数据大盘
+ +
+ +
+ +
+ +
+ + + + + + + + + + + +
+ +
+ +
+ +
+ +
+ +
+ +
    + +
    + +
    + +
    + +
    + +
    玻璃态概览面板
    + +
    + +
    📊 数据概览
    + + 细腻卡片视图 + +
    + +
    账号、验证码、对外 API、邮箱池与系统活动统一看板
    + +
    + +
    + + 最近刷新：-- + + + +
    + +
    + + + +
    + + + + + + + + + + + +
    + + + +
    + +
    + +
    加载中…
    + +
    + +
    + +
    + +
    + +
    点击后加载
    + +
    + +
    + +
    + +
    + +
    点击后加载
    + +
    + +
    + +
    + +
    + +
    点击后加载
    + +
    + +
    + +
    + +
    + +
    点击后加载
    + +
    + +
    + +
    + +
    + + + + + +
    + + + + + +
    + + + +
    + +
    + + 分组 + + + +
    + +
    + + + +
    + +
    + +
    加载中…
    + +
    + +
    + + + + + +
    + + + + + +
    + +
    + + + + + + + + 选择分组 + + + +
    + + + + + +
    + +
    + +
    + + + +
    + +
    + + + + + + + +
    + + + + + +
    + +
    + + + +
    + +
    + + 📁 + +

    请从左侧选择一个分组

    + +
    + +
    + + + +
    + + + + + +
    + + + + + +
    + +
    + +
    + + + + + + + +
    + +
    + + + +
    + +
    + + + + + +
    + +
    + + 📬 + +

    请从左侧选择一个邮箱账号

    + +
    + +
    + + + + + +
    + +
    + + + + + +
    + + + + + +
    + +
    + + + +
    + +
    + + ⚡ 临时邮箱 + + + +
    + +
    + + + + + + + +
    支持自定义前缀和多域名创建。
    + + + +
    + +
    + +
    + + 📭 + +

    暂无临时邮箱

    + + + +
    + +
    + +
    + + + +
    + +
    + + 选择一个临时邮箱 + +
    + + + +
    + +
    + +
    + +
    + + 📬 + +

    选择一个临时邮箱查看邮件

    + +
    + +
    + +
    + + + +
    + +
    + + + + + +
    + +
    + +
    + +
    🔄 刷新日志
    + +
    + +
    + +
    加载中…
    + +
    + +
    + +
    + + + + + +
    + + + +
    + + + + + +
    + + + + + + + + + +
    + + + + + +
    + + + + + +
    + +
    + +
    + +
    ⚙️ 基础设置
    + +
    + +
    + +
    + + + + + +
    用于登录系统的密码
    + +
    + + + +
    + +
    🤖 验证码 AI 增强
    + +
    + + + +
    规则提取优先；仅在规则不足时触发 AI 回退。
    + +
    + +
    + + + + + +
    + +
    + + + + + +
    未设置
    + +
    + +
    + + + + + +
    + +
    + + + + 建议先保存配置再测试。 + +
    + +
    + + + +
    + + + +
    + +
    + +
    + +
    + + + + + +
    + + + +
    + +
    + +
    + + + +
    + + + + + +
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    🔧 GPTMail 配置
    + +
    + +
    + +
    + + + + + +
    上游临时邮箱服务地址，留空则使用默认配置。
    + +
    + +
    + + + + + +
    用于临时邮箱能力。旧版临时邮箱 API Key 字段仅保留兼容读取与迁移，不再作为正式配置字段。
    + +
    + +
    + + + + + +
    支持字符串数组或 `{name, enabled}` 对象数组。
    + +
    + +
    + + + + + +
    + +
    + + + + + +
    + +
    + +
    + + + + + + + + + + + + + + + + + +
    + +
    + +
    + +
    🧩
    + +
    + +
    插件管理
    + +
    安装、卸载并应用第三方 Provider 插件；运行时配置请在上方 Provider 设置区完成
    + +
    + +
    + +
    + + + + ▼ + +
    + +
    + + + +
    + +
    + + + + + +
    + +
    + +
    + +
    🔐 API 安全设置
    + +
    + +
    + +
    + + + +
    + + + + + + + +
    + +
    + + 用于对外开放接口鉴权（请求头：X-API-Key）。如需禁用对外开放接口，可清空后保存。 + +
    + +
    + +
    + + + + + +
    + + 用于按调用方维护多个 Key、邮箱范围授权和启停状态。保留已有脱敏 `api_key` 表示不修改该 Key；清空后保存表示清空全部多 Key。 + +
    + +
    + + + + + +
    + +
    🛡️ 公网安全配置（P1）
    + +
    + + + +
    关闭时（默认）仅做 API Key 鉴权；开启后额外启用 IP 白名单、限流、高风险端点禁用等安全策略。
    + +
    + +
    + + + + + +
    支持精确 IP 和 CIDR 格式。每行一个，保存时自动转为 JSON 数组。
    + +
    + +
    + + + + + +
    每个 IP 每分钟最大请求数（默认 60）。
    + +
    + +
    + + + +
    公网模式下建议禁用，防止泄露完整邮件原文。
    + +
    + +
    + + + +
    公网模式下建议禁用，防止长连接资源耗尽（Slowloris 风险）。
    + +
    + +
    + +
    📦 External Pool
    + +
    + + + +
    开启后才允许调用 `/api/external/pool/*`。仅设置对外 API Key 不会自动开启邮箱池对外接口。
    + +
    + +
    + + + +
    关闭后可供外部调用随机领取邮箱池账号。
    + +
    + +
    + + + +
    关闭后可供外部调用释放已领取账号。
    + +
    + +
    + + + +
    关闭后可供外部调用领取完成/回写结果。
    + +
    + +
    + + + +
    关闭后可供外部读取邮箱池统计信息。
    + +
    + +
    + +
    + +
    + +
    + +
    + + + + + +
    + + + +
    + +
    + +
    🔄 Token 刷新设置
    + +
    + +
    + +
    + + + +
    关闭后将不会自动执行定时刷新任务
    + +
    + +
    + + + +
    + + + + + +
    + +
    + +
    + +
    + + + +
    + + + + 天 + +
    + +
    建议设置为 30 天，防止 Token 因 90 天不使用而过期
    + +
    + +
    + + + +
    + + + +
    + + + + 秒 + +
    + +
    建议设置为 5-10 秒，避免频繁请求触发 API 限流
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    🔔 自动轮询设置
    + +
    + +
    + +
    + + + +
    开启后自动检查当前账号是否有新邮件
    + +
    + +
    + + + +
    + + + + 秒 + +
    + +
    范围：3-300 秒，建议设置为 5-30 秒
    + +
    + +
    + + + +
    + + + + 次 + +
    + +
    范围：0-100 次，设置为 0 表示持续轮询
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    ✉️ Email 通知
    + +
    + +
    + +
    + + + +
    这里只配置 Email 通知通道。普通邮箱需在账号列表开启通知后才会通过 Email 发送；临时邮箱按当前通知规则处理。启用后仅从新到达的邮件开始通知。
    + +
    + +
    + + + + + +
    这里只配置 Email 渠道的接收邮箱，不会让所有普通邮箱自动发送。
    + +
    + +
    + + + +
    按"先保存，再测试"处理；成功语义为"请求已提交，请检查收件箱"。
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    📬 Telegram 通知
    + +
    + +
    + +
    + + + + + +
    这里只配置 Telegram 通知通道。普通邮箱需在账号列表开启通知后才会通过 Telegram 发送；临时邮箱按当前通知规则处理。
    + +
    格式：1234567890:AAxxxxxx（留空则禁用推送）
    + +
    + +
    + + + + + +
    可用 @userinfobot 获取你的 Chat ID
    + +
    + +
    + + + +
    + + + + 秒 + +
    + +
    范围：10-86400 秒，默认 600 秒（10 分钟）
    + +
    + +
    + + + +
    + + + + + + + +
    + +
    留空则直连。格式：socks5://host:port 或 http://host:port
    + +
    + +
    + + + +
    验证当前 Telegram 通知通道是否配置正确
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    📡 Webhook 通知
    + +
    + +
    + +
    + + + +
    全局单 URL 通道；普通邮箱和临时邮箱都遵循当前通知参与规则。
    + +
    + +
    + + + + + +
    仅支持 http:// 或 https://，发送格式为 text/plain。
    + +
    + +
    + + + + + +
    仅当有值时会附带请求头 X-Webhook-Token。
    + +
    + +
    + + + +
    按“先保存，再测试”处理；测试仅使用已保存配置。
    + +
    + +
    + +
    + + + + + +
    + +
    + +
    🔄 一键更新配置
    + +
    + +
    + + + +
    + + + +
    + + + + + +
    + +
    + + Watchtower: 使用外部 Watchtower 容器管理更新（推荐，更安全）
    + + Docker API: 容器直接通过 Docker API 自更新（需挂载 docker.sock，存在安全风险） + +
    + +
    + + + + + + + + + + + +
    + + + + + +
    + +
    + +
    + +
    🚀 触发容器更新
    + +
    拉取最新镜像并重启容器，使用上方选择的更新方式
    + +
    + + + +
    + + + +
    + + + + + +
    + + + +
    + + 📌 首次配置指南：
    + + 1. 在 .env 文件中设置 WATCHTOWER_HTTP_API_TOKEN（使用 python -c "import secrets; print(secrets.token_hex(32))" 生成）
    + + 2. 使用 docker-compose up -d 重启容器以应用 Token
    + + 3. 在下方配置相同的 Token 并保存设置
    + + 4. 点击"测试连通性"验证配置是否正确 + +
    + +
    + + + + + +
    默认 http://watchtower:8080，仅 Docker 部署模式下可用
    + +
    + +
    + + + + + +
    与 docker-compose 中 WATCHTOWER_HTTP_API_TOKEN 保持一致。留空则读取环境变量。
    + +
    + +
    + +
    + + + + + +
    + +
    验证 Watchtower 服务是否可达且 Token 正确
    + +
    + +
    + +
    + +
    + + + + + +
    + + + +
    + +
    + +
    + + + + + +
    + +
    + +
    + +
    🎱 号池管理
    + +
    + +
    + + + +
    + + + + + + + + + + + +
    + + + + + +
    + +
    + +
    加载中…
    + +
    + +
    + +
    + +
    + +
    + + + + +
    + +
    + +
    + +
    📋 审计日志
    + +
    + +
    + +
    加载中…
    + +
    + +
    + +
    + +
    + +
    + + + + + +
    + + + + {% include 'partials/modals.html' %} + + + + {% include 'partials/scripts.html' %} + + + + + + + diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..1ba7665 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,29 @@ +services: + ni-mail: + build: + context: . + dockerfile: Dockerfile + container_name: ni-mail + restart: unless-stopped + stop_grace_period: 30s + env_file: + - .env + volumes: + - ni-mail-data:/app/data + environment: + - TZ= + ports: + - ":8080" + extra_hosts: + - "host.docker.internal:host-gateway" + mem_limit: 512m + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + read_only: true + tmpfs: + - /tmp:size=64m + +volumes: + ni-mail-data: diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a040095 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +Flask==3.1.0 +Flask-SQLAlchemy==3.1.1 +requests==2.32.3 +cryptography==42.0.8 +python-dotenv==1.0.1 diff --git a/scripts/entrypoint.sh b/scripts/entrypoint.sh new file mode 100644 index 0000000..140f077 --- /dev/null +++ b/scripts/entrypoint.sh @@ -0,0 +1,7 @@ +#!/bin/sh +set -e + +DB_PATH="/ni_mail.db" +export DATABASE_URL="sqlite:///" + +exec python -m flask --app "app:create_app()" run --host=0.0.0.0 --port=""