GithubBackup/codex-register
1
0
Fork 0
mirror of https://github.com/cnlimiter/codex-register.git synced 2026-05-14 20:08:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 新增 freemail 邮箱服务渠道支持

Browse Source
This commit is contained in:
yunxilyf
2026-03-19 23:33:31 +08:00
parent 7f8e85b0aa
commit f4f17ebb5d
11 changed files with 489 additions and 1157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/config/constants.py
View File

@@ -36,6 +36,7 @@ class EmailServiceType(str, Enum):
CUSTOM_DOMAIN = "custom_domain"
TEMP_MAIL = "temp_mail"
DUCK_MAIL = "duck_mail"
FREEMAIL = "freemail"
# ============================================================================
@@ -120,6 +121,13 @@ EMAIL_SERVICE_DEFAULTS = {
"password_length": 12,
"timeout": 30,
"max_retries": 3,
},
"freemail": {
"base_url": "",
"admin_token": "",
"domain": "",
"timeout": 30,
"max_retries": 3,
}
}

3
src/services/__init__.py
View File

@@ -15,6 +15,7 @@ from .outlook import OutlookService
from .moe_mail import MeoMailEmailService
from .temp_mail import TempMailService
from .duck_mail import DuckMailService
from .freemail import FreemailService
# 注册服务
EmailServiceFactory.register(EmailServiceType.TEMPMAIL, TempmailService)
@@ -22,6 +23,7 @@ EmailServiceFactory.register(EmailServiceType.OUTLOOK, OutlookService)
EmailServiceFactory.register(EmailServiceType.CUSTOM_DOMAIN, MeoMailEmailService)
EmailServiceFactory.register(EmailServiceType.TEMP_MAIL, TempMailService)
EmailServiceFactory.register(EmailServiceType.DUCK_MAIL, DuckMailService)
EmailServiceFactory.register(EmailServiceType.FREEMAIL, FreemailService)
# 导出 Outlook 模块的额外内容
from .outlook.base import (
@@ -53,6 +55,7 @@ __all__ = [
'MeoMailEmailService',
'TempMailService',
'DuckMailService',
'FreemailService',
# Outlook 模块
'ProviderType',
'EmailMessage',

324
src/services/freemail.py Normal file
View File

@@ -0,0 +1,324 @@
"""
Freemail 邮箱服务实现
基于自部署 Cloudflare Worker 临时邮箱服务 (https://github.com/idinging/freemail)
"""
import re
import time
import logging
import random
import string
from typing import Optional, Dict, Any, List
from .base import BaseEmailService, EmailServiceError, EmailServiceType
from ..core.http_client import HTTPClient, RequestConfig
from ..config.constants import OTP_CODE_PATTERN
logger = logging.getLogger(__name__)
class FreemailService(BaseEmailService):
"""
Freemail 邮箱服务
基于自部署 Cloudflare Worker 的临时邮箱
"""
def __init__(self, config: Dict[str, Any] = None, name: str = None):
"""
初始化 Freemail 服务
Args:
config: 配置字典，支持以下键:
- base_url: Worker 域名地址 (必需)
- admin_token: Admin Token，对应 JWT_TOKEN (必需)
- domain: 邮箱域名，如 example.com
- timeout: 请求超时时间，默认 30
- max_retries: 最大重试次数，默认 3
name: 服务名称
"""
super().__init__(EmailServiceType.FREEMAIL, name)
required_keys = ["base_url", "admin_token"]
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,
}
self.config = {**default_config, **(config or {})}
self.config["base_url"] = self.config["base_url"].rstrip("/")
http_config = RequestConfig(
timeout=self.config["timeout"],
max_retries=self.config["max_retries"],
)
self.http_client = HTTPClient(proxy_url=None, config=http_config)
# 缓存 domain 列表
self._domains = []
def _get_headers(self) -> Dict[str, str]:
"""构造 admin 请求头"""
return {
"Authorization": f"Bearer {self.config['admin_token']}",
"Content-Type": "application/json",
"Accept": "application/json",
}
def _make_request(self, method: str, path: str, **kwargs) -> Any:
"""
发送请求并返回 JSON 数据
Args:
method: HTTP 方法
path: 请求路径（以 / 开头）
**kwargs: 传递给 http_client.request 的额外参数
Returns:
响应 JSON 数据
Raises:
EmailServiceError: 请求失败
"""
url = f"{self.config['base_url']}{path}"
kwargs.setdefault("headers", {})
kwargs["headers"].update(self._get_headers())
try:
response = self.http_client.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 Exception as e:
self.update_status(False, e)
if isinstance(e, EmailServiceError):
raise
raise EmailServiceError(f"请求失败: {method} {path} - {e}")
def _ensure_domains(self):
"""获取并缓存可用域名列表"""
if not self._domains:
try:
domains = self._make_request("GET", "/api/domains")
if isinstance(domains, list):
self._domains = domains
except Exception as e:
logger.warning(f"获取 Freemail 域名列表失败: {e}")
def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]:
"""
通过 API 创建临时邮箱
Returns:
包含邮箱信息的字典:
- email: 邮箱地址
- service_id: 同 email（用作标识）
"""
self._ensure_domains()
req_config = config or {}
domain_index = 0
target_domain = req_config.get("domain") or self.config.get("domain")
if target_domain and self._domains:
for i, d in enumerate(self._domains):
if d == target_domain:
domain_index = i
break
prefix = req_config.get("name")
try:
if prefix:
body = {
"local": prefix,
"domainIndex": domain_index
}
resp = self._make_request("POST", "/api/create", json=body)
else:
params = {"domainIndex": domain_index}
length = req_config.get("length")
if length:
params["length"] = length
resp = self._make_request("GET", "/api/generate", params=params)
email = resp.get("email")
if not email:
raise EmailServiceError(f"创建邮箱失败，未返回邮箱地址: {resp}")
email_info = {
"email": email,
"service_id": email,
"id": email,
"created_at": time.time(),
}
logger.info(f"成功创建 Freemail 邮箱: {email}")
self.update_status(True)
return email_info
except Exception as e:
self.update_status(False, e)
if isinstance(e, EmailServiceError):
raise
raise EmailServiceError(f"创建邮箱失败: {e}")
def get_verification_code(
self,
email: str,
email_id: str = None,
timeout: int = 120,
pattern: str = OTP_CODE_PATTERN,
otp_sent_at: Optional[float] = None,
) -> Optional[str]:
"""
从 Freemail 邮箱获取验证码
Args:
email: 邮箱地址
email_id: 未使用，保留接口兼容
timeout: 超时时间（秒）
pattern: 验证码正则
otp_sent_at: OTP 发送时间戳（暂未使用）
Returns:
验证码字符串，超时返回 None
"""
logger.info(f"正在从 Freemail 邮箱 {email} 获取验证码...")
start_time = time.time()
seen_mail_ids: set = set()
while time.time() - start_time < timeout:
try:
mails = self._make_request("GET", "/api/emails", params={"mailbox": email, "limit": 20})
if not isinstance(mails, list):
time.sleep(3)
continue
for mail in mails:
mail_id = mail.get("id")
if not mail_id or mail_id in seen_mail_ids:
continue
seen_mail_ids.add(mail_id)
sender = str(mail.get("sender", "")).lower()
subject = str(mail.get("subject", ""))
preview = str(mail.get("preview", ""))
content = f"{sender}\n{subject}\n{preview}"
if "openai" not in content.lower():
continue
# 尝试直接使用 Freemail 提取的验证码
v_code = mail.get("verification_code")
if v_code:
logger.info(f"从 Freemail 邮箱 {email} 找到验证码: {v_code}")
self.update_status(True)
return v_code
# 如果没有直接提供，通过正则匹配 preview
match = re.search(pattern, content)
if match:
code = match.group(1)
logger.info(f"从 Freemail 邮箱 {email} 找到验证码: {code}")
self.update_status(True)
return code
# 如果依然未找到，获取邮件详情进行匹配
try:
detail = self._make_request("GET", f"/api/email/{mail_id}")
full_content = str(detail.get("content", "")) + "\n" + str(detail.get("html_content", ""))
match = re.search(pattern, full_content)
if match:
code = match.group(1)
logger.info(f"从 Freemail 邮箱 {email} 找到验证码: {code}")
self.update_status(True)
return code
except Exception as e:
logger.debug(f"获取 Freemail 邮件详情失败: {e}")
except Exception as e:
logger.debug(f"检查 Freemail 邮件时出错: {e}")
time.sleep(3)
logger.warning(f"等待 Freemail 验证码超时: {email}")
return None
def list_emails(self, **kwargs) -> List[Dict[str, Any]]:
"""
列出邮箱
Args:
**kwargs: 额外查询参数
Returns:
邮箱列表
"""
try:
params = {
"limit": kwargs.get("limit", 100),
"offset": kwargs.get("offset", 0)
}
resp = self._make_request("GET", "/api/mailboxes", params=params)
emails = []
if isinstance(resp, list):
for mail in resp:
address = mail.get("address")
if address:
emails.append({
"id": address,
"service_id": address,
"email": address,
"created_at": mail.get("created_at"),
"raw_data": mail
})
self.update_status(True)
return emails
except Exception as e:
logger.warning(f"列出 Freemail 邮箱失败: {e}")
self.update_status(False, e)
return []
def delete_email(self, email_id: str) -> bool:
"""
删除邮箱
"""
try:
self._make_request("DELETE", "/api/mailboxes", params={"address": email_id})
logger.info(f"已删除 Freemail 邮箱: {email_id}")
self.update_status(True)
return True
except Exception as e:
logger.warning(f"删除 Freemail 邮箱失败: {e}")
self.update_status(False, e)
return False
def check_health(self) -> bool:
"""检查服务健康状态"""
try:
self._make_request("GET", "/api/domains")
self.update_status(True)
return True
except Exception as e:
logger.warning(f"Freemail 健康检查失败: {e}")
self.update_status(False, e)
return False

15
src/web/routes/email.py
View File

@@ -84,7 +84,7 @@ class OutlookBatchImportResponse(BaseModel):
# ============== Helper Functions ==============
# 敏感字段列表，返回响应时需要过滤
SENSITIVE_FIELDS = {'password', 'api_key', 'refresh_token', 'access_token'}
SENSITIVE_FIELDS = {'password', 'api_key', 'refresh_token', 'access_token', 'admin_token'}
def filter_sensitive_config(config: Dict[str, Any]) -> Dict[str, Any]:
"""过滤敏感配置信息"""
@@ -145,6 +145,7 @@ async def get_email_services_stats():
'custom_count': 0,
'temp_mail_count': 0,
'duck_mail_count': 0,
'freemail_count': 0,
'tempmail_available': True, # 临时邮箱始终可用
'enabled_count': enabled_count
}
@@ -158,6 +159,8 @@ async def get_email_services_stats():
stats['temp_mail_count'] = count
elif service_type == 'duck_mail':
stats['duck_mail_count'] = count
elif service_type == 'freemail':
stats['freemail_count'] = count
return stats
@@ -218,6 +221,16 @@ async def get_service_types():
{"name": "api_key", "label": "API Key", "required": False, "secret": True},
{"name": "password_length", "label": "随机密码长度", "required": False, "default": 12},
]
},
{
"value": "freemail",
"label": "Freemail",
"description": "Freemail 自部署 Cloudflare Worker 临时邮箱服务",
"config_fields": [
{"name": "base_url", "label": "API 地址", "required": True, "placeholder": "https://freemail.example.com"},
{"name": "admin_token", "label": "Admin Token", "required": True, "secret": True},
{"name": "domain", "label": "邮箱域名", "required": False, "placeholder": "example.com"},
]
}
]
}

39
src/web/routes/registration.py
View File

@@ -208,7 +208,7 @@ def _normalize_email_service_config(
if service_type == EmailServiceType.CUSTOM_DOMAIN:
if 'domain' in normalized and 'default_domain' not in normalized:
normalized['default_domain'] = normalized.pop('domain')
elif service_type == EmailServiceType.TEMP_MAIL:
elif service_type in (EmailServiceType.TEMP_MAIL, EmailServiceType.FREEMAIL):
if 'default_domain' in normalized and 'domain' not in normalized:
normalized['domain'] = normalized.pop('default_domain')
elif service_type == EmailServiceType.DUCK_MAIL:
@@ -358,6 +358,20 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
logger.info(f"使用数据库 DuckMail 服务: {db_service.name}")
else:
raise ValueError("没有可用的 DuckMail 邮箱服务，请先在邮箱服务页面添加服务")
elif service_type == EmailServiceType.FREEMAIL:
from ...database.models import EmailService as EmailServiceModel
db_service = db.query(EmailServiceModel).filter(
EmailServiceModel.service_type == "freemail",
EmailServiceModel.enabled == True
).order_by(EmailServiceModel.priority.asc()).first()
if db_service and db_service.config:
config = _normalize_email_service_config(service_type, db_service.config, actual_proxy_url)
crud.update_registration_task(db, task_uuid, email_service_id=db_service.id)
logger.info(f"使用数据库 Freemail 服务: {db_service.name}")
else:
raise ValueError("没有可用的 Freemail 邮箱服务，请先在邮箱服务页面添加服务")
else:
config = email_service_config or {}
@@ -1091,6 +1105,11 @@ async def get_available_email_services():
"available": False,
"count": 0,
"services": []
},
"freemail": {
"available": False,
"count": 0,
"services": []
}
}
@@ -1182,6 +1201,24 @@ async def get_available_email_services():
result["duck_mail"]["count"] = len(duck_mail_services)
result["duck_mail"]["available"] = len(duck_mail_services) > 0
freemail_services = db.query(EmailServiceModel).filter(
EmailServiceModel.service_type == "freemail",
EmailServiceModel.enabled == True
).order_by(EmailServiceModel.priority.asc()).all()
for service in freemail_services:
config = service.config or {}
result["freemail"]["services"].append({
"id": service.id,
"name": service.name,
"type": "freemail",
"domain": config.get("domain"),
"priority": service.priority
})
result["freemail"]["count"] = len(freemail_services)
result["freemail"]["available"] = len(freemail_services) > 0
return result