diff --git a/CHANGELOG.md b/CHANGELOG.md index a69da157..4a67b2a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to OutlookMail Plus are documented in this file. ### 修复 / Bug Fixes +- **验证码提取大小写被篡改**：验证码提取收口处（规则分支 `_smart_extract_code_by_keywords` / `_fallback_extract_code` 及历史函数 `smart_extract_verification_code` / `fallback_extract_verification_code`，以及 AI 回退 `enhance_verification_with_ai_fallback`）不再强制 `.upper()`，`verification_code` 与 `formatted` 保持邮件原文大小写（如 `ab12cd` 不再返回 `AB12CD`）。补充回归测试。 - **Issue #65 Watchtower 容器镜像过时**：`docker-compose.yml` 中固定 Watchtower 版本为 `containrrr/watchtower:1.7.1`，避免本地缓存的旧版镜像（内嵌 Docker 客户端 API 1.25）连接新版本 Docker Engine（要求 API 1.44+）时失败。README 新增故障排查指引。 ## [v2.7.0] - 2026-05-29 diff --git a/README.en.md b/README.en.md index b0f3d940..109ef098 100644 --- a/README.en.md +++ b/README.en.md @@ -174,7 +174,7 @@ services: - outlook-net watchtower: - image: containrrr/watchtower:1.7.1 + image: nickfedor/watchtower:latest container_name: watchtower restart: unless-stopped volumes: diff --git a/README.md b/README.md index 62475b20..fb79fe1d 100644 --- a/README.md +++ b/README.md @@ -178,7 +178,7 @@ services: - outlook-net watchtower: - image: containrrr/watchtower:1.7.1 + image: nickfedor/watchtower:latest container_name: watchtower restart: unless-stopped volumes: diff --git a/docker-compose.yml b/docker-compose.yml index 0165d251..f8974d08 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -79,7 +79,7 @@ services: - outlook-net watchtower: - image: containrrr/watchtower:1.7.1 + image: nickfedor/watchtower:latest container_name: watchtower restart: unless-stopped volumes: diff --git a/outlook_web/controllers/emails.py b/outlook_web/controllers/emails.py index 97f1450d..15f0310c 100644 --- a/outlook_web/controllers/emails.py +++ b/outlook_web/controllers/emails.py @@ -4,7 +4,9 @@ import time from typing import Any, Dict, List, Optional -from flask import current_app, jsonify, request +import json + +from flask import Response, current_app, jsonify, request, stream_with_context from outlook_web import config from outlook_web.audit import log_audit @@ -19,6 +21,8 @@ from outlook_web.services import graph as graph_service from outlook_web.services import imap as imap_service from outlook_web.services import verification_channel_routing as verification_channel_service +from outlook_web.services.channel_cache import get_cached_channel, set_cached_channel +from outlook_web.services.email_cache import get_cached_emails, set_cached_emails from outlook_web.services.imap_generic import ( get_email_detail_imap_generic_result, get_emails_imap_generic, @@ -1549,3 +1553,100 @@ def api_external_get_probe_status(probe_id: str) -> Any: details={"code": "INTERNAL_ERROR", "probe_id": probe_id}, ) return jsonify(external_api_service.fail("INTERNAL_ERROR", "服务内部错误")), 500 + + +@login_required +def api_stream_emails(email_addr: str) -> Any: + """SSE 流式获取邮件：IMAP 每拉到一封就推一条 event，前端边收边渲染。""" + account = accounts_repo.get_account_by_email(email_addr) + if not account: + return build_error_response( + "ACCOUNT_NOT_FOUND", "账号不存在", + message_en="Account not found", err_type="NotFoundError", + status=404, details=f"email={email_addr}", + ) + + folder = request.args.get("folder", "inbox") + skip = int(request.args.get("skip", 0)) + top = int(request.args.get("top", 20)) + force = request.args.get("force", "").lower() in ("1", "true") + + account_type = (account.get("account_type") or "outlook").strip().lower() + + # 获取分组代理设置 + proxy_url = "" + if account.get("group_id"): + group = groups_repo.get_group_by_id(account["group_id"]) + if group: + proxy_url = group.get("proxy_url", "") or "" + + def _sse(event: str, data: dict) -> str: + return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" + + def generate(): + # --- 服务端缓存（2h TTL）--- + if not force: + cached = get_cached_emails(email_addr, folder) + if cached: + for e in cached["emails"]: + yield _sse("email", e) + yield _sse("done", { + "method": cached["method"] + " (cached)", + "count": len(cached["emails"]), + "has_more": cached["has_more"], + }) + return + + collected_emails = [] + cached_channel = get_cached_channel(email_addr) + + # --- Graph API 尝试 --- + if account_type != "imap" and cached_channel != "imap": + graph_result = graph_service.get_emails_graph( + account["client_id"], account["refresh_token"], folder, skip, top, proxy_url, + ) + if graph_result.get("success"): + set_cached_channel(email_addr, "graph") + emails = graph_result.get("emails", []) + for e in emails: + fmt = { + "id": e.get("id"), + "subject": e.get("subject", "无主题"), + "from": e.get("from", {}).get("emailAddress", {}).get("address", "未知"), + "date": e.get("receivedDateTime", ""), + "is_read": e.get("isRead", False), + "has_attachments": e.get("hasAttachments", False), + "body_preview": e.get("bodyPreview", ""), + } + collected_emails.append(fmt) + yield _sse("email", fmt) + has_more = len(emails) >= top + set_cached_emails(email_addr, folder, collected_emails, "Graph API", has_more) + yield _sse("done", {"method": "Graph API", "count": len(emails), "has_more": has_more}) + return + elif graph_result.get("no_mail_permission"): + set_cached_channel(email_addr, "imap") + + # --- IMAP 流式 --- + yield _sse("status", {"message": "IMAP connecting..."}) + imap_method = "IMAP" + for item in imap_service.stream_emails_imap( + account["email"], account["client_id"], account["refresh_token"], + folder, skip, top, + ): + if item.get("type") == "email": + collected_emails.append(item["data"]) + yield _sse("email", item["data"]) + elif item.get("type") == "done": + imap_method = item.get("method", "IMAP") + set_cached_channel(email_addr, "imap") + set_cached_emails(email_addr, folder, collected_emails, imap_method, False) + yield _sse("done", {"method": imap_method, "count": item.get("count", 0), "has_more": False}) + elif item.get("type") == "error": + yield _sse("error", item.get("data", {})) + + return Response( + stream_with_context(generate()), + mimetype="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) diff --git a/outlook_web/controllers/external_pool.py b/outlook_web/controllers/external_pool.py index f6b216c1..b52eb789 100644 --- a/outlook_web/controllers/external_pool.py +++ b/outlook_web/controllers/external_pool.py @@ -4,6 +4,7 @@ from flask import jsonify, request +from outlook_web.repositories import accounts as accounts_repo from outlook_web.repositories import settings as settings_repo from outlook_web.security.auth import api_key_required, get_external_api_consumer from outlook_web.security.external_api_guard import external_api_guards @@ -40,11 +41,7 @@ def _error_response(endpoint: str, exc: PoolServiceError): def _check_pool_external_enabled(endpoint: str): if settings_repo.get_pool_external_enabled(): return None - _audit( - endpoint, - "error", - details={"code": "FEATURE_DISABLED", "feature": "external_pool"}, - ) + _audit(endpoint, "error", details={"code": "FEATURE_DISABLED", "feature": "external_pool"}) return ( jsonify( external_api_service.fail( @@ -64,11 +61,7 @@ def _check_pool_access(endpoint: str): _audit( endpoint, "error", - details={ - "code": "FORBIDDEN", - "feature": "external_pool", - "reason": "pool_access_required", - }, + details={"code": "FORBIDDEN", "feature": "external_pool", "reason": "pool_access_required"}, ) return ( jsonify( @@ -82,34 +75,65 @@ def _check_pool_access(endpoint: str): ) -@api_key_required -@external_api_guards(feature="pool_claim_random") -def api_external_pool_claim_random(): - endpoint = "/api/external/pool/claim-random" +def _get_consumer_key() -> str: + consumer = get_external_api_consumer() or {} + return str(consumer.get("consumer_key") or "").strip() + + +def _account_payload(account_id: int, *, pool_status: str) -> dict[str, Any]: + account = accounts_repo.get_account_by_id(account_id) or {} + return { + "account_id": account_id, + "pool_status": pool_status, + "provider": account.get("provider") or "", + "email_domain": account.get("email_domain") or "", + } + + +def _parse_account_id(endpoint: str, raw_value) -> int | tuple: + if raw_value is None: + _audit(endpoint, "error", details={"code": "ACCOUNT_ID_MISSING"}) + return jsonify(external_api_service.fail("ACCOUNT_ID_MISSING", "account_id 不能为空")), 400 + try: + return int(raw_value) + except (TypeError, ValueError): + _audit(endpoint, "error", details={"code": "ACCOUNT_ID_INVALID"}) + return jsonify(external_api_service.fail("ACCOUNT_ID_INVALID", "account_id 必须为整数")), 400 + + +def _claim_pool_account(endpoint: str, *, require_email_domain: bool = False): disabled_resp = _check_pool_external_enabled(endpoint) if disabled_resp is not None: return disabled_resp access_resp = _check_pool_access(endpoint) if access_resp is not None: return access_resp + + consumer_key = _get_consumer_key() body = request.get_json(silent=True) or {} caller_id = body.get("caller_id", "") task_id = body.get("task_id", "") + project_key = body.get("project_key", "") provider = body.get("provider") - project_key = body.get("project_key") email_domain = body.get("email_domain") + if require_email_domain and not str(email_domain or "").strip(): + _audit(endpoint, "error", details={"code": "EMAIL_DOMAIN_REQUIRED"}) + return jsonify(external_api_service.fail("EMAIL_DOMAIN_REQUIRED", "email_domain 不能为空")), 400 + try: account = claim_random( + consumer_key=consumer_key, + project_key=project_key, caller_id=caller_id, task_id=task_id, provider=provider, - project_key=project_key, email_domain=email_domain, ) data = { "account_id": account["id"], "email": account["email"], + "provider": account.get("provider") or "", "email_domain": account.get("email_domain") or "", "claim_token": account["claim_token"], "claimed_at": account.get("claimed_at") or "", @@ -119,24 +143,33 @@ def api_external_pool_claim_random(): endpoint, "ok", details={ - "provider": provider or "", - "project_key": project_key or "", - "email_domain": email_domain or "", + "provider": data["provider"], + "email_domain": data["email_domain"], + "project_key": project_key, "account_id": data["account_id"], }, + email_addr=data["email"], ) return jsonify(external_api_service.ok(data)) except PoolServiceError as exc: return _error_response(endpoint, exc) except Exception as exc: - _audit( - endpoint, - "error", - details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}, - ) + _audit(endpoint, "error", details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}) return jsonify(external_api_service.fail("INTERNAL_ERROR", "服务内部错误")), 500 +@api_key_required +@external_api_guards(feature="pool_claim_random") +def api_external_pool_claim_random(): + return _claim_pool_account("/api/external/pool/claim-random") + + +@api_key_required +@external_api_guards(feature="pool_claim_random") +def api_external_pool_claim_domain(): + return _claim_pool_account("/api/external/pool/claim-domain", require_email_domain=True) + + @api_key_required @external_api_guards(feature="pool_claim_release") def api_external_pool_claim_release(): @@ -147,40 +180,35 @@ def api_external_pool_claim_release(): access_resp = _check_pool_access(endpoint) if access_resp is not None: return access_resp + + consumer_key = _get_consumer_key() body = request.get_json(silent=True) or {} - account_id = body.get("account_id") + account_id = _parse_account_id(endpoint, body.get("account_id")) + if isinstance(account_id, tuple): + return account_id + claim_token = body.get("claim_token", "") caller_id = body.get("caller_id", "") task_id = body.get("task_id", "") + project_key = body.get("project_key", "") reason = body.get("reason") - if account_id is None: - _audit(endpoint, "error", details={"code": "ACCOUNT_ID_MISSING"}) - return jsonify(external_api_service.fail("ACCOUNT_ID_MISSING", "account_id 不能为空")), 400 - try: - account_id = int(account_id) - except (TypeError, ValueError): - _audit(endpoint, "error", details={"code": "ACCOUNT_ID_INVALID"}) - return jsonify(external_api_service.fail("ACCOUNT_ID_INVALID", "account_id 必须为整数")), 400 - try: release_claim( + consumer_key=consumer_key, + project_key=project_key, account_id=account_id, claim_token=claim_token, caller_id=caller_id, task_id=task_id, reason=reason, ) - _audit(endpoint, "ok", details={"account_id": account_id}) - return jsonify(external_api_service.ok({"account_id": account_id, "pool_status": "available"})) + _audit(endpoint, "ok", details={"account_id": account_id, "project_key": project_key}) + return jsonify(external_api_service.ok(_account_payload(account_id, pool_status="available"))) except PoolServiceError as exc: return _error_response(endpoint, exc) except Exception as exc: - _audit( - endpoint, - "error", - details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}, - ) + _audit(endpoint, "error", details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}) return jsonify(external_api_service.fail("INTERNAL_ERROR", "服务内部错误")), 500 @@ -194,25 +222,24 @@ def api_external_pool_claim_complete(): access_resp = _check_pool_access(endpoint) if access_resp is not None: return access_resp + + consumer_key = _get_consumer_key() body = request.get_json(silent=True) or {} - account_id = body.get("account_id") + account_id = _parse_account_id(endpoint, body.get("account_id")) + if isinstance(account_id, tuple): + return account_id + claim_token = body.get("claim_token", "") caller_id = body.get("caller_id", "") task_id = body.get("task_id", "") + project_key = body.get("project_key", "") result = body.get("result", "") detail = body.get("detail") - if account_id is None: - _audit(endpoint, "error", details={"code": "ACCOUNT_ID_MISSING"}) - return jsonify(external_api_service.fail("ACCOUNT_ID_MISSING", "account_id 不能为空")), 400 - try: - account_id = int(account_id) - except (TypeError, ValueError): - _audit(endpoint, "error", details={"code": "ACCOUNT_ID_INVALID"}) - return jsonify(external_api_service.fail("ACCOUNT_ID_INVALID", "account_id 必须为整数")), 400 - try: new_status = complete_claim( + consumer_key=consumer_key, + project_key=project_key, account_id=account_id, claim_token=claim_token, caller_id=caller_id, @@ -225,19 +252,16 @@ def api_external_pool_claim_complete(): "ok", details={ "account_id": account_id, + "project_key": project_key, "result": result, "pool_status": new_status, }, ) - return jsonify(external_api_service.ok({"account_id": account_id, "pool_status": new_status})) + return jsonify(external_api_service.ok(_account_payload(account_id, pool_status=new_status))) except PoolServiceError as exc: return _error_response(endpoint, exc) except Exception as exc: - _audit( - endpoint, - "error", - details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}, - ) + _audit(endpoint, "error", details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}) return jsonify(external_api_service.fail("INTERNAL_ERROR", "服务内部错误")), 500 @@ -256,9 +280,5 @@ def api_external_pool_stats(): _audit(endpoint, "ok", details={"snapshot": True}) return jsonify(external_api_service.ok(stats)) except Exception as exc: - _audit( - endpoint, - "error", - details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}, - ) + _audit(endpoint, "error", details={"code": "INTERNAL_ERROR", "err": type(exc).__name__}) return jsonify(external_api_service.fail("INTERNAL_ERROR", "服务内部错误")), 500 diff --git a/outlook_web/db.py b/outlook_web/db.py index 5272f592..65f7daf2 100644 --- a/outlook_web/db.py +++ b/outlook_web/db.py @@ -15,6 +15,7 @@ is_encrypted, is_password_hashed, ) +from outlook_web.services.providers import extract_email_domain # 数据库 Schema 版本（用于升级可验证/可诊断） # v3：对齐 PRD-00005 / FD-00005 / TDD-00005（accounts 表新增多邮箱字段：account_type/provider/imap_host/imap_port/imap_password） @@ -189,6 +190,7 @@ def init_db(database_path: Optional[str] = None): CREATE TABLE IF NOT EXISTS accounts ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT UNIQUE NOT NULL, + email_domain TEXT DEFAULT '', password TEXT, client_id TEXT NOT NULL, refresh_token TEXT NOT NULL, @@ -400,6 +402,8 @@ def init_db(database_path: Optional[str] = None): cursor.execute("ALTER TABLE accounts ADD COLUMN account_type TEXT DEFAULT 'outlook'") if "provider" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN provider TEXT DEFAULT 'outlook'") + if "email_domain" not in columns: + cursor.execute("ALTER TABLE accounts ADD COLUMN email_domain TEXT DEFAULT ''") if "imap_host" not in columns: cursor.execute("ALTER TABLE accounts ADD COLUMN imap_host TEXT") if "imap_port" not in columns: @@ -1028,16 +1032,28 @@ def init_db(database_path: Optional[str] = None): id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, claim_token TEXT NOT NULL, + consumer_key TEXT NOT NULL DEFAULT '', + project_key TEXT NOT NULL DEFAULT '', caller_id TEXT NOT NULL, task_id TEXT NOT NULL, action TEXT NOT NULL, result TEXT DEFAULT NULL, detail TEXT DEFAULT NULL, claimed_at TEXT DEFAULT NULL, + claim_read_context TEXT DEFAULT NULL, created_at TEXT NOT NULL, FOREIGN KEY (account_id) REFERENCES accounts(id) ) """) + cursor.execute("PRAGMA table_info(account_claim_logs)") + claim_log_columns = [col[1] for col in cursor.fetchall()] + for col_def in [ + ("consumer_key", "TEXT NOT NULL DEFAULT ''"), + ("project_key", "TEXT NOT NULL DEFAULT ''"), + ("claim_read_context", "TEXT DEFAULT NULL"), + ]: + if col_def[0] not in claim_log_columns: + cursor.execute(f"ALTER TABLE account_claim_logs ADD COLUMN {col_def[0]} {col_def[1]}") cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_claim_logs_account_id ON account_claim_logs(account_id) @@ -1050,6 +1066,10 @@ def init_db(database_path: Optional[str] = None): CREATE INDEX IF NOT EXISTS idx_claim_logs_claim_token ON account_claim_logs(claim_token) """) + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_claim_logs_consumer_project + ON account_claim_logs(consumer_key, project_key, action) + """) cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_accounts_pool_status ON accounts(pool_status) diff --git a/outlook_web/repositories/accounts.py b/outlook_web/repositories/accounts.py index 103de64f..d5fcead3 100644 --- a/outlook_web/repositories/accounts.py +++ b/outlook_web/repositories/accounts.py @@ -5,6 +5,7 @@ from outlook_web.db import get_db from outlook_web.security.crypto import decrypt_data, encrypt_data +from outlook_web.services.providers import extract_email_domain COMPACT_SUMMARY_FIELDS = ( "latest_email_subject", @@ -99,6 +100,9 @@ def _hydrate_accounts(rows: List[sqlite3.Row], db: sqlite3.Connection) -> List[D except Exception: account_id_value = None + if not account.get("email_domain"): + account["email_domain"] = extract_email_domain(account.get("email") or "") + account["tags"] = tags_by_account.get(account_id_value, []) if account_id_value is not None else [] accounts.append(account) @@ -251,11 +255,21 @@ def load_accounts_page( def get_account_by_email(email_addr: str) -> Optional[Dict]: """根据邮箱地址获取账号（自动解密敏感字段）""" db = get_db() - cursor = db.execute("SELECT * FROM accounts WHERE email = ?", (email_addr,)) + cursor = db.execute( + """ + SELECT * + FROM accounts + WHERE email = ? COLLATE NOCASE + ORDER BY CASE WHEN email = ? THEN 0 ELSE 1 END + LIMIT 1 + """, + (email_addr, email_addr), + ) row = cursor.fetchone() if not row: return None account = dict(row) + _normalize_account_email_domain(account) _decrypt_account_field(account, "password") _decrypt_account_field(account, "refresh_token") _decrypt_account_field(account, "imap_password") @@ -278,6 +292,7 @@ def get_account_by_id(account_id: int) -> Optional[Dict]: if not row: return None account = dict(row) + _normalize_account_email_domain(account) _decrypt_account_field(account, "password") _decrypt_account_field(account, "refresh_token") _decrypt_account_field(account, "imap_password") @@ -491,11 +506,13 @@ def update_account( if not email_addr: return False + email_domain = extract_email_domain(email_addr) db.execute( """ UPDATE accounts SET email = ?, + email_domain = ?, imap_password = ?, group_id = ?, remark = ?, @@ -506,6 +523,7 @@ def update_account( """, ( email_addr, + email_domain, encrypted_imap_password, group_id, remark, @@ -529,6 +547,7 @@ def update_account( if not email_addr or not new_client_id or not encrypted_refresh_token: return False + email_domain = extract_email_domain(email_addr) db.execute( """ diff --git a/outlook_web/repositories/pool.py b/outlook_web/repositories/pool.py index 0abd70d4..d21c47d5 100644 --- a/outlook_web/repositories/pool.py +++ b/outlook_web/repositories/pool.py @@ -68,14 +68,40 @@ def _parse_claimed_by(claimed_by: Optional[str]) -> tuple[str, str]: """从 claimed_by 字段解析 caller_id 和 task_id（兼容旧格式）。""" if not claimed_by: return "", "" - parts = (claimed_by or ":").split(":", 1) - return parts[0], parts[1] if len(parts) > 1 else "" + parts = str(claimed_by).split("||") + if len(parts) == 4: + return parts[2], parts[3] + legacy_parts = str(claimed_by).split(":", 1) + return legacy_parts[0], legacy_parts[1] if len(legacy_parts) > 1 else "" + + +def _build_claimed_by(*, consumer_key: str, project_key: str, caller_id: str, task_id: str) -> str: + return "||".join([consumer_key, project_key, caller_id, task_id]) + + +def parse_claimed_by(claimed_by: str | None) -> dict: + parts = str(claimed_by or "").split("||") + if len(parts) == 4: + return { + "consumer_key": parts[0], + "project_key": parts[1], + "caller_id": parts[2], + "task_id": parts[3], + } + caller_id, task_id = _parse_claimed_by(claimed_by) + return { + "consumer_key": "", + "project_key": "", + "caller_id": caller_id, + "task_id": task_id, + } def insert_claimed_account( conn: sqlite3.Connection, *, email: str, + consumer_key: str, caller_id: str, task_id: str, lease_seconds: int, @@ -133,7 +159,12 @@ def insert_claimed_account( normalized_email, account_type, provider, - f"{caller_id}:{task_id}", + _build_claimed_by( + consumer_key=consumer_key, + project_key=project_key or "", + caller_id=caller_id, + task_id=task_id, + ), now_str, lease_expires_at_str, token, @@ -157,7 +188,7 @@ def insert_claimed_account( ) # project_key 存在时写入 project usage - if project_key and caller_id: + if project_key and consumer_key: conn.execute( """ INSERT INTO account_project_usage @@ -166,7 +197,7 @@ def insert_claimed_account( ON CONFLICT(account_id, consumer_key, project_key) DO UPDATE SET last_claimed_at = excluded.last_claimed_at """, - (account_id, caller_id, project_key, now_str, now_str), + (account_id, consumer_key, project_key, now_str, now_str), ) conn.execute("COMMIT") @@ -206,6 +237,9 @@ def insert_claimed_account( def claim_atomic( conn: sqlite3.Connection, + *, + consumer_key: str, + project_key: str, caller_id: str, task_id: str, lease_seconds: int, @@ -213,7 +247,6 @@ def claim_atomic( group_id: Optional[int] = None, tags: Optional[List[str]] = None, exclude_recent_minutes: Optional[int] = None, - project_key: Optional[str] = None, email_domain: Optional[str] = None, ) -> Optional[dict]: sql = """ @@ -255,7 +288,7 @@ def claim_atomic( # PR#27 + v22 语义变更: project_key 防同项目复用 # v22 前：NOT EXISTS 即排除（含 claim trace），导致 release 后需删 usage 行（Bug #28） # v22 后：只排除 success_count > 0 的记录，release/expire 产生的 trace 不阻断再次领取 - if project_key and caller_id: + if project_key and consumer_key: sql += """ AND NOT EXISTS ( SELECT 1 FROM account_project_usage apu @@ -265,7 +298,7 @@ def claim_atomic( AND apu.success_count > 0 ) """ - params.append(caller_id) + params.append(consumer_key) params.append(project_key) sql += " ORDER BY RANDOM() LIMIT 1" @@ -295,7 +328,12 @@ def claim_atomic( WHERE id = ? """, ( - f"{caller_id}:{task_id}", + _build_claimed_by( + consumer_key=consumer_key, + project_key=project_key, + caller_id=caller_id, + task_id=task_id, + ), now_str, lease_expires_at_str, token, @@ -314,8 +352,7 @@ def claim_atomic( (account["id"], token, caller_id, task_id, now_str), ) - # PR#27: 记录 project 维度使用（project_key 存在时） - if project_key and caller_id: + if project_key and consumer_key: conn.execute( """ INSERT INTO account_project_usage @@ -324,7 +361,7 @@ def claim_atomic( ON CONFLICT(account_id, consumer_key, project_key) DO UPDATE SET last_claimed_at = excluded.last_claimed_at """, - (account["id"], caller_id, project_key, now_str, now_str), + (account["id"], consumer_key, project_key, now_str, now_str), ) conn.execute("COMMIT") @@ -413,6 +450,8 @@ def release( conn: sqlite3.Connection, account_id: int, claim_token: str, + consumer_key: str, + project_key: str, caller_id: str, task_id: str, reason: Optional[str], @@ -455,6 +494,8 @@ def complete( conn: sqlite3.Connection, account_id: int, claim_token: str, + consumer_key: str, + project_key: str, caller_id: str, task_id: str, result: str, @@ -540,7 +581,7 @@ def complete( """, ( account_id, - caller_id, + consumer_key, effective_claimed_project_key, now_str, now_str, diff --git a/outlook_web/routes/emails.py b/outlook_web/routes/emails.py index deaba6af..63ab33eb 100644 --- a/outlook_web/routes/emails.py +++ b/outlook_web/routes/emails.py @@ -21,6 +21,11 @@ def create_blueprint() -> Blueprint: view_func=emails_controller.api_get_emails, methods=["GET"], ) + bp.add_url_rule( + "/api/emails//stream", + view_func=emails_controller.api_stream_emails, + methods=["GET"], + ) bp.add_url_rule( "/api/emails//extract-verification", view_func=emails_controller.api_extract_verification, diff --git a/outlook_web/routes/external_pool.py b/outlook_web/routes/external_pool.py index c1966252..d395b762 100644 --- a/outlook_web/routes/external_pool.py +++ b/outlook_web/routes/external_pool.py @@ -20,6 +20,11 @@ def create_blueprint(csrf_exempt: Optional[Callable] = None) -> Blueprint: external_pool_controller.api_external_pool_claim_random, ["POST"], ), + ( + "/api/external/pool/claim-domain", + external_pool_controller.api_external_pool_claim_domain, + ["POST"], + ), ( "/api/external/pool/claim-release", external_pool_controller.api_external_pool_claim_release, diff --git a/outlook_web/services/channel_cache.py b/outlook_web/services/channel_cache.py new file mode 100644 index 00000000..90a26250 --- /dev/null +++ b/outlook_web/services/channel_cache.py @@ -0,0 +1,43 @@ +""" +邮箱通道缓存：记住每个邮箱可用的读取方式（graph / imap），避免重复尝试必然失败的通道。 + +- graph: 有 Mail.Read 权限，直接走 Graph API +- imap: 无 Mail.Read 权限或 Graph API 持续失败，跳过 Graph 直走 IMAP +- TTL 1 小时后过期重新探测 +""" + +from __future__ import annotations + +import threading +import time +from typing import Optional + +_channel_cache: dict[str, dict] = {} +_cache_lock = threading.Lock() +_CACHE_TTL = 3600 # 1 hour + + +def get_cached_channel(email: str) -> Optional[str]: + """获取缓存的可用通道，过期返回 None。""" + with _cache_lock: + entry = _channel_cache.get(email) + if entry and entry["expires_at"] > time.time(): + return entry["method"] + if entry: + del _channel_cache[email] + return None + + +def set_cached_channel(email: str, method: str) -> None: + """缓存某邮箱的可用通道。""" + with _cache_lock: + _channel_cache[email] = { + "method": method, + "expires_at": time.time() + _CACHE_TTL, + } + + +def invalidate_channel(email: str) -> None: + """失效某邮箱的通道缓存。""" + with _cache_lock: + _channel_cache.pop(email, None) diff --git a/outlook_web/services/email_cache.py b/outlook_web/services/email_cache.py new file mode 100644 index 00000000..cc290919 --- /dev/null +++ b/outlook_web/services/email_cache.py @@ -0,0 +1,67 @@ +""" +邮件列表缓存：避免 2h 内重复拉取同一邮箱的同一文件夹。 + +- key: (email, folder) +- value: {emails, method, has_more, cached_at} +- TTL: 2 小时 +- 线程安全 +""" + +from __future__ import annotations + +import threading +import time +from typing import Any, Dict, List, Optional + +_email_cache: dict[str, dict] = {} +_cache_lock = threading.Lock() +_CACHE_TTL = 7200 # 2 hours + + +def _make_key(email: str, folder: str) -> str: + return f"{email}|{folder}" + + +def get_cached_emails(email: str, folder: str) -> Optional[Dict[str, Any]]: + """命中缓存返回 {emails, method, has_more}，未命中或过期返回 None。""" + key = _make_key(email, folder) + with _cache_lock: + entry = _email_cache.get(key) + if entry and entry["expires_at"] > time.time(): + return { + "emails": entry["emails"], + "method": entry["method"], + "has_more": entry["has_more"], + } + if entry: + del _email_cache[key] + return None + + +def set_cached_emails( + email: str, + folder: str, + emails: List[Dict], + method: str, + has_more: bool = False, +) -> None: + """写入缓存。""" + key = _make_key(email, folder) + with _cache_lock: + _email_cache[key] = { + "emails": emails, + "method": method, + "has_more": has_more, + "expires_at": time.time() + _CACHE_TTL, + } + + +def invalidate_email_cache(email: str, folder: Optional[str] = None) -> None: + """失效缓存。folder=None 时清除该邮箱所有文件夹。""" + with _cache_lock: + if folder: + _email_cache.pop(_make_key(email, folder), None) + else: + keys_to_remove = [k for k in _email_cache if k.startswith(f"{email}|")] + for k in keys_to_remove: + del _email_cache[k] diff --git a/outlook_web/services/graph.py b/outlook_web/services/graph.py index b1e03fe9..98278830 100644 --- a/outlook_web/services/graph.py +++ b/outlook_web/services/graph.py @@ -75,12 +75,18 @@ def get_access_token_graph_result(client_id: str, refresh_token: str, proxy_url: # 根据 Microsoft Learn 文档：refresh token 可能会在每次使用时"自我替换"，应保存新的 refresh_token（如有）。 new_refresh_token = payload.get("refresh_token") + + # 检查返回的 scope 是否包含 Mail.Read 权限 + scope = payload.get("scope", "") + has_mail_read = "Mail.Read" in scope or "Mail.ReadWrite" in scope + return { "success": True, "access_token": access_token, "refresh_token": new_refresh_token, "new_refresh_token": new_refresh_token, - "scope": payload.get("scope", ""), + "scope": scope, + "has_mail_read": has_mail_read, } except Exception as exc: return { @@ -121,6 +127,20 @@ def get_emails_graph( if not token_result.get("success"): return {"success": False, "error": token_result.get("error")} + # 无 Mail.Read 权限时直接跳过，不浪费一次 Graph API 请求 + if not token_result.get("has_mail_read"): + return { + "success": False, + "no_mail_permission": True, + "error": build_error_payload( + "GRAPH_NO_MAIL_PERMISSION", + "Token 无 Mail.Read 权限，跳过 Graph API", + "GraphPermissionError", + 403, + token_result.get("scope", ""), + ), + } + access_token = token_result.get("access_token") scope = token_result.get("scope", "") if not has_mail_read_permission(scope): diff --git a/outlook_web/services/imap.py b/outlook_web/services/imap.py index 95d1e398..b7e33fe4 100644 --- a/outlook_web/services/imap.py +++ b/outlook_web/services/imap.py @@ -428,18 +428,36 @@ def get_emails_imap_with_server( return {"success": True, "emails": []} paged_ids = message_ids[start_idx:end_idx][::-1] + + # 批量 FETCH：一次请求取所有消息（避免 N 次串行网络往返） + id_list = b",".join(paged_ids) emails_data = [] + status, msg_data = connection.fetch(id_list, "(RFC822)") - 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} + if status == "OK" and msg_data: + for item in msg_data: + if not isinstance(item, tuple) or len(item) < 2: + continue + try: + raw_email = item[1] + msg = email.message_from_bytes(raw_email) + seq = item[0].split(b" ", 1)[0] if isinstance(item[0], bytes) else b"0" + msg_id_str = seq.decode() if isinstance(seq, bytes) else str(seq) + + 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: + continue + 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) @@ -545,7 +563,14 @@ def fetch_and_detail_imap_with_server( if status != "OK": return {"success": True, "emails": [], "detail": None} - for i, (msg_id_str, raw_email) in enumerate(_parse_batch_fetch_response(all_data or [])): + 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 = { @@ -771,3 +796,116 @@ def delete_emails_imap( return {"success": False, "error": "IMAP 删除暂不支持 (ID 格式不兼容)"} except Exception as e: return {"success": False, "error": str(e)} +IMAP_SERVER_OLD = "outlook.office365.com" + + +def stream_emails_imap( + account: str, + client_id: str, + refresh_token: str, + folder: str = "inbox", + skip: int = 0, + top: int = 20, +): + """Generator：逐条 yield 邮件，供 SSE 流式推送。并发尝试新旧 IMAP 服务器。""" + from concurrent.futures import ThreadPoolExecutor, as_completed + + token_result = get_access_token_imap_result(client_id, refresh_token) + if not token_result.get("success"): + yield {"type": "error", "data": token_result.get("error", {})} + return + + access_token = token_result.get("access_token") + + # 并发尝试两个 IMAP 服务器，用第一个成功连接的 + def _try_connect(server): + try: + conn = imaplib.IMAP4_SSL(server, IMAP_PORT) + auth_string = f"user={account}\1auth=Bearer {access_token}\1\1".encode("utf-8") + conn.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"], + } + for imap_folder in folder_map.get((folder or "").lower(), ['"INBOX"']): + try: + status, _ = conn.select(imap_folder, readonly=True) + if status == "OK": + return conn, server + except Exception: + continue + conn.logout() + return None, server + except Exception: + return None, server + + connection = None + winner_server = None + with ThreadPoolExecutor(max_workers=2) as ex: + futures = [ex.submit(_try_connect, s) for s in [IMAP_SERVER_NEW, IMAP_SERVER_OLD]] + for f in as_completed(futures): + conn, srv = f.result() + if conn and not connection: + connection = conn + winner_server = srv + elif conn: + try: + conn.logout() + except Exception: + pass + + if not connection: + yield {"type": "error", "data": {"code": "IMAP_CONNECT_FAILED", "message": "所有 IMAP 服务器连接失败"}} + return + + try: + status, messages = connection.search(None, "ALL") + if status != "OK" or not messages or not messages[0]: + yield {"type": "done", "method": "IMAP", "count": 0} + return + + 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: + yield {"type": "done", "method": "IMAP", "count": 0} + return + + paged_ids = message_ids[start_idx:end_idx][::-1] + method = "IMAP (New)" if winner_server == IMAP_SERVER_NEW else "IMAP (Old)" + count = 0 + + for msg_id in paged_ids: + try: + status, msg_data = connection.fetch(msg_id, "(RFC822)") + if status == "OK" and msg_data and msg_data[0]: + raw_email = msg_data[0][1] + msg = email.message_from_bytes(raw_email) + body_preview = get_email_body(msg) + count += 1 + yield { + "type": "email", + "data": { + "id": (msg_id.decode() if isinstance(msg_id, bytes) else str(msg_id)), + "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: + continue + + yield {"type": "done", "method": method, "count": count} + finally: + try: + connection.logout() + except Exception: + pass + + + diff --git a/outlook_web/services/pool.py b/outlook_web/services/pool.py index c8a862f3..85458249 100644 --- a/outlook_web/services/pool.py +++ b/outlook_web/services/pool.py @@ -90,12 +90,17 @@ def _validate_lease_seconds(lease_seconds: int, max_lease: int = 3600) -> None: raise PoolServiceError(f"lease_seconds 不能超过 {max_lease} 秒", "lease_seconds_too_large") -def _validate_project_key(project_key: Optional[str]) -> Optional[str]: - if project_key is None: - return None - pk = project_key.strip() +def _validate_consumer_key(consumer_key: str) -> str: + normalized = str(consumer_key or "").strip() + if not normalized: + raise PoolServiceError("consumer_key 不能为空", "consumer_key_empty", http_status=403) + return normalized + + +def _validate_project_key(project_key: str) -> str: + pk = str(project_key or "").strip() if not pk: - return None + raise PoolServiceError("project_key 不能为空", "project_key_empty") if len(pk) > PROJECT_KEY_MAX_LEN: raise PoolServiceError(f"project_key 超过最大长度 {PROJECT_KEY_MAX_LEN}", "project_key_too_long") return pk @@ -151,16 +156,18 @@ def _is_project_reuse_eligible_account( def claim_random( *, + consumer_key: str, + project_key: str, caller_id: str, task_id: str, provider: Optional[str] = None, - project_key: Optional[str] = None, email_domain: Optional[str] = None, ) -> dict: + consumer_key = _validate_consumer_key(consumer_key) + project_key = _validate_project_key(project_key) _validate_caller_id(caller_id) _validate_task_id(task_id) provider = _validate_provider(provider) - project_key = _validate_project_key(project_key) email_domain = _validate_email_domain(email_domain) conn = create_sqlite_connection() @@ -172,11 +179,12 @@ def claim_random( try: account = pool_repo.claim_atomic( conn, + consumer_key=consumer_key, + project_key=project_key, caller_id=caller_id, task_id=task_id, lease_seconds=default_lease, provider=provider, - project_key=project_key, email_domain=email_domain, ) except pool_repo.PoolRepositoryError as e: @@ -209,6 +217,7 @@ def claim_random( inserted = pool_repo.insert_claimed_account( conn, email=created_email, + consumer_key=consumer_key, caller_id=caller_id, task_id=task_id, lease_seconds=default_lease, @@ -237,6 +246,8 @@ def _validate_claim_ownership( *, action: str, claim_token: str, + consumer_key: str, + project_key: str, caller_id: str, task_id: str, ) -> None: @@ -251,7 +262,19 @@ def _validate_claim_ownership( ) if row.get("claim_token") != claim_token: raise PoolServiceError("claim_token 不匹配", "token_mismatch", http_status=403) - if row.get("claimed_by") != f"{caller_id}:{task_id}": + + claimed_by = str(row.get("claimed_by") or "") + if "||" in claimed_by: + claimed_context = pool_repo.parse_claimed_by(claimed_by) + if claimed_context["consumer_key"] and claimed_context["consumer_key"] != consumer_key: + raise PoolServiceError("consumer_key 与领取记录不一致", "consumer_mismatch", http_status=403) + if claimed_context["project_key"] and claimed_context["project_key"] != project_key: + raise PoolServiceError("project_key 与领取记录不一致", "project_mismatch", http_status=403) + if claimed_context["caller_id"] != caller_id: + raise PoolServiceError("caller_id 与领取记录不一致", "caller_mismatch", http_status=403) + if claimed_context["task_id"] != task_id: + raise PoolServiceError("task_id 与领取记录不一致", "task_mismatch", http_status=403) + elif claimed_by != f"{caller_id}:{task_id}": raise PoolServiceError( "caller_id 或 task_id 与领取记录不一致", "caller_mismatch", @@ -261,6 +284,8 @@ def _validate_claim_ownership( def release_claim( *, + consumer_key: str, + project_key: str, account_id: int, claim_token: str, caller_id: str, @@ -268,6 +293,8 @@ def release_claim( reason: Optional[str] = None, ) -> None: """释放已领取的邮箱账号（不计入成功/失败统计，直接回 available）。""" + consumer_key = _validate_consumer_key(consumer_key) + project_key = _validate_project_key(project_key) _validate_caller_id(caller_id) _validate_task_id(task_id) if not claim_token or not claim_token.strip(): @@ -282,7 +309,13 @@ def release_claim( temp_id = pool_repo.temp_id_from_account_id(account_id) temp_row = pool_repo.get_temp_mailbox_pool_row(conn, temp_id) _validate_claim_ownership( - temp_row, action="release", claim_token=claim_token, caller_id=caller_id, task_id=task_id + temp_row, + action="release", + claim_token=claim_token, + consumer_key=consumer_key, + project_key=project_key, + caller_id=caller_id, + task_id=task_id, ) pool_repo.release_temp_mailbox(conn, temp_id, claim_token, caller_id, task_id, reason) return @@ -295,17 +328,30 @@ def release_claim( dict(row) if row is not None else None, action="release", claim_token=claim_token, + consumer_key=consumer_key, + project_key=project_key, caller_id=caller_id, task_id=task_id, ) - pool_repo.release(conn, account_id, claim_token, caller_id, task_id, reason) + pool_repo.release( + conn, + account_id, + claim_token, + consumer_key, + project_key, + caller_id, + task_id, + reason, + ) finally: conn.close() def complete_claim( *, + consumer_key: str, + project_key: str, account_id: int, claim_token: str, caller_id: str, @@ -318,6 +364,8 @@ def complete_claim( 返回账号的新 pool_status。 """ + consumer_key = _validate_consumer_key(consumer_key) + project_key = _validate_project_key(project_key) _validate_caller_id(caller_id) _validate_task_id(task_id) if not claim_token or not claim_token.strip(): @@ -337,7 +385,13 @@ def complete_claim( temp_id = pool_repo.temp_id_from_account_id(account_id) temp_row = pool_repo.get_temp_mailbox_pool_row(conn, temp_id) _validate_claim_ownership( - temp_row, action="complete", claim_token=claim_token, caller_id=caller_id, task_id=task_id + temp_row, + action="complete", + claim_token=claim_token, + consumer_key=consumer_key, + project_key=project_key, + caller_id=caller_id, + task_id=task_id, ) return pool_repo.complete_temp_mailbox(conn, temp_id, claim_token, caller_id, task_id, result, detail) @@ -355,6 +409,8 @@ def complete_claim( dict(row) if row is not None else None, action="complete", claim_token=claim_token, + consumer_key=consumer_key, + project_key=project_key, caller_id=caller_id, task_id=task_id, ) @@ -373,6 +429,8 @@ def complete_claim( conn, account_id, claim_token, + consumer_key, + project_key, caller_id, task_id, result, diff --git a/outlook_web/services/providers.py b/outlook_web/services/providers.py index cdcad0cd..e3c2cc04 100644 --- a/outlook_web/services/providers.py +++ b/outlook_web/services/providers.py @@ -3,9 +3,8 @@ from typing import Any, Dict, List, Optional # 对齐：PRD-00005 / FD-00005 / TDD-00005 / PRD-00006 / FD-00006 -# 职责：集中维护“邮箱提供商”元数据与 IMAP 文件夹映射，避免前后端重复维护默认 host/port 与 folder 兼容策略。 +# 职责：集中维护邮箱提供商元数据、邮箱域名归一化规则，以及 provider/domain 一致性校验。 -# 邮箱提供商配置（用于前端选择与默认 IMAP host/port） MAIL_PROVIDERS: Dict[str, Dict[str, Any]] = { "outlook": { "label": "Outlook", @@ -65,33 +64,24 @@ }, } -# FD-00006: 域名 → provider 反向映射（用于 auto 模式域名推断） DOMAIN_PROVIDER_MAP: Dict[str, str] = { - # Gmail "gmail.com": "gmail", "googlemail.com": "gmail", - # QQ "qq.com": "qq", "foxmail.com": "qq", - # 163 "163.com": "163", - # 126 "126.com": "126", - # Yahoo "yahoo.com": "yahoo", "yahoo.co.jp": "yahoo", "yahoo.co.uk": "yahoo", - # 阿里云 "aliyun.com": "aliyun", "alimail.com": "aliyun", - # 微软（2段格式按 IMAP 兜底处理，OAuth 至少4段） "outlook.com": "outlook", "hotmail.com": "outlook", "live.com": "outlook", "live.cn": "outlook", } -# FD-00006: provider → 自动分组名映射 PROVIDER_GROUP_NAME: Dict[str, str] = { "outlook": "Outlook", "gmail": "Gmail", @@ -105,19 +95,59 @@ "gptmail": "临时邮箱", } -# FD-00006: 已知 provider key 集合（用于 3 段格式校验） -KNOWN_PROVIDER_KEYS: set = set(MAIL_PROVIDERS.keys()) +KNOWN_PROVIDER_KEYS: set[str] = set(MAIL_PROVIDERS.keys()) +PROVIDER_FAMILY_DOMAINS: Dict[str, set[str]] = {} +for domain_name, provider_key in DOMAIN_PROVIDER_MAP.items(): + PROVIDER_FAMILY_DOMAINS.setdefault(provider_key, set()).add(domain_name) -def infer_provider_from_email(email: str) -> Optional[str]: - """从邮箱地址推断 provider。返回 provider key 或 None。""" - if not email or "@" not in email: + +def normalize_email_domain(value: str | None) -> str: + text = str(value or "").strip().lower() + if not text: + return "" + if "@" in text: + text = text.rsplit("@", 1)[-1] + return text.strip().strip(".") + + +def extract_email_domain(email: str | None) -> str: + text = str(email or "").strip().lower() + if "@" not in text: + return "" + return normalize_email_domain(text.rsplit("@", 1)[-1]) + + +def infer_provider_from_email(email: str | None) -> Optional[str]: + domain = extract_email_domain(email) + if not domain: return None - domain = email.rsplit("@", 1)[-1].strip().lower() return DOMAIN_PROVIDER_MAP.get(domain) -# provider -> 逻辑文件夹名（inbox/junkemail/deleteditems）-> 候选 IMAP 文件夹名列表 +def infer_provider_from_domain(domain: str | None) -> Optional[str]: + normalized = normalize_email_domain(domain) + if not normalized: + return None + return DOMAIN_PROVIDER_MAP.get(normalized) + + +def provider_supports_email_domain(provider: str | None, email_domain: str | None) -> bool: + provider_key = str(provider or "").strip().lower() + normalized_domain = normalize_email_domain(email_domain) + if not provider_key or not normalized_domain: + return False + inferred_provider = infer_provider_from_domain(normalized_domain) + if inferred_provider is not None: + return inferred_provider == provider_key + return provider_key == "custom" + + +def get_provider_domains(provider: str | None) -> set[str]: + provider_key = str(provider or "").strip().lower() + return set(PROVIDER_FAMILY_DOMAINS.get(provider_key, set())) + + PROVIDER_FOLDER_MAP: Dict[str, Dict[str, List[str]]] = { "gmail": { "inbox": ["INBOX"], @@ -148,14 +178,8 @@ def infer_provider_from_email(email: str) -> Optional[str]: def get_imap_folder_candidates(provider: str, folder: str) -> List[str]: - """ - 根据 provider 和逻辑文件夹名（inbox/junkemail/deleteditems）， - 返回候选 IMAP 文件夹名列表（按优先级排序）。 - 不存在的 provider 退回 _default。 - """ provider_key = (provider or "").strip() or "_default" folder_key = (folder or "").strip().lower() or "inbox" - folder_map = PROVIDER_FOLDER_MAP.get(provider_key, PROVIDER_FOLDER_MAP["_default"]) return folder_map.get(folder_key, PROVIDER_FOLDER_MAP["_default"].get(folder_key, ["INBOX"])) @@ -212,7 +236,6 @@ def get_provider_domains(provider: str) -> List[str]: def get_provider_list() -> List[Dict[str, Any]]: - """返回供前端展示的 provider 列表（auto 在最前，outlook 其次，custom 在后）""" result: List[Dict[str, Any]] = [ { "key": "auto", @@ -225,13 +248,13 @@ def get_provider_list() -> List[Dict[str, Any]]: for key in order: if key not in MAIL_PROVIDERS: continue - p = MAIL_PROVIDERS[key] + provider = MAIL_PROVIDERS[key] result.append( { "key": key, - "label": p.get("label", key), - "account_type": p.get("account_type", "imap" if key != "outlook" else "outlook"), - "note": p.get("note", ""), + "label": provider.get("label", key), + "account_type": provider.get("account_type", "imap" if key != "outlook" else "outlook"), + "note": provider.get("note", ""), } ) return result diff --git a/outlook_web/services/verification_channel_routing.py b/outlook_web/services/verification_channel_routing.py index b6243408..767b0657 100644 --- a/outlook_web/services/verification_channel_routing.py +++ b/outlook_web/services/verification_channel_routing.py @@ -408,13 +408,22 @@ def extract_verification_for_outlook( reverse=True, )[0] + latest_id = str(latest.get("id") or "") if channel.startswith("imap_"): detail = channel_result.get("detail") + detail_id = str((detail or {}).get("id") or "") + if latest_id and detail_id and detail_id != latest_id: + detail = fetch_email_detail_for_channel( + account=account, + channel=channel, + message_id=latest_id, + proxy_url=proxy_url, + ) else: detail = fetch_email_detail_for_channel( account=account, channel=channel, - message_id=latest.get("id", ""), + message_id=latest_id, proxy_url=proxy_url, ) diff --git a/outlook_web/services/verification_extractor.py b/outlook_web/services/verification_extractor.py index 340df2c8..3d0d4a7b 100644 --- a/outlook_web/services/verification_extractor.py +++ b/outlook_web/services/verification_extractor.py @@ -154,7 +154,7 @@ def smart_extract_verification_code(email_content: str) -> Optional[str]: # 过滤掉纯字母的匹配（验证码通常包含数字） for match in matches: if any(c.isdigit() for c in match): - return match.upper() + return match return None @@ -183,8 +183,6 @@ def fallback_extract_verification_code(email_content: str) -> Optional[str]: # 过滤规则 filtered = [] for match in matches: - match_upper = match.upper() - # 必须包含至少一个数字 if not any(c.isdigit() for c in match): continue @@ -209,7 +207,7 @@ def fallback_extract_verification_code(email_content: str) -> Optional[str]: if 2020 <= num <= 2030: continue - filtered.append(match_upper) + filtered.append(match) return filtered[0] if filtered else None @@ -460,7 +458,7 @@ def _smart_extract_code_by_keywords(email_content: str, code_re: re.Pattern) -> for m in code_re.finditer(context): value = m.group(0) if value and any(c.isdigit() for c in value): - return value.upper() + return value return None @@ -492,7 +490,7 @@ def _fallback_extract_code(email_content: str, code_re: re.Pattern) -> Optional[ if 2020 <= num <= 2030: continue - candidates.append(value.upper()) + candidates.append(value) return candidates[0] if candidates else None @@ -1055,7 +1053,7 @@ def _apply_output_policy(payload: Dict[str, Any]) -> Dict[str, Any]: updated = False if ai_code: - result["verification_code"] = ai_code.upper() + result["verification_code"] = ai_code result["code_confidence"] = "high" if ai_confidence == "high" else "low" updated = True if ai_link: diff --git a/registration-mail-pool-api.en.md b/registration-mail-pool-api.en.md index bc3c3006..76bad4de 100644 --- a/registration-mail-pool-api.en.md +++ b/registration-mail-pool-api.en.md @@ -104,7 +104,8 @@ Time fields use ISO 8601, for example: | Endpoint | Purpose | Recommended | | --- | --- | --- | -| `POST /api/external/pool/claim-random` | claim a mailbox | Common | +| `POST /api/external/pool/claim-random` | claim a mailbox, optionally filtered by domain | Common | +| `POST /api/external/pool/claim-domain` | claim a mailbox from a required domain | Common | | `POST /api/external/pool/claim-release` | release a mailbox | Common | | `POST /api/external/pool/claim-complete` | submit the task result | Common | | `GET /api/external/pool/stats` | inspect pool counts | Optional | @@ -324,12 +325,15 @@ Request body: | `caller_id` | string | Yes | caller instance, node, or worker identity | | `task_id` | string | Yes | unique task ID | | `provider` | string | No | provider filter: `outlook` / `imap` / `custom` / `cloudflare_temp_mail` | +| `project_key` | string | No | project-level reuse and duplicate-prevention context | +| `email_domain` | string | No | mailbox domain filter; when provided, only eligible mailboxes in that domain are claimed | Current implementation notes: -- the current pool API supports filtering only by `provider` -- `outlook.com`, `hotmail.com`, `live.com`, and `live.cn` all map to `provider=outlook` -- the current external pool API does not support extra filtering by domain, group, or tags +- `claim-random` claims an eligible mailbox randomly by default; when `email_domain` is provided, it claims randomly within that domain +- for a clearer domain-specific contract, use `POST /api/external/pool/claim-domain`; that endpoint requires `email_domain` +- `outlook.com`, `hotmail.com`, `live.com`, and `live.cn` all map to `provider=outlook`; use `email_domain` when those domains need to be distinguished +- the current external pool API does not support claiming a specific full mailbox, group, or tag - when `provider=cloudflare_temp_mail` and no eligible mailbox exists in pool, the service dynamically creates a CF temp mailbox and returns it as claimed Success response fields: @@ -391,6 +395,35 @@ No-available response example: } ``` +### `POST /api/external/pool/claim-domain` + +Purpose: claim a mailbox from a specific domain. This is the explicit endpoint for `claim-random + email_domain`; internally it reuses the same pool claim, lease, audit, and completion state machine. + +Request body: + +| Parameter | Type | Required | Description | +| --- | --- | --- | --- | +| `caller_id` | string | Yes | caller instance, node, or worker identity | +| `task_id` | string | Yes | unique task ID | +| `email_domain` | string | Yes | mailbox domain to claim from, for example `zerodotsix.top` | +| `provider` | string | No | optional provider filter | +| `project_key` | string | No | project-level reuse and duplicate-prevention context | + +If `email_domain` is missing or blank, the endpoint returns HTTP `400` with code `EMAIL_DOMAIN_REQUIRED`. + +Copy-paste example: + +```bash +curl -X POST https://api.example.com/api/external/pool/claim-domain \ + -H "X-API-Key: YOUR_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "caller_id": "reg-worker-001", + "task_id": "task-20260409-0001", + "email_domain": "zerodotsix.top" + }' +``` + ### `POST /api/external/pool/claim-release` Request body: diff --git a/static/js/features/emails.js b/static/js/features/emails.js index ea9e4499..8a55a23f 100644 --- a/static/js/features/emails.js +++ b/static/js/features/emails.js @@ -3,24 +3,6 @@ // 模块内变量：存储上次获取邮件失败的错误详情 let lastFetchErrorDetails = {}; - function resolveEmailSortTimestamp(email) { - const rawDate = email && (email.receivedDateTime || email.date || email.created_at || email.received_at); - const parsed = Date.parse(String(rawDate || '')); - return Number.isFinite(parsed) ? parsed : Number.NEGATIVE_INFINITY; - } - - function sortEmailsByNewestFirst(list) { - const source = Array.isArray(list) ? list : []; - return source - .map((item, index) => ({ item, index, timestamp: resolveEmailSortTimestamp(item) })) - .sort((a, b) => (b.timestamp - a.timestamp) || (a.index - b.index)) - .map(entry => entry.item); - } - - if (typeof window !== 'undefined') { - window.sortEmailsByNewestFirst = sortEmailsByNewestFirst; - } - // 加载邮件列表 async function loadEmails(email, forceRefresh = false) { const container = document.getElementById('emailList'); @@ -33,13 +15,11 @@ const cacheKey = `${email}_${currentFolder}`; if (!forceRefresh && emailListCache[cacheKey]) { const cache = emailListCache[cacheKey]; - currentEmails = sortEmailsByNewestFirst(cache.emails || []); + currentEmails = cache.emails; hasMoreEmails = cache.has_more; currentSkip = cache.skip; currentMethod = cache.method || 'graph'; - cache.emails = currentEmails; - // 恢复 UI const methodTag = document.getElementById('methodTag'); methodTag.textContent = currentMethod; @@ -65,78 +45,105 @@ container.innerHTML = `
${translateAppTextLocal('获取中…')}
`; - try { - // 每次只查询20封邮件 - const response = await fetch( - `/api/emails/${encodeURIComponent(email)}?method=${currentMethod}&folder=${currentFolder}&skip=0&top=20` - ); - const data = await response.json(); + // SSE 流式加载：每收到一封邮件立即渲染，不等全部完成 + const forceParam = forceRefresh ? '&force=1' : ''; + const streamUrl = `/api/emails/${encodeURIComponent(email)}/stream?folder=${currentFolder}&skip=0&top=20${forceParam}`; + const evtSource = new EventSource(streamUrl); + let streamEmails = []; + let firstEmail = true; + const clickHandler = isTempEmailGroup ? 'getTempEmailDetail' : 'selectEmail'; - if (data.success) { - const sortedEmails = sortEmailsByNewestFirst(data.emails || []); - currentEmails = sortedEmails; - currentMethod = data.method === 'Graph API' ? 'graph' : 'imap'; - hasMoreEmails = data.has_more; - if (typeof syncAccountSummaryToAccountCache === 'function' && data.account_summary) { - syncAccountSummaryToAccountCache(email, data.account_summary); - } + evtSource.addEventListener('email', function(e) { + const emailData = JSON.parse(e.data); + streamEmails.push(emailData); - if (typeof syncAccountSummaryToAccountCache === 'function' && data.account_summary) { - syncAccountSummaryToAccountCache(email, data.account_summary); - } + // 首封邮件到达时清除 loading + if (firstEmail) { + container.innerHTML = ''; + firstEmail = false; + } - // 保存到缓存 - emailListCache[cacheKey] = { - emails: currentEmails, - has_more: hasMoreEmails, - skip: currentSkip, - method: currentMethod - }; + // 逐条 DOM append + const index = streamEmails.length - 1; + const initial = (emailData.from || '?')[0].toUpperCase(); + const div = document.createElement('div'); + div.className = `email-item${emailData.is_read === false ? ' unread' : ''}`; + div.setAttribute('onclick', `${clickHandler}('${emailData.id}', ${index})`); + div.innerHTML = ` + + + + + `; + container.appendChild(div); - // 显示使用的方法和邮件数量 - const methodTag = document.getElementById('methodTag'); - methodTag.textContent = data.method; - methodTag.style.display = 'inline'; + // 实时更新计数 + document.getElementById('emailCount').textContent = `(${streamEmails.length})`; + }); - document.getElementById('emailCount').textContent = `(${data.emails.length})`; + evtSource.addEventListener('done', function(e) { + const info = JSON.parse(e.data); + evtSource.close(); - document.getElementById('emailCount').textContent = `(${currentEmails.length})`; + currentEmails = streamEmails; + currentMethod = info.method.includes('Graph API') ? 'graph' : 'imap'; + hasMoreEmails = info.has_more || false; - renderEmailList(currentEmails); - } else { - // 显示详细的多方法失败弹框 - if (data.details) { - showEmailFetchErrorModal(data.details); - } else { - handleApiError(data, '获取邮件失败'); - } + // 如果没收到任何邮件 + if (streamEmails.length === 0) { container.innerHTML = `
- ⚠️

${translateAppTextLocal('获取邮件失败，')}${translateAppTextLocal('点击查看详情')}

+ 📭 +

${translateAppTextLocal('收件箱为空')}

`; - lastFetchErrorDetails = data.details || {}; - // 绑定事件监听器 - const errorLink = document.getElementById('showEmailErrorLink'); - if (errorLink) { - errorLink.addEventListener('click', () => showEmailFetchErrorModal(lastFetchErrorDetails)); - } } - } catch (error) { - console.error('加载邮件列表失败:', error); - container.innerHTML = ` -
- ⚠️

${translateAppTextLocal('网络错误，请重试')}

-
- `; - } finally { + + // 更新 method tag + const methodTag = document.getElementById('methodTag'); + methodTag.textContent = info.method; + methodTag.style.display = 'inline'; + + // 保存到缓存 + emailListCache[cacheKey] = { + emails: currentEmails, + has_more: hasMoreEmails, + skip: currentSkip, + method: currentMethod + }; + // 启用按钮 if (refreshBtn) { refreshBtn.disabled = false; refreshBtn.textContent = translateAppTextLocal('获取邮件'); } folderTabs.forEach(tab => tab.disabled = false); - } + }); + + evtSource.addEventListener('error', function(e) { + // SSE 自身 error event（非服务端 error event） + if (evtSource.readyState === EventSource.CLOSED) return; + evtSource.close(); + + if (streamEmails.length === 0) { + container.innerHTML = ` +
+ ⚠️

${translateAppTextLocal('获取邮件失败')}

+
+ `; + } + if (refreshBtn) { + refreshBtn.disabled = false; + refreshBtn.textContent = translateAppTextLocal('获取邮件'); + } + folderTabs.forEach(tab => tab.disabled = false); + }); } // 渲染邮件列表 @@ -144,7 +151,7 @@ let selectedEmailIds = new Set(); let isBatchSelectMode = false; - function renderEmailList(emails, options = {}) { + function renderEmailList(emails) { const container = document.getElementById('emailList'); const actionBar = document.getElementById('emailBatchActionBar'); @@ -157,7 +164,6 @@ `; selectedEmailIds.clear(); updateEmailBatchActionBar(); - if (options.scrollToTop !== false) container.scrollTop = 0; return; } @@ -185,9 +191,6 @@ `}).join(''); - // Issue #52: 加载/刷新后自动回到列表顶部，避免滚动位置乱跑 - if (options.scrollToTop !== false) container.scrollTop = 0; - updateEmailBatchActionBar(); } @@ -201,7 +204,7 @@ // Re-render to update checkbox UI (or efficiently update DOM) // For simplicity, we just find the checkbox and update it // implementation below is cheap - renderEmailList(currentEmails, { scrollToTop: false }); + renderEmailList(currentEmails); } function updateEmailBatchActionBar() { @@ -229,60 +232,6 @@ return isTempEmailGroup || currentPage === 'temp-emails' ? 'temp' : 'mailbox'; } - // detail-focus 断点阈值：低于此宽度时切换列表/详情为互斥模式 - // 注意: CSS 平板断点为 1024px，此处 900px 为功能切换阈值而非布局断点 - function isNarrowWorkspaceViewport() { - return window.innerWidth <= 900; - } - - // 邮箱列表/详情互斥切换 — 窄视口下点击邮件时隐藏列表、全宽展示详情 - // 被调用方: accounts.js(切换账户重置)、emails.js(点击邮件/返回列表) - // CSS 配套: #emailListPanel.detail-focus 规则(平板+移动端) - function setMailboxDetailFocus(active) { - const panel = document.getElementById('emailListPanel'); - if (!panel) return; - const shouldFocus = Boolean(active) && isNarrowWorkspaceViewport(); - panel.classList.toggle('detail-focus', shouldFocus); - // 内联样式作为 CSS 的即时保障，避免布局闪烁 - const listEl = document.getElementById('emailList'); - const detailEl = document.getElementById('emailDetailSection'); - if (shouldFocus) { - if (listEl) listEl.style.display = 'none'; - if (detailEl) detailEl.style.display = 'flex'; - } else if (isNarrowWorkspaceViewport()) { - if (listEl) listEl.style.display = ''; - if (detailEl) detailEl.style.display = 'none'; - } else { - // 桌面端：退回 CSS 控制，清除内联覆盖 - if (listEl) listEl.style.display = ''; - if (detailEl) detailEl.style.display = ''; - } - } - - // 临时邮箱消息列表/详情互斥切换 — 与 setMailboxDetailFocus 对称设计 - // 被调用方: temp_emails.js(点击消息/刷新列表)、emails.js(切换回邮箱列表) - // CSS 配套: .workspace.workspace-temp-emails.detail-focus 规则(平板+移动端) - function setTempDetailFocus(active) { - const workspace = document.querySelector('.workspace.workspace-temp-emails'); - const messagePanel = document.getElementById('tempEmailMessagePanel'); - const detailPanel = document.getElementById('tempEmailDetailSection'); - if (!workspace) return; - - const shouldFocus = Boolean(active) && isNarrowWorkspaceViewport(); - workspace.classList.toggle('detail-focus', shouldFocus); - - if (shouldFocus) { - if (messagePanel) messagePanel.style.display = 'none'; - if (detailPanel) detailPanel.style.display = 'flex'; - } else if (isNarrowWorkspaceViewport()) { - if (messagePanel) messagePanel.style.display = ''; - if (detailPanel) detailPanel.style.display = 'none'; - } else { - if (messagePanel) messagePanel.style.display = ''; - if (detailPanel) detailPanel.style.display = ''; - } - } - function getEmailDetailRefs(options = {}) { const source = resolveEmailDetailSource(options); if (source === 'temp') { @@ -328,7 +277,7 @@ return; } if (refs.section) { - refs.section.style.display = 'none'; + refs.section.style.display = 'flex'; } } @@ -466,7 +415,7 @@ currentEmails = currentEmails.filter(e => !deletedIds.has(e.id)); selectedEmailIds.clear(); - renderEmailList(currentEmails, { scrollToTop: false }); + renderEmailList(currentEmails); // If current viewed email was deleted, clear view if (currentEmailDetail && deletedIds.has(currentEmailDetail.id)) { @@ -518,7 +467,7 @@ const deletedId = currentEmailDetail.id; currentEmails = currentEmails.filter(email => email.id !== deletedId); currentEmailDetail = null; - renderEmailList(currentEmails, { scrollToTop: false }); + renderEmailList(currentEmails); const tempContainer = document.getElementById('tempEmailMessageList'); if (tempContainer && typeof renderTempEmailMessageList === 'function') { @@ -555,7 +504,6 @@ // 显示工具栏 setEmailDetailToolbarVisibility(true, { source: 'mailbox' }); - setMailboxDetailFocus(true); // 加载邮件详情 const container = refs.container; @@ -704,12 +652,12 @@ } else if (typeof DOMPurify !== 'undefined') { // 使用 DOMPurify 净化 HTML 内容，防止 XSS 攻击 sanitizedBody = DOMPurify.sanitize(renderableBody, { - ALLOWED_TAGS: ['a', 'b', 'i', 'u', 'strong', 'em', 'p', 'br', 'div', 'span', 'img', 'table', 'tr', 'td', 'th', 'thead', 'tbody', 'ul', 'ol', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'blockquote', 'pre', 'code', 'style'], + ALLOWED_TAGS: ['a', 'b', 'i', 'u', 'strong', 'em', 'p', 'br', 'div', 'span', 'img', 'table', 'tr', 'td', 'th', 'thead', 'tbody', 'ul', 'ol', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'blockquote', 'pre', 'code'], ALLOWED_ATTR: ['href', 'src', 'alt', 'title', 'style', 'class', 'width', 'height', 'align', 'border', 'cellpadding', 'cellspacing'], ALLOW_DATA_ATTR: false, ADD_DATA_URI_TAGS: ['img'], ALLOWED_URI_REGEXP: /^(?:(?:https?|mailto|tel|cid):|data:image\/(?:png|gif|jpe?g|webp|bmp|x-icon|vnd\.microsoft\.icon|avif);base64,|[^a-z]|[a-z+.\-]+(?:[^a-z+.\-:]|$))/i, - FORBID_TAGS: ['script', 'iframe', 'object', 'embed', 'form', 'input', 'button'], + FORBID_TAGS: ['script', 'style', 'iframe', 'object', 'embed', 'form', 'input', 'button'], FORBID_ATTR: ['onerror', 'onload', 'onclick', 'onmouseover', 'onfocus', 'onblur'] }); } else { @@ -988,17 +936,12 @@ // 显示邮件列表（移动端） function showEmailList() { if (resolveEmailDetailSource() === 'temp') { - if (typeof setTempDetailFocus === 'function') { - setTempDetailFocus(false); - } currentEmailDetail = null; isTrustedMode = false; resetEmailDetailState({ source: 'temp' }); - hideEmailDetailContainer({ source: 'temp' }); return; } - setMailboxDetailFocus(false); syncEmailListVisibility(true); isListVisible = true; var t = document.getElementById('toggleListText'); diff --git a/tests/test_ai_fallback_trigger_condition.py b/tests/test_ai_fallback_trigger_condition.py index e9966619..dc42c4d7 100644 --- a/tests/test_ai_fallback_trigger_condition.py +++ b/tests/test_ai_fallback_trigger_condition.py @@ -193,6 +193,31 @@ def test_code_high_link_low_enforces_mutual_exclusion(self, mock_config, mock_ai self.assertEqual(result["link_confidence"], "low") +class AiFallbackCasePreservationTests(unittest.TestCase): + """验证码大小写：AI fallback 返回的验证码必须保持原大小写""" + + @patch("outlook_web.services.verification_extractor._call_verification_ai") + @patch("outlook_web.services.verification_extractor.get_verification_ai_runtime_config") + def test_ai_code_case_preserved(self, mock_config, mock_ai_call): + """AI 返回小写 code → 结果保持小写，不转大写""" + mock_config.return_value = _AI_CONFIG + mock_ai_call.return_value = { + "schema_version": "verification_ai_v1", + "verification_code": "cbdfdb", + "verification_link": "", + "confidence": "high", + "reason": "test", + } + extracted = _make_extracted("low", "low") + + result = extractor.enhance_verification_with_ai_fallback(email=_EMAIL_OBJ, extracted=extracted) + + mock_ai_call.assert_called_once() + self.assertTrue(result.get("ai_used")) + self.assertEqual(result["verification_code"], "cbdfdb") + self.assertEqual(result["formatted"], "cbdfdb") + + class AiFallbackEdgeCaseTests(unittest.TestCase): """边界情况：缺失 confidence 字段时默认为 low""" diff --git a/tests/test_domain_provider_map.py b/tests/test_domain_provider_map.py index 4b201c8f..7110510c 100644 --- a/tests/test_domain_provider_map.py +++ b/tests/test_domain_provider_map.py @@ -59,6 +59,20 @@ def test_infer_invalid_input(self): self.assertIsNone(infer_provider_from_email(None)) self.assertIsNone(infer_provider_from_email("no-at-sign")) + def test_extract_and_normalize_email_domain(self): + from outlook_web.services.providers import extract_email_domain, normalize_email_domain + + self.assertEqual(extract_email_domain("User@HotMail.COM"), "hotmail.com") + self.assertEqual(normalize_email_domain(" HotMail.COM "), "hotmail.com") + self.assertEqual(normalize_email_domain("user@live.com"), "live.com") + + def test_provider_supports_email_domain(self): + from outlook_web.services.providers import provider_supports_email_domain + + self.assertTrue(provider_supports_email_domain("outlook", "hotmail.com")) + self.assertTrue(provider_supports_email_domain("outlook", "HotMail.COM")) + self.assertFalse(provider_supports_email_domain("gmail", "hotmail.com")) + def test_known_provider_keys(self): from outlook_web.services.providers import KNOWN_PROVIDER_KEYS, MAIL_PROVIDERS diff --git a/tests/test_external_api.py b/tests/test_external_api.py index 75fd6d99..8d86c4ee 100644 --- a/tests/test_external_api.py +++ b/tests/test_external_api.py @@ -78,6 +78,52 @@ def _insert_outlook_account(self, email_addr: str | None = None) -> str: db.commit() return email_addr + def _insert_claimed_pool_account( + self, + *, + email_addr: str | None = None, + project_key: str = "register", + consumer_key: str = "legacy:settings.external_api_key", + caller_id: str = "pool-worker", + task_id: str = "pool-task", + claim_token: str = "clm_test_token", + claimed_at: str | None = None, + ) -> str: + email_addr = email_addr or f"{uuid.uuid4().hex}@extapi.test" + claimed_at = claimed_at or self._utc_iso() + claimed_by = f"{consumer_key}||{project_key}||{caller_id}||{task_id}" + with self.app.app_context(): + from outlook_web.db import get_db + + db = get_db() + db.execute( + """ + INSERT INTO accounts ( + email, email_domain, password, client_id, refresh_token, group_id, + status, account_type, provider, pool_status, claimed_by, claimed_at, + lease_expires_at, claim_token + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'claimed', ?, ?, ?, ?) + """, + ( + email_addr, + email_addr.rsplit("@", 1)[-1].lower(), + "pw", + "cid-test", + "rt-test", + 1, + "active", + "outlook", + "outlook", + claimed_by, + claimed_at, + self._utc_iso(minutes_delta=10), + claim_token, + ), + ) + db.commit() + return email_addr + def _insert_imap_account(self, email_addr: str | None = None) -> str: email_addr = email_addr or f"{uuid.uuid4().hex}@extapi.test" with self.app.app_context(): @@ -606,6 +652,65 @@ def test_external_verification_link_defaults_to_recent_10_minutes(self, mock_get self.assertEqual(resp.status_code, 404) self.assertEqual(resp.get_json().get("code"), "MAIL_NOT_FOUND") + @patch("outlook_web.services.graph.get_email_raw_graph") + @patch("outlook_web.services.graph.get_email_detail_graph") + @patch("outlook_web.services.graph.get_emails_graph") + def test_external_verification_code_supports_claim_token_baseline( + self, + mock_get_emails_graph, + mock_get_email_detail_graph, + mock_get_email_raw_graph, + ): + claimed_at = "2026-03-08T12:00:00Z" + claim_email = self._insert_claimed_pool_account( + email_addr=f"{uuid.uuid4().hex}@extapi.test", + claim_token="clm_claim_read_001", + claimed_at=claimed_at, + ) + self._set_external_api_key("abc123") + mock_get_emails_graph.return_value = { + "success": True, + "emails": [ + self._graph_email(message_id="old-msg", received_at="2026-03-08T11:59:00Z"), + self._graph_email(message_id="new-msg", received_at="2026-03-08T12:01:00Z"), + ], + } + mock_get_email_detail_graph.return_value = self._graph_detail( + message_id="new-msg", + body_text="Your code is 123456", + received_at="2026-03-08T12:01:00Z", + ) + mock_get_email_raw_graph.return_value = "RAW MIME CONTENT" + + client = self.app.test_client() + resp = client.get( + "/api/external/verification-code?claim_token=clm_claim_read_001", + headers=self._auth_headers(), + ) + + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertTrue(data.get("success")) + self.assertEqual(data.get("data", {}).get("email"), claim_email) + self.assertEqual(data.get("data", {}).get("matched_email_id"), "new-msg") + self.assertEqual(data.get("data", {}).get("claim_token"), "clm_claim_read_001") + + def test_external_verification_code_rejects_claim_email_mismatch(self): + self._insert_claimed_pool_account( + email_addr=f"{uuid.uuid4().hex}@extapi.test", + claim_token="clm_claim_mismatch_001", + ) + self._set_external_api_key("abc123") + client = self.app.test_client() + + resp = client.get( + "/api/external/verification-code?claim_token=clm_claim_mismatch_001&email=other@extapi.test", + headers=self._auth_headers(), + ) + + self.assertEqual(resp.status_code, 400) + self.assertEqual(resp.get_json().get("code"), "CLAIM_CONTEXT_MISMATCH") + @patch("outlook_web.services.external_api.time.sleep") @patch("outlook_web.services.external_api.time.time") @patch("outlook_web.services.external_api.get_latest_message_for_external") @@ -1112,6 +1217,34 @@ def test_wait_message_http_returns_400_for_missing_email(self): self.assertEqual(resp.status_code, 400) self.assertEqual(resp.get_json().get("code"), "INVALID_PARAM") + @patch("outlook_web.services.external_api.time.sleep") + @patch("outlook_web.services.external_api.time.time") + @patch("outlook_web.services.graph.get_emails_graph") + def test_wait_message_http_supports_claim_token_baseline(self, mock_get_emails_graph, mock_time, mock_sleep): + self._insert_claimed_pool_account( + email_addr=f"{uuid.uuid4().hex}@extapi.test", + claim_token="clm_wait_claim_001", + claimed_at="2026-03-08T12:00:00Z", + ) + self._set_external_api_key("abc123") + mock_time.side_effect = [2000000000, 2000000000, 2000000000] + mock_get_emails_graph.side_effect = [ + {"success": True, "emails": [self._graph_email(message_id="old-msg", received_at="2026-03-08T11:59:00Z")]}, + {"success": True, "emails": [self._graph_email(message_id="new-msg", received_at="2026-03-08T12:01:00Z")]}, + ] + + client = self.app.test_client() + resp = client.get( + "/api/external/wait-message?claim_token=clm_wait_claim_001&timeout_seconds=30&poll_interval=5", + headers=self._auth_headers(), + ) + + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertTrue(data.get("success")) + self.assertEqual(data.get("data", {}).get("id"), "new-msg") + self.assertEqual(data.get("data", {}).get("claim_token"), "clm_wait_claim_001") + @patch("outlook_web.services.external_api.wait_for_message") def test_wait_message_http_unexpected_error_logs_audit(self, mock_wait_for_message): """wait-message 未预期异常也应写 external_api 审计日志""" diff --git a/tests/test_external_pool.py b/tests/test_external_pool.py index 4b099ec8..0a0f249f 100644 --- a/tests/test_external_pool.py +++ b/tests/test_external_pool.py @@ -23,6 +23,9 @@ def setUp(self): db.execute( "DELETE FROM account_claim_logs WHERE account_id IN (SELECT id FROM accounts WHERE email LIKE '%@extpool.test')" ) + db.execute( + "DELETE FROM account_project_usage WHERE account_id IN (SELECT id FROM accounts WHERE email LIKE '%@extpool.test')" + ) db.execute("DELETE FROM accounts WHERE email LIKE '%@extpool.test'") db.commit() settings_repo.set_setting("external_api_key", "") @@ -164,7 +167,7 @@ def test_external_pool_claim_release_requires_api_key(self): claim_resp = client.post( "/api/external/pool/claim-random", headers=self._auth_headers(), - json={"caller_id": "ext-worker-01", "task_id": "release-no-key", "provider": "outlook"}, + json={"caller_id": "ext-worker-01", "project_key": "test_project", "task_id": "release-no-key", "provider": "outlook"}, ) self.assertEqual(claim_resp.status_code, 200) claim_data = claim_resp.get_json()["data"] @@ -176,6 +179,7 @@ def test_external_pool_claim_release_requires_api_key(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "release-no-key", + "project_key": "test_project", }, ) @@ -194,7 +198,7 @@ def test_external_pool_claim_complete_requires_api_key(self): claim_resp = client.post( "/api/external/pool/claim-random", headers=self._auth_headers(), - json={"caller_id": "ext-worker-01", "task_id": "complete-no-key", "provider": "outlook"}, + json={"caller_id": "ext-worker-01", "project_key": "test_project", "task_id": "complete-no-key", "provider": "outlook"}, ) self.assertEqual(claim_resp.status_code, 200) claim_data = claim_resp.get_json()["data"] @@ -206,6 +210,7 @@ def test_external_pool_claim_complete_requires_api_key(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "complete-no-key", + "project_key": "test_project", "result": "success", }, ) @@ -228,6 +233,7 @@ def test_external_pool_claim_random_success(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-001", + "project_key": "test_project", "provider": "outlook", }, ) @@ -256,6 +262,7 @@ def test_external_pool_post_does_not_require_csrf(self): json={ "caller_id": "csrf-free-worker", "task_id": "csrf-free-task", + "project_key": "test_project", "provider": "outlook", }, ) @@ -280,6 +287,7 @@ def fake_csrf_exempt(handler): set(wrapped_handlers), { "api_external_pool_claim_random", + "api_external_pool_claim_domain", "api_external_pool_claim_release", "api_external_pool_claim_complete", "api_external_pool_stats", @@ -301,6 +309,7 @@ def test_external_pool_claim_release_caller_mismatch(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-002", + "project_key": "test_project", "provider": "outlook", }, ) @@ -315,6 +324,7 @@ def test_external_pool_claim_release_caller_mismatch(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-02", "task_id": "task-ext-002", + "project_key": "test_project", }, ) @@ -338,6 +348,7 @@ def test_external_pool_claim_complete_success(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-complete", + "project_key": "test_project", "provider": "outlook", }, ) @@ -352,6 +363,7 @@ def test_external_pool_claim_complete_success(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "task-ext-complete", + "project_key": "test_project", "result": "success", "detail": "done", }, @@ -361,7 +373,7 @@ def test_external_pool_claim_complete_success(self): data = complete_resp.get_json() self.assertTrue(data.get("success")) self.assertEqual(data.get("code"), "OK") - self.assertEqual(data.get("data", {}).get("pool_status"), "used") + self.assertEqual(data.get("data", {}).get("pool_status"), "available") def test_external_pool_stats_success(self): client = self.app.test_client() @@ -428,7 +440,7 @@ def test_external_pool_claim_random_disabled_in_public_mode(self): resp = client.post( "/api/external/pool/claim-random", headers=self._auth_headers(), - json={"caller_id": "ext-worker-01", "task_id": "task-ext-disabled"}, + json={"caller_id": "ext-worker-01", "project_key": "test_project", "task_id": "task-ext-disabled"}, ) self.assertEqual(resp.status_code, 403) @@ -450,6 +462,7 @@ def test_external_pool_claim_release_disabled_in_public_mode(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-rel", + "project_key": "test_project", "provider": "outlook", }, ) @@ -468,6 +481,7 @@ def test_external_pool_claim_release_disabled_in_public_mode(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "task-ext-rel", + "project_key": "test_project", }, ) @@ -490,6 +504,7 @@ def test_external_pool_claim_complete_disabled_in_public_mode(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-comp", + "project_key": "test_project", "provider": "outlook", }, ) @@ -508,6 +523,7 @@ def test_external_pool_claim_complete_disabled_in_public_mode(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "task-ext-comp", + "project_key": "test_project", "result": "success", }, ) @@ -550,6 +566,7 @@ def test_external_pool_claim_random_requires_pool_access_for_multi_key(self): json={ "caller_id": "ext-worker-01", "task_id": "task-ext-no-access", + "project_key": "test_project", "provider": "outlook", }, ) @@ -571,7 +588,7 @@ def test_external_pool_claim_release_requires_pool_access_for_multi_key(self): claim_resp = client.post( "/api/external/pool/claim-random", headers=self._auth_headers("multi-pool-release-allow"), - json={"caller_id": "ext-worker-01", "task_id": "release-deny", "provider": "outlook"}, + json={"caller_id": "ext-worker-01", "project_key": "test_project", "task_id": "release-deny", "provider": "outlook"}, ) self.assertEqual(claim_resp.status_code, 200) claim_data = claim_resp.get_json()["data"] @@ -584,6 +601,7 @@ def test_external_pool_claim_release_requires_pool_access_for_multi_key(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "release-deny", + "project_key": "test_project", }, ) @@ -603,7 +621,7 @@ def test_external_pool_claim_complete_requires_pool_access_for_multi_key(self): claim_resp = client.post( "/api/external/pool/claim-random", headers=self._auth_headers("multi-pool-complete-allow"), - json={"caller_id": "ext-worker-01", "task_id": "complete-deny", "provider": "outlook"}, + json={"caller_id": "ext-worker-01", "project_key": "test_project", "task_id": "complete-deny", "provider": "outlook"}, ) self.assertEqual(claim_resp.status_code, 200) claim_data = claim_resp.get_json()["data"] @@ -616,6 +634,7 @@ def test_external_pool_claim_complete_requires_pool_access_for_multi_key(self): "claim_token": claim_data["claim_token"], "caller_id": "ext-worker-01", "task_id": "complete-deny", + "project_key": "test_project", "result": "success", }, ) diff --git a/tests/test_imap_connection_reuse.py b/tests/test_imap_connection_reuse.py index c4845e8d..adc726a7 100644 --- a/tests/test_imap_connection_reuse.py +++ b/tests/test_imap_connection_reuse.py @@ -194,6 +194,32 @@ def test_imap_connection_created_only_once(self, mock_token, mock_imap_cls): self.assertEqual(mock_imap_cls.call_count, 1) self.assertEqual(mock_conn.authenticate.call_count, 1) + @patch("outlook_web.services.imap.imaplib.IMAP4_SSL") + @patch("outlook_web.services.imap.get_access_token_imap_result") + def test_batch_fetch_response_is_reordered_to_requested_latest_first(self, mock_token, mock_imap_cls): + from outlook_web.services.imap import fetch_and_detail_imap_with_server + + mock_token.return_value = self._mock_token_result(True) + raw_old = _build_rfc822_bytes("Old code", "Your code is 990595") + raw_middle = _build_rfc822_bytes("Middle code", "Your code is 118658") + raw_new = _build_rfc822_bytes("New code", "Your code is 701280") + mock_conn = self._setup_imap_mock(mock_imap_cls, search_ids=[b"1", b"2", b"3", b"4", b"5"]) + mock_conn.fetch.return_value = ( + "OK", + [ + ((b"3 (RFC822)", raw_old), b")"), + ((b"4 (RFC822)", raw_middle), b")"), + ((b"5 (RFC822)", raw_new), b")"), + ], + ) + + result = fetch_and_detail_imap_with_server("user@test.com", "cid", "rt", folder="inbox", top=3) + + self.assertTrue(result.get("success")) + self.assertEqual([item["id"] for item in result.get("emails", [])], ["5", "4", "3"]) + self.assertEqual(result.get("detail", {}).get("id"), "5") + self.assertIn("701280", result.get("detail", {}).get("body", "")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_pool_flow_suite.py b/tests/test_pool_flow_suite.py index 17b70779..a787bef1 100644 --- a/tests/test_pool_flow_suite.py +++ b/tests/test_pool_flow_suite.py @@ -51,6 +51,21 @@ def setUp(self): def _auth_headers(): return {"X-API-Key": "abc123"} + def _claim(self, *, task_id: str, project_key: str = "register", provider: str = "outlook", email_domain: str | None = None): + payload = { + "caller_id": "suite_bot", + "task_id": task_id, + "project_key": project_key, + "provider": provider, + } + if email_domain is not None: + payload["email_domain"] = email_domain + return self.client.post( + "/api/external/pool/claim-random", + headers=self._auth_headers(), + json=payload, + ) + def _make_pool_account( self, *, @@ -81,14 +96,10 @@ def _make_pool_account( finally: conn.close() - def test_claim_complete_success_without_project_key_still_changes_status_to_used(self): + def test_claim_complete_requires_project_key(self): self._make_pool_account() - claim_resp = self.client.post( - "/api/external/pool/claim-random", - headers=self._auth_headers(), - json={"caller_id": "suite_bot", "task_id": "success_flow"}, - ) + claim_resp = self._claim(task_id="success_flow", project_key="register") self.assertEqual(claim_resp.status_code, 200) claim_data = json.loads(claim_resp.data) self.assertTrue(claim_data["success"]) @@ -105,23 +116,10 @@ def test_claim_complete_success_without_project_key_still_changes_status_to_used "detail": "manual suite success", }, ) - self.assertEqual(complete_resp.status_code, 200) + self.assertEqual(complete_resp.status_code, 400) complete_data = json.loads(complete_resp.data) - self.assertTrue(complete_data["success"]) - # 未传 project_key 时，旧行为仍为 success → used - self.assertEqual(complete_data["data"]["pool_status"], "used") - - conn = self.create_conn() - try: - row = conn.execute( - "SELECT pool_status, success_count, fail_count FROM accounts WHERE id = ?", - (claim_data["data"]["account_id"],), - ).fetchone() - self.assertEqual(row["pool_status"], "used") - self.assertEqual(row["success_count"], 1) - self.assertEqual(row["fail_count"], 0) - finally: - conn.close() + self.assertFalse(complete_data["success"]) + self.assertEqual(complete_data["code"], "PROJECT_KEY_EMPTY") def test_claim_complete_success_with_project_key_on_long_lived_account_returns_available(self): account = self._make_pool_account(email_domain=f"reuse_{uuid.uuid4().hex[:8]}.test") @@ -148,6 +146,7 @@ def test_claim_complete_success_with_project_key_on_long_lived_account_returns_a "claim_token": claim_data["data"]["claim_token"], "caller_id": "reuse_bot", "task_id": "reuse_success_1", + "project_key": "project_alpha", "result": "success", }, ) @@ -175,11 +174,7 @@ def test_claim_complete_success_with_project_key_on_long_lived_account_returns_a def test_claim_complete_failure_changes_status_to_cooldown(self): self._make_pool_account() - claim_resp = self.client.post( - "/api/external/pool/claim-random", - headers=self._auth_headers(), - json={"caller_id": "suite_bot", "task_id": "cooldown_flow"}, - ) + claim_resp = self._claim(task_id="cooldown_flow", project_key="register") self.assertEqual(claim_resp.status_code, 200) claim_data = json.loads(claim_resp.data) self.assertTrue(claim_data["success"]) @@ -192,6 +187,7 @@ def test_claim_complete_failure_changes_status_to_cooldown(self): "claim_token": claim_data["data"]["claim_token"], "caller_id": "suite_bot", "task_id": "cooldown_flow", + "project_key": "register", "result": "verification_timeout", "detail": "manual suite timeout", }, @@ -230,7 +226,7 @@ def test_multiple_consecutive_claims_do_not_repeat_accounts(self): headers=self._auth_headers(), json={ "caller_id": "suite_bot", - "task_id": f"batch_claim_{idx}", + "project_key": "test_project", "task_id": f"batch_claim_{idx}", "email_domain": email_domain, # 使用 email_domain 过滤 }, ) @@ -257,7 +253,7 @@ def test_multiple_consecutive_claims_do_not_repeat_accounts(self): "account_id": account_id, "claim_token": claim_token, "caller_id": "suite_bot", - "task_id": task_id, + "project_key": "test_project", "task_id": task_id, "reason": "suite cleanup", }, ) @@ -286,7 +282,7 @@ def test_claim_response_includes_email_domain_and_claimed_at(self): resp = self.client.post( "/api/external/pool/claim-random", headers=self._auth_headers(), - json={"caller_id": "domain_bot", "task_id": "domain_check"}, + json={"caller_id": "domain_bot", "project_key": "test_project", "task_id": "domain_check"}, ) self.assertEqual(resp.status_code, 200) data = json.loads(resp.data) @@ -295,6 +291,40 @@ def test_claim_response_includes_email_domain_and_claimed_at(self): self.assertIn("claimed_at", data["data"]) self.assertIsNotNone(data["data"]["claimed_at"]) + def test_claim_domain_requires_email_domain(self): + resp = self.client.post( + "/api/external/pool/claim-domain", + headers=self._auth_headers(), + json={"caller_id": "domain_bot", "project_key": "test_project", "task_id": "domain_required"}, + ) + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertFalse(data["success"]) + self.assertEqual(data["code"], "EMAIL_DOMAIN_REQUIRED") + + def test_claim_domain_filters_to_requested_domain(self): + requested_domain = f"requested_{uuid.uuid4().hex[:8]}.test" + other_domain = f"other_{uuid.uuid4().hex[:8]}.test" + requested = self._make_pool_account(email_domain=requested_domain) + other = self._make_pool_account(email_domain=other_domain) + + resp = self.client.post( + "/api/external/pool/claim-domain", + headers=self._auth_headers(), + json={ + "caller_id": "domain_bot", + "task_id": "domain_claim", + "project_key": "test_project", + "email_domain": requested_domain.upper(), + }, + ) + self.assertEqual(resp.status_code, 200) + data = json.loads(resp.data) + self.assertTrue(data["success"]) + self.assertEqual(data["data"]["account_id"], requested["id"]) + self.assertEqual(data["data"]["email_domain"], requested_domain) + self.assertNotEqual(data["data"]["account_id"], other["id"]) + def test_claim_with_project_key_prevents_same_project_reuse_without_manual_status_reset(self): """同 caller_id + project_key 下，新语义应原生阻止再次领取，无需手工改状态。""" # 使用唯一的 email_domain 隔离测试数据 @@ -327,6 +357,7 @@ def test_claim_with_project_key_prevents_same_project_reuse_without_manual_statu "claim_token": data1["data"]["claim_token"], "caller_id": "proj_bot", "task_id": "proj_task_1", + "project_key": "project_alpha", "result": "success", }, ) @@ -383,6 +414,7 @@ def test_claim_with_different_project_key_allows_immediate_reuse_after_success(s "claim_token": data1["data"]["claim_token"], "caller_id": "proj_bot", "task_id": "pb_task_1", + "project_key": "project_beta", "result": "success", }, ) @@ -416,6 +448,7 @@ def test_claim_with_different_project_key_allows_immediate_reuse_after_success(s "claim_token": data2["data"]["claim_token"], "caller_id": "proj_bot", "task_id": "pg_task_1", + "project_key": "test_project", }, ) @@ -445,6 +478,7 @@ def test_claim_complete_success_updates_stats_to_available_not_used(self): "claim_token": claim_data["data"]["claim_token"], "caller_id": "stats_bot", "task_id": "stats_task_1", + "project_key": "project_stats", "result": "success", }, ) @@ -491,6 +525,7 @@ def test_same_project_verification_timeout_does_not_block_retry_after_recovery(s "claim_token": claim_data["data"]["claim_token"], "caller_id": "timeout_bot", "task_id": "timeout_task_1", + "project_key": "project_timeout", "result": "verification_timeout", }, ) @@ -548,6 +583,7 @@ def test_same_project_manual_release_does_not_block_retry(self): "claim_token": claim_data["data"]["claim_token"], "caller_id": "release_bot", "task_id": "release_task_1", + "project_key": "project_release", }, ) self.assertEqual(release_resp.status_code, 200) @@ -601,6 +637,7 @@ def test_cloudflare_temp_mail_success_with_project_key_still_returns_old_status_ "claim_token": claim_data["data"]["claim_token"], "caller_id": "cf_bot", "task_id": "cf_task_1", + "project_key": "project_cf", "result": "success", }, ) diff --git a/tests/test_verification_extract_log.py b/tests/test_verification_extract_log.py index 1641f3a0..97d0b237 100644 --- a/tests/test_verification_extract_log.py +++ b/tests/test_verification_extract_log.py @@ -327,3 +327,74 @@ def test_log_channel_is_ai_fallback_when_ai_is_used(self): self.assertIn("_log_channel", result) self.assertEqual(result["_log_channel"], "ai_fallback") + + def test_imap_detail_mismatch_refetches_latest_message_detail(self): + """IMAP 连接复用返回的 detail 与最新邮件不一致时，应按 latest.id 重新取详情。""" + with self.app.app_context(): + from outlook_web.services import verification_channel_routing as vcr + + fake_account = { + "id": 3, + "email": "imap@outlook.com", + "account_type": "outlook", + "provider": "outlook", + "group_id": None, + "preferred_verification_channel": "imap_new", + "client_id": "cid", + "refresh_token": "rt", + } + + fake_channel_result = { + "success": True, + "emails": [ + { + "id": "3", + "subject": "Old code", + "from": "OpenAI", + "date": "Tue, 19 May 2026 10:01:15 +0000", + }, + { + "id": "5", + "subject": "New code", + "from": "OpenAI", + "date": "Tue, 19 May 2026 10:38:27 +0000", + }, + ], + "detail": { + "id": "3", + "subject": "Old code", + "from": "OpenAI", + "date": "Tue, 19 May 2026 10:01:15 +0000", + "body": "Your code is 990595", + }, + } + latest_detail = { + "id": "5", + "subject": "New code", + "from": "OpenAI", + "date": "Tue, 19 May 2026 10:38:27 +0000", + "body": "Your code is 701280", + } + + with ( + patch.object(vcr, "build_verification_channel_plan", return_value=["imap_new"]), + patch.object(vcr, "fetch_emails_and_detail_for_channel", return_value=fake_channel_result), + patch.object(vcr, "fetch_email_detail_for_channel", return_value=latest_detail) as mock_fetch_detail, + patch( + "outlook_web.services.graph.get_access_token_graph_result", + return_value={"success": False}, + ), + patch("outlook_web.repositories.accounts.update_preferred_verification_channel"), + ): + result = vcr.extract_verification_for_outlook( + account=fake_account, + resolved_policy={"code_regex": r"(?