diff --git a/src/core/auto_registration.py b/src/core/auto_registration.py
index 8535e9cd..8890a6e5 100644
--- a/src/core/auto_registration.py
+++ b/src/core/auto_registration.py
@@ -163,6 +163,7 @@ def start(self) -> None:
async def stop(self) -> None:
if not self._task:
return
+ self._wake_event.set()
self._task.cancel()
try:
await self._task
diff --git a/src/core/register.py b/src/core/register.py
index 08450164..24c0d7cb 100644
--- a/src/core/register.py
+++ b/src/core/register.py
@@ -3,6 +3,7 @@
从 main.py 中提取并重构的注册流程
"""
+import asyncio
import re
import json
import time
@@ -39,6 +40,10 @@
logger = logging.getLogger(__name__)
+class RegistrationCancelledError(asyncio.CancelledError):
+ """注册任务收到取消请求时抛出的协作式取消异常。"""
+
+
@dataclass
class RegistrationResult:
"""注册结果"""
@@ -98,7 +103,8 @@ def __init__(
email_service: BaseEmailService,
proxy_url: Optional[str] = None,
callback_logger: Optional[Callable[[str], None]] = None,
- task_uuid: Optional[str] = None
+ task_uuid: Optional[str] = None,
+ check_cancelled: Optional[Callable[[], bool]] = None,
):
"""
初始化注册引擎
@@ -108,11 +114,13 @@ def __init__(
proxy_url: 代理 URL
callback_logger: 日志回调函数
task_uuid: 任务 UUID（用于数据库记录）
+ check_cancelled: 取消检查回调（返回 True 表示任务应尽快停止）
"""
self.email_service = email_service
self.proxy_url = proxy_url
self.callback_logger = callback_logger or (lambda msg: logger.info(msg))
self.task_uuid = task_uuid
+ self._check_cancelled = check_cancelled or (lambda: False)
# 创建 HTTP 客户端
self.http_client = OpenAIHTTPClient(proxy_url=proxy_url)
@@ -155,6 +163,24 @@ def __init__(
self._last_otp_validation_status_code: Optional[int] = None
self._last_otp_validation_outcome: str = "" # success/http_non_200/network_timeout/network_error
+ def _is_cancel_requested(self) -> bool:
+ try:
+ return bool(self._check_cancelled())
+ except Exception:
+ return False
+
+ def _raise_if_cancelled(self, reason: str = "任务已取消") -> None:
+ if self._is_cancel_requested():
+ raise RegistrationCancelledError(reason)
+
+ def _sleep_interruptible(self, seconds: float) -> None:
+ remaining = max(0.0, float(seconds or 0.0))
+ while remaining > 0:
+ self._raise_if_cancelled("任务在等待重试阶段被取消")
+ chunk = min(0.2, remaining)
+ time.sleep(chunk)
+ remaining -= chunk
+
def _log(self, message: str, level: str = "info"):
"""记录日志"""
timestamp = datetime.now().strftime("%H:%M:%S")
@@ -360,6 +386,7 @@ def _generate_password(self, length: int = DEFAULT_PASSWORD_LENGTH) -> str:
def _check_ip_location(self) -> Tuple[bool, Optional[str]]:
"""检查 IP 地理位置"""
+ self._raise_if_cancelled("任务已取消，跳过 IP 地理位置检查")
try:
return self.http_client.check_ip_location()
except Exception as e:
@@ -368,6 +395,7 @@ def _check_ip_location(self) -> Tuple[bool, Optional[str]]:
def _create_email(self) -> bool:
"""创建邮箱"""
+ self._raise_if_cancelled("任务已取消，跳过邮箱创建")
try:
self._log(f"正在创建 {self.email_service.service_type.value} 邮箱，先给新账号整个收件箱...")
self.email_info = self.email_service.create_email()
@@ -396,6 +424,7 @@ def _create_email(self) -> bool:
def _start_oauth(self) -> bool:
"""开始 OAuth 流程"""
+ self._raise_if_cancelled("任务已取消，跳过 OAuth 初始化")
try:
self._log("开始 OAuth 授权流程，去门口刷个脸...")
self.oauth_start = self.oauth_manager.start_oauth()
@@ -407,6 +436,7 @@ def _start_oauth(self) -> bool:
def _init_session(self) -> bool:
"""初始化会话"""
+ self._raise_if_cancelled("任务已取消，跳过会话初始化")
try:
self.session = self.http_client.session
return True
@@ -416,11 +446,13 @@ def _init_session(self) -> bool:
def _get_device_id(self) -> Optional[str]:
"""获取 Device ID"""
+ self._raise_if_cancelled("任务已取消，停止获取 Device ID")
if not self.oauth_start:
return None
max_attempts = 3
for attempt in range(1, max_attempts + 1):
+ self._raise_if_cancelled("任务已取消，停止获取 Device ID")
try:
if not self.session:
self.session = self.http_client.session
@@ -460,7 +492,7 @@ def _get_device_id(self) -> Optional[str]:
)
if attempt < max_attempts:
- time.sleep(attempt)
+ self._sleep_interruptible(attempt)
self.http_client.close()
self.session = self.http_client.session
@@ -476,6 +508,7 @@ def _get_device_id(self) -> Optional[str]:
def _check_sentinel(self, did: str) -> Optional[str]:
"""检查 Sentinel 拦截"""
+ self._raise_if_cancelled("任务已取消，停止 Sentinel 检查")
try:
sen_token = self.http_client.check_sentinel(did)
if sen_token:
@@ -508,6 +541,7 @@ def _submit_auth_start(
current_did = str(did or "").strip()
current_sen_token = str(sen_token or "").strip() if sen_token else None
for attempt in range(1, max_attempts + 1):
+ self._raise_if_cancelled("任务已取消，停止提交授权入口")
try:
request_body = json.dumps({
"username": {
@@ -547,7 +581,7 @@ def _submit_auth_start(
f"{log_label}命中限流 429（第 {attempt}/{max_attempts} 次），{wait_seconds}s 后自动重试...",
"warning",
)
- time.sleep(wait_seconds)
+ self._sleep_interruptible(wait_seconds)
continue
# 部分网络/会话边界情况下会返回 409，做自愈重试而非直接失败。
@@ -571,7 +605,7 @@ def _submit_auth_start(
self.session.get(str(self.oauth_start.auth_url), timeout=12)
except Exception:
pass
- time.sleep(wait_seconds)
+ self._sleep_interruptible(wait_seconds)
continue
if response.status_code != 200:
@@ -614,7 +648,7 @@ def _submit_auth_start(
f"{log_label}异常（第 {attempt}/{max_attempts} 次）: {e}，准备重试...",
"warning",
)
- time.sleep(2 * attempt)
+ self._sleep_interruptible(2 * attempt)
continue
self._log(f"{log_label}失败: {e}", "error")
return SignupFormResult(success=False, error_message=str(e))
@@ -651,6 +685,7 @@ def _submit_login_start(self, did: str, sen_token: Optional[str]) -> SignupFormR
def _submit_login_password(self) -> SignupFormResult:
"""提交登录密码，进入邮箱验证码页面。"""
+ self._raise_if_cancelled("任务已取消，停止提交登录密码")
max_attempts = 3
password_text = str(self.password or "").strip()
if not password_text and self.email:
@@ -672,6 +707,7 @@ def _submit_login_password(self) -> SignupFormResult:
)
for attempt in range(1, max_attempts + 1):
+ self._raise_if_cancelled("任务已取消，停止登录密码重试")
try:
response = self.session.post(
OPENAI_API_ENDPOINTS["password_verify"],
@@ -691,7 +727,7 @@ def _submit_login_password(self) -> SignupFormResult:
f"提交登录密码命中限流 429（第 {attempt}/{max_attempts} 次），{wait_seconds}s 后自动重试...",
"warning",
)
- time.sleep(wait_seconds)
+ self._sleep_interruptible(wait_seconds)
continue
if response.status_code == 401 and attempt < max_attempts:
@@ -703,7 +739,7 @@ def _submit_login_password(self) -> SignupFormResult:
f"疑似密码尚未生效或历史账号密码不一致，{wait_seconds}s 后自动重试...",
"warning",
)
- time.sleep(wait_seconds)
+ self._sleep_interruptible(wait_seconds)
continue
if response.status_code != 200:
@@ -734,7 +770,7 @@ def _submit_login_password(self) -> SignupFormResult:
f"提交登录密码异常（第 {attempt}/{max_attempts} 次）: {e}，准备重试...",
"warning",
)
- time.sleep(2 * attempt)
+ self._sleep_interruptible(2 * attempt)
continue
self._log(f"提交登录密码失败: {e}", "error")
return SignupFormResult(success=False, error_message=str(e))
@@ -751,6 +787,7 @@ def _reset_auth_flow(self) -> None:
def _prepare_authorize_flow(self, label: str) -> Tuple[Optional[str], Optional[str]]:
"""初始化当前阶段的授权流程，返回 device id 和 sentinel token。"""
+ self._raise_if_cancelled(f"任务已取消，停止执行 {label}")
self._log(f"{label}: 先把会话热热身...")
if not self._init_session():
return None, None
@@ -2074,6 +2111,7 @@ def _register_password_with_retry(
sen_token: Optional[str] = None,
) -> Tuple[bool, Optional[str]]:
"""Retry password registration when OpenAI returns a generic recoverable 400."""
+ self._raise_if_cancelled("任务已取消，停止密码注册重试")
max_attempts = 3
retryable_markers = (
"failed to create account",
@@ -2083,6 +2121,7 @@ def _register_password_with_retry(
)
for attempt in range(1, max_attempts + 1):
+ self._raise_if_cancelled("任务已取消，停止密码注册重试")
success, password = self._register_password(did, sen_token)
if success:
return True, password
@@ -2097,7 +2136,7 @@ def _register_password_with_retry(
f"密码注册命中可重试 400，准备重新生成密码后重试 ({attempt}/{max_attempts})...",
"warning",
)
- time.sleep(min(2 * attempt, 4))
+ self._sleep_interruptible(min(2 * attempt, 4))
return False, None
@@ -2124,6 +2163,7 @@ def _mark_email_as_registered(self):
def _send_verification_code(self, referer: Optional[str] = None) -> bool:
"""发送验证码"""
+ self._raise_if_cancelled("任务已取消，停止发送验证码")
try:
# 记录发送时间戳
self._otp_sent_at = time.time()
@@ -2146,6 +2186,7 @@ def _send_verification_code(self, referer: Optional[str] = None) -> bool:
def _get_verification_code(self, timeout: Optional[int] = None) -> Optional[str]:
"""获取验证码"""
+ self._raise_if_cancelled("任务已取消，停止拉取验证码")
try:
mailbox_email = str(self.inbox_email or self.email or "").strip()
self._log(f"正在等待邮箱 {mailbox_email} 的验证码...")
@@ -2173,6 +2214,7 @@ def _get_verification_code(self, timeout: Optional[int] = None) -> Optional[str]
def _validate_verification_code(self, code: str) -> bool:
"""验证验证码"""
+ self._raise_if_cancelled("任务已取消，停止校验验证码")
try:
self._last_otp_validation_code = str(code or "").strip()
self._last_otp_validation_status_code = None
@@ -2267,11 +2309,13 @@ def _verify_email_otp_with_retry(
用于规避邮箱里历史验证码导致的 400（第一次取到旧码，第二次取新码）。
"""
# 每轮验证码阶段开始前，清理上轮 OTP 校验缓存，避免 continue_url/workspace 被旧阶段污染。
+ self._raise_if_cancelled(f"任务已取消，停止{stage_label}校验")
self._last_validate_otp_continue_url = None
self._last_validate_otp_workspace_id = None
if attempted_codes is None:
attempted_codes = set()
for attempt in range(1, max_attempts + 1):
+ self._raise_if_cancelled(f"任务已取消，停止{stage_label}重试")
code = (
self._get_verification_code(timeout=fetch_timeout)
if fetch_timeout
@@ -2283,7 +2327,7 @@ def _verify_email_otp_with_retry(
f"{stage_label}第 {attempt}/{max_attempts} 次未取到验证码，稍后重试...",
"warning",
)
- time.sleep(2)
+ self._sleep_interruptible(2)
continue
return False
@@ -2301,7 +2345,7 @@ def _verify_email_otp_with_retry(
if self._validate_verification_code(code):
return True
if attempt < max_attempts:
- time.sleep(2)
+ self._sleep_interruptible(2)
continue
return False
@@ -2310,7 +2354,7 @@ def _verify_email_otp_with_retry(
f"{stage_label}第 {attempt}/{max_attempts} 次命中重复验证码 {code}，等待新邮件...",
"warning",
)
- time.sleep(2)
+ self._sleep_interruptible(2)
continue
return False
@@ -2324,12 +2368,13 @@ def _verify_email_otp_with_retry(
f"{stage_label}第 {attempt}/{max_attempts} 次校验未通过，疑似旧验证码，自动重试下一封...",
"warning",
)
- time.sleep(2)
+ self._sleep_interruptible(2)
return False
def _create_user_account(self) -> bool:
"""创建用户账户"""
+ self._raise_if_cancelled("任务已取消，停止创建用户账户")
try:
user_info = generate_random_user_info()
self._log(f"生成用户信息: {user_info['name']}, 生日: {user_info['birthdate']}")
@@ -2392,6 +2437,7 @@ def _create_user_account(self) -> bool:
def _get_workspace_id(self) -> Optional[str]:
"""获取 Workspace ID"""
+ self._raise_if_cancelled("任务已取消，停止获取 Workspace ID")
try:
def _extract_workspace_id(payload: Any) -> str:
if not isinstance(payload, dict):
@@ -2487,6 +2533,7 @@ def _extract_workspace_id(payload: Any) -> str:
def _select_workspace(self, workspace_id: str) -> Optional[str]:
"""选择 Workspace"""
+ self._raise_if_cancelled("任务已取消，停止选择 Workspace")
try:
select_body = f'{{"workspace_id":"{workspace_id}"}}'
@@ -2551,6 +2598,7 @@ def _select_workspace(self, workspace_id: str) -> Optional[str]:
def _follow_redirects(self, start_url: str) -> Tuple[Optional[str], str]:
"""手动跟随重定向链，返回 (callback_url, final_url)。"""
+ self._raise_if_cancelled("任务已取消，停止跟随重定向")
try:
def _is_oauth_callback(url: str) -> bool:
try:
@@ -2571,6 +2619,7 @@ def _is_oauth_callback(url: str) -> bool:
max_redirects = 12
for i in range(max_redirects):
+ self._raise_if_cancelled("任务已取消，停止跟随重定向")
self._log(f"重定向 {i+1}/{max_redirects}: {current_url[:100]}...")
if _is_oauth_callback(current_url) and not callback_url:
callback_url = current_url
@@ -2639,6 +2688,7 @@ def _is_oauth_callback(url: str) -> bool:
def _handle_oauth_callback(self, callback_url: str) -> Optional[Dict[str, Any]]:
"""处理 OAuth 回调"""
+ self._raise_if_cancelled("任务已取消，停止处理 OAuth 回调")
try:
if not self.oauth_start:
self._log("OAuth 流程未初始化", "error")
@@ -2670,6 +2720,7 @@ def _run_primary_registration(self) -> RegistrationResult:
Returns:
RegistrationResult: 注册结果
"""
+ self._raise_if_cancelled("任务已取消，停止注册流程")
result = RegistrationResult(success=False, logs=self.logs)
try:
@@ -2697,6 +2748,7 @@ def _run_primary_registration(self) -> RegistrationResult:
# 1. 检查 IP 地理位置
self._log("1. 先看看这条网络从哪儿来，别一开局就站错片场...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
ip_ok, location = self._check_ip_location()
if not ip_ok:
result.error_message = f"IP 地理位置不支持: {location}"
@@ -2707,6 +2759,7 @@ def _run_primary_registration(self) -> RegistrationResult:
# 2. 创建邮箱
self._log("2. 开个新邮箱，准备收信...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._create_email():
result.error_message = "创建邮箱失败"
return result
@@ -2714,6 +2767,7 @@ def _run_primary_registration(self) -> RegistrationResult:
result.email = self.email
# 3. 准备首轮授权流程
+ self._raise_if_cancelled("任务已取消，停止注册流程")
did, sen_token = self._prepare_authorize_flow("首次授权")
if not did:
result.error_message = "获取 Device ID 失败"
@@ -2725,6 +2779,7 @@ def _run_primary_registration(self) -> RegistrationResult:
# 4. 提交注册入口邮箱
self._log("4. 递上邮箱，看看 OpenAI 这球怎么接...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
signup_result = self._submit_signup_form(did, sen_token)
if not signup_result.success:
result.error_message = f"提交注册表单失败: {signup_result.error_message}"
@@ -2734,28 +2789,33 @@ def _run_primary_registration(self) -> RegistrationResult:
self._log("检测到这是老朋友账号，直接切去登录拿 token，不走弯路")
else:
self._log("5. 设置密码，别让小偷偷笑...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
password_ok, _ = self._register_password_with_retry(did, sen_token)
if not password_ok:
result.error_message = self._last_register_password_error or "注册密码失败"
return result
self._log("6. 催一下注册验证码出门，邮差该冲刺了...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._send_verification_code():
result.error_message = "发送验证码失败"
return result
self._log("7. 等验证码飞来，邮箱请注意查收...")
self._log("8. 对一下验证码，看看是不是本人...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._verify_email_otp_with_retry(stage_label="注册验证码", max_attempts=3):
result.error_message = "验证验证码失败"
return result
self._log("9. 给账号办个正式户口，名字写档案里...")
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._create_user_account():
result.error_message = "创建用户账户失败"
return result
if effective_entry_flow in {"native", "outlook"}:
+ self._raise_if_cancelled("任务已取消，停止注册流程")
login_ready, login_error = self._restart_login_flow()
if not login_ready:
result.error_message = login_error
@@ -2766,13 +2826,16 @@ def _run_primary_registration(self) -> RegistrationResult:
self._log("注册入口链路: ABCard（新账号不重登，直接抓取会话）")
if effective_entry_flow == "native":
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._complete_token_exchange_native_backup(result):
return result
elif effective_entry_flow == "outlook":
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._complete_token_exchange_outlook(result):
return result
else:
use_abcard_entry = (effective_entry_flow == "abcard") and (not self._is_existing_account)
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if not self._complete_token_exchange(result, require_login_otp=not use_abcard_entry):
return result
@@ -2886,6 +2949,7 @@ def _build_anyauto_fallback_result(
def _run_anyauto_fallback(self, primary_error: str = "") -> RegistrationResult:
"""Run the PR60 AnyAuto V2 engine as a controlled fallback."""
+ self._raise_if_cancelled("任务已取消，停止回退注册流程")
settings = get_settings()
max_retries = int(getattr(settings, "registration_max_retries", 3) or 3)
browser_mode = str(
@@ -2957,13 +3021,16 @@ def _should_try_anyauto_fallback(self, result: RegistrationResult) -> bool:
def run(self) -> RegistrationResult:
"""Run the current primary flow first, then selectively fall back to PR60 AnyAuto V2."""
+ self._raise_if_cancelled("任务已取消，停止注册流程")
primary_result = self._run_primary_registration()
+ self._raise_if_cancelled("任务已取消，停止注册流程")
if primary_result.success:
return primary_result
if not self._should_try_anyauto_fallback(primary_result):
return primary_result
+ self._raise_if_cancelled("任务已取消，跳过回退注册流程")
primary_error = str(primary_result.error_message or "").strip()
self._log("主注册链路未成功，开始尝试 PR60 anyauto V2 回退流程...", "warning")
fallback_result = self._run_anyauto_fallback(primary_error=primary_error)
diff --git a/src/services/cloudmail.py b/src/services/cloudmail.py
index 86e93530..142a6d67 100644
--- a/src/services/cloudmail.py
+++ b/src/services/cloudmail.py
@@ -1,37 +1,529 @@
"""
-CloudMail ??????
+Cloud Mail 邮箱服务实现
+基于 Cloudflare Workers 的邮箱服务 (https://doc.skymail.ink)
"""
-from typing import Any, Dict, List, Optional
+import re
+import sys
+import time
+import logging
+import random
+import string
+import requests
+from typing import Optional, Dict, Any, List
+from datetime import datetime
-from .temp_mail import TempMailService
-from .base import EmailServiceType
+from .base import BaseEmailService, EmailServiceError, EmailServiceType
+from ..config.constants import OTP_CODE_PATTERN
+logger = logging.getLogger(__name__)
-class CloudMailService(TempMailService):
- """?? CloudMail ????????"""
+
+class CloudMailService(BaseEmailService):
+ """
+ Cloud Mail 邮箱服务
+ 基于 Cloudflare Workers 的自部署邮箱服务
+ """
+
+ # 类变量：所有实例共享token（按base_url区分）
+ _shared_tokens: Dict[str, tuple] = {} # {base_url: (token, expires_at)}
+ _token_lock = None # 延迟初始化
+ _seen_ids_lock = None # seen_email_ids 的锁
+ _shared_seen_email_ids: Dict[str, set] = {} # 所有实例共享已处理的邮件ID（按邮箱地址区分）
def __init__(self, config: Dict[str, Any] = None, name: str = None):
- normalized = dict(config or {})
- normalized.setdefault("enable_prefix", True)
- super().__init__(normalized, name)
- self.service_type = EmailServiceType.CLOUDMAIL
+ """
+ 初始化 Cloud Mail 服务
+
+ Args:
+ config: 配置字典，支持以下键:
+ - base_url: API 基础地址 (必需)
+ - admin_email: 管理员邮箱 (必需)
+ - admin_password: 管理员密码 (必需)
+ - domain: 邮箱域名 (可选，用于生成邮箱地址)
+ - subdomain: 子域名 (可选)，会插入到 @ 和域名之间，例如 subdomain="test" 会生成 xxx@test.example.com
+ - timeout: 请求超时时间，默认 30
+ - max_retries: 最大重试次数，默认 3
+ - proxy_url: 代理地址 (可选)
+ name: 服务名称
+ """
+ super().__init__(EmailServiceType.CLOUDMAIL, name)
+
+ required_keys = ["base_url", "admin_email", "admin_password"]
+ missing_keys = [key for key in required_keys if not (config or {}).get(key)]
+ if missing_keys:
+ raise ValueError(f"缺少必需配置: {missing_keys}")
+
+ default_config = {
+ "timeout": 30,
+ "max_retries": 3,
+ "proxy_url": None,
+ }
+ self.config = {**default_config, **(config or {})}
+ self.config["base_url"] = self.config["base_url"].rstrip("/")
+
+ # 创建 requests session
+ self.session = requests.Session()
+ self.session.headers.update({
+ "Accept": "application/json",
+ "Content-Type": "application/json",
+ })
+
+ # 初始化类级别的锁（线程安全）
+ if CloudMailService._token_lock is None:
+ import threading
+ CloudMailService._token_lock = threading.Lock()
+ CloudMailService._seen_ids_lock = threading.Lock()
+
+ # 缓存邮箱信息（实例级别）
+ self._created_emails: Dict[str, Dict[str, Any]] = {}
+
+ def _generate_token(self) -> str:
+ """
+ 生成身份令牌
+
+ Returns:
+ token 字符串
+
+ Raises:
+ EmailServiceError: 生成失败
+ """
+ url = f"{self.config['base_url']}/api/public/genToken"
+ payload = {
+ "email": self.config["admin_email"],
+ "password": self.config["admin_password"]
+ }
+
+ try:
+ response = self.session.post(
+ url,
+ json=payload,
+ timeout=self.config["timeout"]
+ )
+
+ if response.status_code >= 400:
+ error_msg = f"生成 token 失败: {response.status_code}"
+ try:
+ error_data = response.json()
+ error_msg = f"{error_msg} - {error_data}"
+ except Exception:
+ error_msg = f"{error_msg} - {response.text[:200]}"
+ raise EmailServiceError(error_msg)
+
+ data = response.json()
+ if data.get("code") != 200:
+ raise EmailServiceError(f"生成 token 失败: {data.get('message', 'Unknown error')}")
+
+ token = data.get("data", {}).get("token")
+ if not token:
+ raise EmailServiceError("生成 token 失败: 未返回 token")
+
+ return token
+
+ except requests.RequestException as e:
+ self.update_status(False, e)
+ raise EmailServiceError(f"生成 token 失败: {e}")
+ except Exception as e:
+ self.update_status(False, e)
+ if isinstance(e, EmailServiceError):
+ raise
+ raise EmailServiceError(f"生成 token 失败: {e}")
+
+ def _get_token(self, force_refresh: bool = False) -> str:
+ """
+ 获取有效的 token（带缓存，所有实例共享）
+
+ Args:
+ force_refresh: 是否强制刷新
+
+ Returns:
+ token 字符串
+ """
+ base_url = self.config["base_url"]
+
+ with CloudMailService._token_lock:
+ # 检查共享缓存（token 有效期设为 1 小时）
+ if not force_refresh and base_url in CloudMailService._shared_tokens:
+ token, expires_at = CloudMailService._shared_tokens[base_url]
+ if time.time() < expires_at:
+ return token
+
+ # 生成新 token
+ token = self._generate_token()
+ expires_at = time.time() + 3600 # 1 小时后过期
+ CloudMailService._shared_tokens[base_url] = (token, expires_at)
+ return token
+
+ def _get_headers(self, token: Optional[str] = None) -> Dict[str, str]:
+ """构造请求头"""
+ if token is None:
+ token = self._get_token()
+
+ return {
+ "Authorization": token,
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ }
+
+ def _make_request(
+ self,
+ method: str,
+ path: str,
+ retry_on_auth_error: bool = True,
+ **kwargs
+ ) -> Any:
+ """
+ 发送请求并返回 JSON 数据
+
+ Args:
+ method: HTTP 方法
+ path: 请求路径（以 / 开头）
+ retry_on_auth_error: 认证失败时是否重试
+ **kwargs: 传递给 requests 的额外参数
+
+ Returns:
+ 响应 JSON 数据
+
+ Raises:
+ EmailServiceError: 请求失败
+ """
+ url = f"{self.config['base_url']}{path}"
+ kwargs.setdefault("headers", {})
+ kwargs["headers"].update(self._get_headers())
+ kwargs.setdefault("timeout", self.config["timeout"])
+
+ try:
+ response = self.session.request(method, url, **kwargs)
+
+ if response.status_code >= 400:
+ # 如果是认证错误且允许重试，刷新 token 后重试一次
+ if response.status_code == 401 and retry_on_auth_error:
+ logger.warning("Cloud Mail 认证失败，尝试刷新 token")
+ kwargs["headers"].update(self._get_headers(self._get_token(force_refresh=True)))
+ response = self.session.request(method, url, **kwargs)
+
+ if response.status_code >= 400:
+ error_msg = f"请求失败: {response.status_code}"
+ try:
+ error_data = response.json()
+ error_msg = f"{error_msg} - {error_data}"
+ except Exception:
+ error_msg = f"{error_msg} - {response.text[:200]}"
+ self.update_status(False, EmailServiceError(error_msg))
+ raise EmailServiceError(error_msg)
+
+ try:
+ return response.json()
+ except Exception:
+ return {"raw_response": response.text}
+
+ except requests.RequestException as e:
+ self.update_status(False, e)
+ raise EmailServiceError(f"请求失败: {method} {path} - {e}")
+ except Exception as e:
+ self.update_status(False, e)
+ if isinstance(e, EmailServiceError):
+ raise
+ raise EmailServiceError(f"请求失败: {method} {path} - {e}")
+
+ def _generate_email_address(self, prefix: Optional[str] = None, domain: Optional[str] = None, subdomain: Optional[str] = None) -> str:
+ """
+ 生成邮箱地址
+
+ Args:
+ prefix: 邮箱前缀，如果不提供则随机生成
+ domain: 指定域名，如果不提供则从配置中选择
+ subdomain: 子域名，可选参数，会插入到 @ 和域名之间
+
+ Returns:
+ 完整的邮箱地址
+ """
+ if not prefix:
+ # 生成随机前缀：首字母 + 9位随机字符（共10位）
+ first = random.choice(string.ascii_lowercase)
+ rest = "".join(random.choices(string.ascii_lowercase + string.digits, k=9))
+ prefix = f"{first}{rest}"
+
+ # 如果没有指定域名，从配置中获取
+ if not domain:
+ domain_config = self.config.get("domain")
+ if not domain_config:
+ raise EmailServiceError("未配置邮箱域名，无法生成邮箱地址")
+
+ # 支持多个域名（列表）或单个域名（字符串）
+ if isinstance(domain_config, list):
+ if not domain_config:
+ raise EmailServiceError("域名列表为空")
+ # 随机选择一个域名
+ domain = random.choice(domain_config)
+ else:
+ domain = domain_config
+
+ # 如果提供了子域，插入到域名前面
+ if subdomain:
+ domain = f"{subdomain}.{domain}"
- def list_emails(self, limit: int = 100, offset: int = 0, **kwargs) -> List[Dict[str, Any]]:
- return super().list_emails(limit=limit, offset=offset, **kwargs)
+ return f"{prefix}@{domain}"
+
+ def _generate_password(self, length: int = 12) -> str:
+ """生成随机密码"""
+ alphabet = string.ascii_letters + string.digits
+ return "".join(random.choices(alphabet, k=length))
+
+ def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]:
+ """
+ 创建新邮箱地址
+
+ Args:
+ config: 配置参数:
+ - name: 邮箱前缀（可选）
+ - password: 邮箱密码（可选，不提供则自动生成）
+ - domain: 邮箱域名（可选，覆盖默认域名）
+ - subdomain: 子域名（可选），会插入到 @ 和域名之间，例如 subdomain="test" 会生成 xxx@test.example.com
+
+ Returns:
+ 包含邮箱信息的字典:
+ - email: 邮箱地址
+ - service_id: 邮箱地址（用作标识）
+ - password: 邮箱密码
+ """
+ req_config = config or {}
+
+ # 生成邮箱地址
+ prefix = req_config.get("name")
+ specified_domain = req_config.get("domain")
+ subdomain = req_config.get("subdomain") or self.config.get("subdomain")
+
+ if specified_domain:
+ email_address = self._generate_email_address(prefix, specified_domain, subdomain)
+ else:
+ email_address = self._generate_email_address(prefix, subdomain=subdomain)
+
+ # 生成或使用提供的密码
+ password = req_config.get("password") or self._generate_password()
+
+ # 直接生成邮箱信息（catch-all 域名无需预先创建）
+ email_info = {
+ "email": email_address,
+ "service_id": email_address,
+ "id": email_address,
+ "password": password,
+ "created_at": time.time(),
+ }
+
+ # 缓存邮箱信息
+ self._created_emails[email_address] = email_info
+ self.update_status(True)
+
+ logger.info(f"生成 CloudMail 邮箱: {email_address}")
+ return email_info
def get_verification_code(
self,
email: str,
email_id: str = None,
timeout: int = 120,
- pattern: str = r"(? Optional[str]:
- return super().get_verification_code(
- email=email,
- email_id=email_id,
- timeout=timeout,
- pattern=pattern,
- otp_sent_at=otp_sent_at,
- )
+ """
+ 从 Cloud Mail 邮箱获取验证码
+
+ Args:
+ email: 邮箱地址
+ email_id: 未使用，保留接口兼容
+ timeout: 超时时间（秒）
+ pattern: 验证码正则
+ otp_sent_at: OTP 发送时间戳
+
+ Returns:
+ 验证码字符串，超时返回 None
+ """
+ start_time = time.time()
+
+ # 每次调用时，记录本次查询开始前已存在的邮件ID
+ # 这样可以支持同一个邮箱多次接收验证码（注册+OAuth）
+ initial_seen_ids = set()
+ with CloudMailService._seen_ids_lock:
+ if email not in CloudMailService._shared_seen_email_ids:
+ CloudMailService._shared_seen_email_ids[email] = set()
+ else:
+ # 记录本次查询开始前的已处理邮件
+ initial_seen_ids = CloudMailService._shared_seen_email_ids[email].copy()
+
+ # 本次查询中新处理的邮件ID（仅在本次查询中有效）
+ current_seen_ids = set()
+
+ check_count = 0
+
+ while time.time() - start_time < timeout:
+ try:
+ check_count += 1
+
+ # 查询邮件列表
+ url_path = "/api/public/emailList"
+ payload = {
+ "toEmail": email,
+ "timeSort": "desc" # 最新的邮件优先
+ }
+
+ result = self._make_request("POST", url_path, json=payload)
+
+ if result.get("code") != 200:
+ time.sleep(3)
+ continue
+
+ emails = result.get("data", [])
+ if not isinstance(emails, list):
+ time.sleep(3)
+ continue
+
+ for email_item in emails:
+ email_id = email_item.get("emailId")
+
+ if not email_id:
+ continue
+
+ # 跳过本次查询开始前已存在的邮件
+ if email_id in initial_seen_ids:
+ continue
+
+ # 跳过本次查询中已处理的邮件（防止同一轮查询重复处理）
+ if email_id in current_seen_ids:
+ continue
+
+ # 标记为本次已处理
+ current_seen_ids.add(email_id)
+
+ # 同时更新全局已处理列表（防止其他并发任务重复处理）
+ with CloudMailService._seen_ids_lock:
+ CloudMailService._shared_seen_email_ids[email].add(email_id)
+
+ sender_email = str(email_item.get("sendEmail", "")).lower()
+ sender_name = str(email_item.get("sendName", "")).lower()
+ subject = str(email_item.get("subject", ""))
+ to_email = email_item.get("toEmail", "")
+
+ # 检查收件人是否匹配
+ if to_email != email:
+ continue
+
+ if "openai" not in sender_email and "openai" not in sender_name:
+ continue
+
+ # 从主题提取
+ match = re.search(pattern, subject)
+ if match:
+ code = match.group(1)
+ self.update_status(True)
+ return code
+
+ # 从内容提取
+ content = str(email_item.get("content", ""))
+ if content:
+ clean_content = re.sub(r"<[^>]+>", " ", content)
+ email_pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
+ clean_content = re.sub(email_pattern, "", clean_content)
+
+ match = re.search(pattern, clean_content)
+ if match:
+ code = match.group(1)
+ self.update_status(True)
+ return code
+
+ except Exception as e:
+ # 如果是认证错误，强制刷新token
+ if "401" in str(e) or "认证" in str(e):
+ try:
+ self._get_token(force_refresh=True)
+ except Exception:
+ pass
+ logger.error(f"检查邮件时出错: {e}", exc_info=True)
+
+ time.sleep(3)
+
+ # 超时
+ logger.warning(f"等待验证码超时: {email}")
+ return None
+
+ def list_emails(self, **kwargs) -> List[Dict[str, Any]]:
+ """
+ 列出已创建的邮箱（从缓存中获取）
+
+ Returns:
+ 邮箱列表
+ """
+ return list(self._created_emails.values())
+
+ def delete_email(self, email_id: str) -> bool:
+ """
+ 删除邮箱（Cloud Mail API 不支持删除用户，仅从缓存中移除）
+
+ Args:
+ email_id: 邮箱地址
+
+ Returns:
+ 是否删除成功
+ """
+ if email_id in self._created_emails:
+ del self._created_emails[email_id]
+ return True
+
+ return False
+
+ def check_health(self) -> bool:
+ """检查服务健康状态"""
+ try:
+ # 尝试生成 token
+ self._get_token(force_refresh=True)
+ self.update_status(True)
+ return True
+ except Exception as e:
+ logger.warning(f"Cloud Mail 健康检查失败: {e}")
+ self.update_status(False, e)
+ return False
+
+ def get_email_messages(self, email_id: str, **kwargs) -> List[Dict[str, Any]]:
+ """
+ 获取邮箱中的邮件列表
+
+ Args:
+ email_id: 邮箱地址
+ **kwargs: 额外参数（如 timeSort）
+
+ Returns:
+ 邮件列表
+ """
+ try:
+ url_path = "/api/public/emailList"
+ payload = {
+ "toEmail": email_id,
+ "timeSort": kwargs.get("timeSort", "desc")
+ }
+
+ result = self._make_request("POST", url_path, json=payload)
+
+ if result.get("code") != 200:
+ logger.warning(f"获取邮件列表失败: {result.get('message')}")
+ return []
+
+ self.update_status(True)
+ return result.get("data", [])
+
+ except Exception as e:
+ logger.error(f"获取 Cloud Mail 邮件列表失败: {email_id} - {e}")
+ self.update_status(False, e)
+ return []
+
+ def get_service_info(self) -> Dict[str, Any]:
+ """获取服务信息"""
+ return {
+ "service_type": self.service_type.value,
+ "name": self.name,
+ "base_url": self.config["base_url"],
+ "admin_email": self.config["admin_email"],
+ "domain": self.config.get("domain"),
+ "subdomain": self.config.get("subdomain"),
+ "cached_emails_count": len(self._created_emails),
+ "status": self.status.value,
+ }
diff --git a/src/web/app.py b/src/web/app.py
index 40bbef8e..e8c65b31 100644
--- a/src/web/app.py
+++ b/src/web/app.py
@@ -120,21 +120,29 @@ async def periodic_log_cleanup() -> None:
try:
yield
finally:
+ async def _cancel_and_wait(task):
+ if not task:
+ return
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
+ pass
+ except Exception as exc:
+ logger.warning("Background task shutdown failed: %s", exc)
+
cleanup_task = getattr(app.state, "log_cleanup_task", None)
- if cleanup_task:
- cleanup_task.cancel()
+ await _cancel_and_wait(cleanup_task)
scheduler_service = getattr(app.state, "scheduled_registration_service", None)
if scheduler_service:
await scheduler_service.stop()
auto_quick_refresh_task = getattr(app.state, "auto_quick_refresh_task", None)
- if auto_quick_refresh_task:
- auto_quick_refresh_task.cancel()
+ await _cancel_and_wait(auto_quick_refresh_task)
selfcheck_scheduler_task = getattr(app.state, "selfcheck_scheduler_task", None)
- if selfcheck_scheduler_task:
- selfcheck_scheduler_task.cancel()
+ await _cancel_and_wait(selfcheck_scheduler_task)
if auto_registration_coordinator is not None:
await auto_registration_coordinator.stop()
diff --git a/src/web/auto_quick_refresh_scheduler.py b/src/web/auto_quick_refresh_scheduler.py
index b826b93c..4861fc5b 100644
--- a/src/web/auto_quick_refresh_scheduler.py
+++ b/src/web/auto_quick_refresh_scheduler.py
@@ -13,7 +13,7 @@
import logging
import threading
from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Set, Tuple
from ..config.settings import get_settings
@@ -60,6 +60,7 @@ def __init__(self) -> None:
self._last_result: Dict[str, Any] = {}
self._consecutive_failures: int = 0
self._logs: List[Dict[str, str]] = []
+ self._run_tasks: Set[asyncio.Task] = set()
def _append_log_locked(self, level: str, message: str, when: Optional[datetime] = None) -> None:
entry = {
@@ -157,6 +158,18 @@ def request_run_now(self, reason: str = "manual") -> Dict[str, Any]:
self._append_log_locked("info", "已请求立即执行一次", when=now)
return self._snapshot_locked()
+ def _track_run_task(self, task: asyncio.Task) -> None:
+ self._run_tasks.add(task)
+ task.add_done_callback(lambda done: self._run_tasks.discard(done))
+
+ async def _cancel_run_tasks(self) -> None:
+ pending = [task for task in list(self._run_tasks) if not task.done()]
+ if not pending:
+ return
+ for task in pending:
+ task.cancel()
+ await asyncio.gather(*pending, return_exceptions=True)
+
async def run_loop(self) -> None:
logger.info("自动一键刷新调度器启动")
self._append_log("info", "调度器已启动")
@@ -167,6 +180,7 @@ async def run_loop(self) -> None:
except asyncio.CancelledError:
logger.info("自动一键刷新调度器已停止")
self._append_log("info", "调度器已停止")
+ await self._cancel_run_tasks()
break
except Exception as exc:
logger.warning("自动一键刷新调度器异常: %s", exc)
@@ -207,7 +221,8 @@ async def _tick_once(self) -> None:
self._append_log_locked("info", f"开始执行（{reason}）", when=now)
if should_start:
- asyncio.create_task(self._run_once(schedule, reason))
+ task = asyncio.create_task(self._run_once(schedule, reason))
+ self._track_run_task(task)
async def _run_once(self, schedule: Dict[str, Any], reason: str) -> None:
run_status = "failed"
diff --git a/src/web/routes/registration.py b/src/web/routes/registration.py
index 0a82c773..be9934ab 100644
--- a/src/web/routes/registration.py
+++ b/src/web/routes/registration.py
@@ -20,7 +20,11 @@
from ...database import crud
from ...database.session import get_db
from ...database.models import RegistrationTask, ScheduledRegistrationJob, Proxy
-from ...core.register import RegistrationEngine, RegistrationResult
+from ...core.register import (
+ RegistrationEngine,
+ RegistrationResult,
+ RegistrationCancelledError,
+)
from ...services import EmailServiceFactory, EmailServiceType
from ...config.settings import get_settings, Settings
from ...core.auto_registration import (
@@ -346,11 +350,25 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
这个函数会被 run_in_executor 调用，运行在独立线程中
"""
with get_db() as db:
- try:
- # 检查是否已取消
+ def _mark_cancelled(reason: str = "任务已取消") -> None:
+ safe_reason = str(reason or "任务已取消").strip()
+ crud.update_registration_task(
+ db,
+ task_uuid,
+ status="cancelled",
+ completed_at=utcnow_naive(),
+ error_message=safe_reason,
+ )
+ task_manager.update_status(task_uuid, "cancelled", error=safe_reason)
+ task_manager.add_log(task_uuid, f"{log_prefix} [取消] {safe_reason}" if log_prefix else f"[取消] {safe_reason}")
+ logger.info("任务 %s 已取消: %s", task_uuid, safe_reason)
+
+ def _raise_if_cancelled(reason: str = "任务已取消") -> None:
if task_manager.is_cancelled(task_uuid):
- logger.info(f"任务 {task_uuid} 已取消，跳过执行")
- return
+ raise RegistrationCancelledError(reason)
+
+ try:
+ _raise_if_cancelled("任务在入队后收到取消请求，已跳过执行")
# 更新任务状态为运行中
task = crud.update_registration_task(
@@ -365,6 +383,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
# 更新 TaskManager 状态
task_manager.update_status(task_uuid, "running")
+ _raise_if_cancelled("任务在启动后收到取消请求，停止执行")
# 确定使用的代理
# 如果前端传入了代理参数，使用传入的
@@ -379,6 +398,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
# 更新任务的代理记录
crud.update_registration_task(db, task_uuid, proxy=actual_proxy_url)
+ _raise_if_cancelled("任务在准备阶段收到取消请求，停止执行")
# 创建邮箱服务
service_type = EmailServiceType(email_service_type)
@@ -580,13 +600,15 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
email_service=email_service,
proxy_url=actual_proxy_url,
callback_logger=log_callback,
- task_uuid=task_uuid
+ task_uuid=task_uuid,
+ check_cancelled=task_manager.create_check_cancelled_callback(task_uuid),
)
# 执行注册
role_tag = normalize_role_tag(registration_type)
account_label = role_tag_to_account_label(role_tag)
result = engine.run()
+ _raise_if_cancelled("任务在注册流程中收到取消请求，停止后续处理")
marker = getattr(email_service, "mark_registration_outcome", None)
marker_context = {}
try:
@@ -599,6 +621,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
marker_context = {}
if result.success:
+ _raise_if_cancelled("任务在注册成功后收到取消请求，跳过后处理")
# 更新代理使用时间
update_proxy_usage(db, proxy_id)
@@ -609,6 +632,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
result.metadata = metadata
# 保存到数据库
+ _raise_if_cancelled("任务在保存账户前收到取消请求，停止后处理")
engine.save_to_database(result, account_label=account_label, role_tag=role_tag)
if callable(marker) and result.email:
@@ -623,6 +647,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
# 自动上传到 CPA（可多服务）
if auto_upload_cpa:
+ _raise_if_cancelled("任务在 CPA 上传前收到取消请求，停止后处理")
try:
from ...core.upload.cpa_upload import upload_to_cpa, generate_token_json
from ...database.models import Account as AccountModel
@@ -636,6 +661,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
if not _cpa_ids:
log_callback("[CPA] 无可用 CPA 服务，跳过上传")
for _sid in _cpa_ids:
+ _raise_if_cancelled("任务在 CPA 上传过程中收到取消请求")
try:
_svc = crud.get_cpa_service_by_id(db, _sid)
if not _svc:
@@ -656,6 +682,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
# 自动上传到 Sub2API（可多服务）
if auto_upload_sub2api:
+ _raise_if_cancelled("任务在 Sub2API 上传前收到取消请求，停止后处理")
try:
from ...core.upload.sub2api_upload import upload_to_sub2api
from ...database.models import Account as AccountModel
@@ -667,6 +694,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
if not _s2a_ids:
log_callback("[Sub2API] 无可用 Sub2API 服务，跳过上传")
for _sid in _s2a_ids:
+ _raise_if_cancelled("任务在 Sub2API 上传过程中收到取消请求")
try:
_svc = crud.get_sub2api_service_by_id(db, _sid)
if not _svc:
@@ -681,6 +709,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
# 自动上传到 Team Manager（可多服务）
if auto_upload_tm:
+ _raise_if_cancelled("任务在 Team Manager 上传前收到取消请求，停止后处理")
try:
from ...core.upload.team_manager_upload import upload_to_team_manager
from ...database.models import Account as AccountModel
@@ -692,6 +721,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
if not _tm_ids:
log_callback("[TM] 无可用 Team Manager 服务，跳过上传")
for _sid in _tm_ids:
+ _raise_if_cancelled("任务在 Team Manager 上传过程中收到取消请求")
try:
_svc = crud.get_tm_service_by_id(db, _sid)
if not _svc:
@@ -705,6 +735,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
log_callback(f"[TM] 上传异常: {tm_err}")
if auto_upload_new_api:
+ _raise_if_cancelled("任务在 NewAPI 上传前收到取消请求，停止后处理")
try:
from ...core.upload.new_api_upload import upload_to_new_api
from ...database.models import Account as AccountModel
@@ -716,6 +747,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
if not _new_api_ids:
log_callback("[NewAPI] 无可用 new-api 服务，跳过上传")
for _sid in _new_api_ids:
+ _raise_if_cancelled("任务在 NewAPI 上传过程中收到取消请求")
try:
_svc = crud.get_new_api_service_by_id(db, _sid)
if not _svc:
@@ -734,6 +766,7 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
log_callback(f"[NewAPI] 上传异常: {new_api_err}")
# 更新任务状态
+ _raise_if_cancelled("任务在收尾前收到取消请求，停止标记成功")
crud.update_registration_task(
db, task_uuid,
status="completed",
@@ -746,6 +779,9 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
logger.info(f"注册任务完成: {task_uuid}, 邮箱: {result.email}")
else:
+ if task_manager.is_cancelled(task_uuid):
+ _mark_cancelled("任务在注册失败分支检测到取消请求")
+ return
if callable(marker) and result.email:
try:
marker(
@@ -770,6 +806,8 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
logger.warning(f"注册任务失败: {task_uuid}, 原因: {result.error_message}")
+ except RegistrationCancelledError as cancelled:
+ _mark_cancelled(str(cancelled) or "任务已取消")
except Exception as e:
logger.error(f"注册任务异常: {task_uuid}, 错误: {e}")
@@ -892,6 +930,18 @@ async def run_batch_parallel(
async def _run_one(idx: int, uuid: str):
prefix = f"[任务{idx + 1}]"
async with semaphore:
+ if task_manager.is_batch_cancelled(batch_id) or batch_tasks[batch_id]["cancelled"]:
+ with get_db() as db:
+ crud.update_registration_task(
+ db,
+ uuid,
+ status="cancelled",
+ completed_at=utcnow_naive(),
+ error_message="批量任务已取消",
+ )
+ task_manager.cancel_task(uuid)
+ task_manager.update_status(uuid, "cancelled", error="批量任务已取消")
+ return
await run_registration_task(
uuid, email_service_type, proxy, email_service_config, email_service_id,
log_prefix=prefix, batch_id=batch_id,
@@ -963,6 +1013,18 @@ async def run_batch_pipeline(
async def _run_and_release(idx: int, uuid: str, pfx: str):
try:
+ if task_manager.is_batch_cancelled(batch_id) or batch_tasks[batch_id]["cancelled"]:
+ with get_db() as db:
+ crud.update_registration_task(
+ db,
+ uuid,
+ status="cancelled",
+ completed_at=utcnow_naive(),
+ error_message="批量任务已取消",
+ )
+ task_manager.cancel_task(uuid)
+ task_manager.update_status(uuid, "cancelled", error="批量任务已取消")
+ return
await run_registration_task(
uuid, email_service_type, proxy, email_service_config, email_service_id,
log_prefix=pfx, batch_id=batch_id,
@@ -1637,7 +1699,16 @@ async def cancel_task(task_uuid: str):
if task.status not in ["pending", "running"]:
raise HTTPException(status_code=400, detail="任务已完成或已取消")
- task = crud.update_registration_task(db, task_uuid, status="cancelled")
+ task_manager.cancel_task(task_uuid)
+ task = crud.update_registration_task(
+ db,
+ task_uuid,
+ status="cancelled",
+ completed_at=utcnow_naive(),
+ error_message="任务取消请求已提交",
+ )
+ task_manager.update_status(task_uuid, "cancelled", error="任务取消请求已提交")
+ task_manager.add_log(task_uuid, "[取消] 任务取消请求已提交，等待后台线程有序退出")
return {"success": True, "message": "任务已取消"}
@@ -2141,6 +2212,7 @@ async def cancel_outlook_batch(batch_id: str):
# 同时更新两个系统的取消状态
batch["cancelled"] = True
task_manager.cancel_batch(batch_id)
+ _cancel_batch_tasks(batch_id)
return {"success": True, "message": "批量任务取消请求已提交，正在让它们有序收工"}
diff --git a/src/web/selfcheck_scheduler.py b/src/web/selfcheck_scheduler.py
index a9b141d2..dad762d7 100644
--- a/src/web/selfcheck_scheduler.py
+++ b/src/web/selfcheck_scheduler.py
@@ -8,7 +8,7 @@
import logging
import threading
from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Set
from ..config.settings import get_settings
from ..core.system_selfcheck import create_selfcheck_run, execute_selfcheck_run, has_running_selfcheck_run
@@ -59,6 +59,7 @@ def __init__(self) -> None:
self._last_run: Optional[Dict[str, Any]] = None
self._consecutive_failures: int = 0
self._logs: List[Dict[str, str]] = []
+ self._run_tasks: Set[asyncio.Task] = set()
def _read_schedule(self) -> Dict[str, Any]:
settings = get_settings()
@@ -141,6 +142,18 @@ def request_run_now(self, reason: str = "manual") -> Dict[str, Any]:
self._append_log_locked("info", "已请求立即执行一次系统自检", now)
return self._snapshot_locked()
+ def _track_run_task(self, task: asyncio.Task) -> None:
+ self._run_tasks.add(task)
+ task.add_done_callback(lambda done: self._run_tasks.discard(done))
+
+ async def _cancel_run_tasks(self) -> None:
+ pending = [task for task in list(self._run_tasks) if not task.done()]
+ if not pending:
+ return
+ for task in pending:
+ task.cancel()
+ await asyncio.gather(*pending, return_exceptions=True)
+
async def run_loop(self) -> None:
logger.info("系统自检调度器启动")
self._append_log("info", "调度器已启动")
@@ -151,6 +164,7 @@ async def run_loop(self) -> None:
except asyncio.CancelledError:
logger.info("系统自检调度器已停止")
self._append_log("info", "调度器已停止")
+ await self._cancel_run_tasks()
break
except Exception as exc:
logger.warning("系统自检调度器异常: %s", exc)
@@ -189,7 +203,8 @@ async def _tick_once(self) -> None:
self._append_log_locked("info", f"开始执行系统自检（{reason}）", now)
if should_start:
- asyncio.create_task(self._run_once(schedule, reason))
+ task = asyncio.create_task(self._run_once(schedule, reason))
+ self._track_run_task(task)
async def _run_once(self, schedule: Dict[str, Any], reason: str) -> None:
mode = _normalize_mode(schedule.get("mode"))
diff --git a/static/js/app.js b/static/js/app.js
index d4d84193..232e52ec 100644
--- a/static/js/app.js
+++ b/static/js/app.js
@@ -669,6 +669,8 @@ function buildCurrentRegistrationConfig() {
throw new Error('请选择一个邮箱服务');
}
+ isBatchMode = elements.regMode.value == "batch"
+
const [emailServiceType, serviceId] = selectedValue.split(':');
const baseConfig = {
email_service_type: emailServiceType,
diff --git a/static/js/email_services.js b/static/js/email_services.js
index be426a16..0cd23ceb 100644
--- a/static/js/email_services.js
+++ b/static/js/email_services.js
@@ -60,6 +60,7 @@ const elements = {
addDuckmailFields: document.getElementById('add-duckmail-fields'),
addLuckmailFields: document.getElementById('add-luckmail-fields'),
addFreemailFields: document.getElementById('add-freemail-fields'),
+ addCloudmailFields: document.getElementById('add-cloudmail-fields'),
addImapFields: document.getElementById('add-imap-fields'),
// 编辑自定义域名模态框
@@ -74,6 +75,7 @@ const elements = {
editDuckmailFields: document.getElementById('edit-duckmail-fields'),
editLuckmailFields: document.getElementById('edit-luckmail-fields'),
editFreemailFields: document.getElementById('edit-freemail-fields'),
+ editCloudmailFields: document.getElementById('edit-cloudmail-fields'),
editImapFields: document.getElementById('edit-imap-fields'),
editCustomTypeBadge: document.getElementById('edit-custom-type-badge'),
editCustomSubTypeHidden: document.getElementById('edit-custom-sub-type-hidden'),
@@ -200,10 +202,11 @@ function switchAddSubType(subType) {
elements.addMoemailFields.style.display = subType === 'moemail' ? '' : 'none';
elements.addTempmailBuiltinFields.style.display = subType === 'tempmail_builtin' ? '' : 'none';
elements.addYydsFields.style.display = subType === 'yyds_mail' ? '' : 'none';
- elements.addTempmailFields.style.display = (subType === 'tempmail' || subType === 'cloudmail') ? '' : 'none';
+ elements.addTempmailFields.style.display = subType === 'tempmail' ? '' : 'none';
elements.addDuckmailFields.style.display = subType === 'duckmail' ? '' : 'none';
elements.addLuckmailFields.style.display = subType === 'luckmail' ? '' : 'none';
elements.addFreemailFields.style.display = subType === 'freemail' ? '' : 'none';
+ elements.addCloudmailFields.style.display = subType === 'cloudmail' ? '' : 'none';
elements.addImapFields.style.display = subType === 'imap' ? '' : 'none';
}
@@ -213,10 +216,11 @@ function switchEditSubType(subType) {
elements.editMoemailFields.style.display = subType === 'moemail' ? '' : 'none';
elements.editTempmailBuiltinFields.style.display = subType === 'tempmail_builtin' ? '' : 'none';
elements.editYydsFields.style.display = subType === 'yyds_mail' ? '' : 'none';
- elements.editTempmailFields.style.display = (subType === 'tempmail' || subType === 'cloudmail') ? '' : 'none';
+ elements.editTempmailFields.style.display = subType === 'tempmail' ? '' : 'none';
elements.editDuckmailFields.style.display = subType === 'duckmail' ? '' : 'none';
elements.editLuckmailFields.style.display = subType === 'luckmail' ? '' : 'none';
elements.editFreemailFields.style.display = subType === 'freemail' ? '' : 'none';
+ elements.editCloudmailFields.style.display = subType === 'cloudmail' ? '' : 'none';
elements.editImapFields.style.display = subType === 'imap' ? '' : 'none';
elements.editCustomTypeBadge.textContent = CUSTOM_SUBTYPE_LABELS[subType] || CUSTOM_SUBTYPE_LABELS.moemail;
}
@@ -548,8 +552,8 @@ async function handleAddCustom(e) {
timeout: parseInt(formData.get('yyds_timeout'), 10) || 30,
max_retries: parseInt(formData.get('yyds_max_retries'), 10) || 3
};
- } else if (subType === 'tempmail' || subType === 'cloudmail') {
- serviceType = subType === 'cloudmail' ? 'cloudmail' : 'temp_mail';
+ } else if (subType === 'tempmail') {
+ serviceType = 'temp_mail';
config = {
base_url: formData.get('tm_base_url'),
admin_password: formData.get('tm_admin_password'),
@@ -580,6 +584,23 @@ async function handleAddCustom(e) {
admin_token: formData.get('fm_admin_token'),
domain: formData.get('fm_domain')
};
+ } else if (subType === 'cloudmail') {
+ serviceType = 'cloud_mail';
+ const domainInput = formData.get('cm_domain');
+ let domain = domainInput;
+ if (domainInput && domainInput.includes(',')) {
+ domain = domainInput.split(',').map(d => d.trim()).filter(d => d);
+ }
+ config = {
+ base_url: formData.get('cm_base_url'),
+ admin_email: formData.get('cm_admin_email'),
+ admin_password: formData.get('cm_admin_password'),
+ domain: domain
+ };
+ const subdomain = formData.get('cm_subdomain');
+ if (subdomain && subdomain.trim()) {
+ config.subdomain = subdomain.trim();
+ }
} else {
serviceType = 'imap_mail';
config = {
@@ -798,7 +819,7 @@ async function editCustomService(id, subType) {
document.getElementById('edit-yyds-default-domain').value = service.config?.default_domain || '';
document.getElementById('edit-yyds-timeout').value = service.config?.timeout || 30;
document.getElementById('edit-yyds-max-retries').value = service.config?.max_retries || 3;
- } else if (resolvedSubType === 'tempmail' || resolvedSubType === 'cloudmail') {
+ } else if (resolvedSubType === 'tempmail') {
document.getElementById('edit-tm-base-url').value = service.config?.base_url || '';
document.getElementById('edit-tm-admin-password').value = '';
document.getElementById('edit-tm-admin-password').placeholder = service.config?.admin_password ? '已设置，留空保持不变' : '请输入 Admin 密码';
@@ -821,6 +842,15 @@ async function editCustomService(id, subType) {
document.getElementById('edit-fm-admin-token').value = '';
document.getElementById('edit-fm-admin-token').placeholder = service.config?.admin_token ? '已设置，留空保持不变' : '请输入 Admin Token';
document.getElementById('edit-fm-domain').value = service.config?.domain || '';
+ }else if (resolvedSubType === 'cloudmail') {
+ document.getElementById('edit-cm-base-url').value = service.config?.base_url || '';
+ document.getElementById('edit-cm-admin-email').value = service.config?.admin_email || '';
+ document.getElementById('edit-cm-admin-password').value = '';
+ document.getElementById('edit-cm-admin-password').placeholder = service.config?.admin_password ? '已设置，留空保持不变' : '请输入管理员密码';
+ const domain = service.config?.domain;
+ const domainStr = Array.isArray(domain) ? domain.join(', ') : (domain || '');
+ document.getElementById('edit-cm-domain').value = domainStr;
+ document.getElementById('edit-cm-subdomain').value = service.config?.subdomain || '';
} else {
document.getElementById('edit-imap-host').value = service.config?.host || '';
document.getElementById('edit-imap-port').value = service.config?.port || 993;
@@ -866,7 +896,7 @@ async function handleEditCustom(e) {
};
const apiKey = formData.get('yyds_api_key');
if (apiKey && apiKey.trim()) config.api_key = apiKey.trim();
- } else if (subType === 'tempmail' || subType === 'cloudmail') {
+ } else if (subType === 'tempmail') {
config = {
base_url: formData.get('tm_base_url'),
domain: formData.get('tm_domain'),
@@ -898,6 +928,25 @@ async function handleEditCustom(e) {
};
const token = formData.get('fm_admin_token');
if (token && token.trim()) config.admin_token = token.trim();
+ } else if (subType === 'cloudmail') {
+ const domainInput = formData.get('cm_domain');
+ // 处理域名：如果包含逗号，转换为数组；否则保持字符串
+ let domain = domainInput;
+ if (domainInput && domainInput.includes(',')) {
+ domain = domainInput.split(',').map(d => d.trim()).filter(d => d);
+ }
+ config = {
+ base_url: formData.get('cm_base_url'),
+ admin_email: formData.get('cm_admin_email'),
+ domain: domain
+ };
+ // 添加子域配置（如果有）
+ const subdomain = formData.get('cm_subdomain');
+ if (subdomain && subdomain.trim()) {
+ config.subdomain = subdomain.trim();
+ }
+ const pwd = formData.get('cm_admin_password');
+ if (pwd && pwd.trim()) config.admin_password = pwd.trim();
} else {
config = {
host: formData.get('imap_host'),
diff --git a/templates/email_services.html b/templates/email_services.html
index 6a9d7d6f..9baeb325 100644
--- a/templates/email_services.html
+++ b/templates/email_services.html
@@ -406,6 +406,31 @@ ➕ 添加自定义邮箱服务
+
+
+
+