feat(config,email_service): Add Cloud Mail API path auto-completion utility
- Add get_cloudmail_api_base() helper function to automatically append /api/public path to EMAIL_API_BASE - Update create_email_user() to use get_cloudmail_api_base() for consistent API endpoint construction - Update get_verification_code() to use get_cloudmail_api_base() for email list retrieval - Update fetch_email_content() to use get_cloudmail_api_base() for email list retrieval - Refactor telegram_bot.py to use centralized path completion logic instead of inline implementation - Improve API endpoint consistency across email service operations and reduce code duplication
This commit is contained in:
13
config.py
13
config.py
@@ -418,7 +418,7 @@ def reload_config() -> dict:
|
||||
# 邮箱系统选择
|
||||
EMAIL_PROVIDER = _cfg.get("email_provider", "kyx") # "kyx" 或 "gptmail"
|
||||
|
||||
# 原有邮箱系统 (KYX)
|
||||
# 原有邮箱系统 (KYX / Cloud Mail)
|
||||
_email = _cfg.get("email", {})
|
||||
EMAIL_API_BASE = _email.get("api_base", "")
|
||||
EMAIL_API_AUTH = _email.get("api_auth", "")
|
||||
@@ -427,6 +427,17 @@ EMAIL_DOMAIN = EMAIL_DOMAINS[0] if EMAIL_DOMAINS else ""
|
||||
EMAIL_ROLE = _email.get("role", "gpt-team")
|
||||
EMAIL_WEB_URL = _email.get("web_url", "")
|
||||
|
||||
|
||||
def get_cloudmail_api_base() -> str:
|
||||
"""获取 Cloud Mail API 地址,自动补全 /api/public 路径"""
|
||||
if not EMAIL_API_BASE:
|
||||
return ""
|
||||
api_base = EMAIL_API_BASE.rstrip("/")
|
||||
if not api_base.endswith("/api/public"):
|
||||
api_base = f"{api_base}/api/public"
|
||||
return api_base
|
||||
|
||||
|
||||
# GPTMail 临时邮箱配置
|
||||
_gptmail = _cfg.get("gptmail", {})
|
||||
GPTMAIL_API_BASE = _gptmail.get("api_base", "https://mail.chatgpt.org.uk")
|
||||
|
||||
@@ -27,6 +27,7 @@ from config import (
|
||||
get_random_gptmail_domain,
|
||||
get_next_gptmail_key,
|
||||
get_gptmail_keys,
|
||||
get_cloudmail_api_base,
|
||||
)
|
||||
from logger import log
|
||||
|
||||
@@ -446,7 +447,7 @@ def create_email_user(email: str, password: str = None, role_name: str = None) -
|
||||
if role_name is None:
|
||||
role_name = EMAIL_ROLE
|
||||
|
||||
url = f"{EMAIL_API_BASE}/addUser"
|
||||
url = f"{get_cloudmail_api_base()}/addUser"
|
||||
headers = {
|
||||
"Authorization": EMAIL_API_AUTH,
|
||||
"Content-Type": "application/json"
|
||||
@@ -484,7 +485,7 @@ def get_verification_code(email: str, max_retries: int = None, interval: int = N
|
||||
Returns:
|
||||
tuple: (code, error, email_time) - 验证码、错误信息、邮件时间
|
||||
"""
|
||||
url = f"{EMAIL_API_BASE}/emailList"
|
||||
url = f"{get_cloudmail_api_base()}/emailList"
|
||||
headers = {
|
||||
"Authorization": EMAIL_API_AUTH,
|
||||
"Content-Type": "application/json"
|
||||
@@ -565,7 +566,7 @@ def fetch_email_content(email: str) -> list:
|
||||
Returns:
|
||||
list: 邮件列表
|
||||
"""
|
||||
url = f"{EMAIL_API_BASE}/emailList"
|
||||
url = f"{get_cloudmail_api_base()}/emailList"
|
||||
headers = {
|
||||
"Authorization": EMAIL_API_AUTH,
|
||||
"Content-Type": "application/json"
|
||||
|
||||
@@ -4506,7 +4506,12 @@ class ProvisionerBot:
|
||||
if not EMAIL_DOMAINS:
|
||||
return False, "未配置 email.domains"
|
||||
|
||||
url = f"{EMAIL_API_BASE}/emailList"
|
||||
# 自动补全 /api/public 路径
|
||||
api_base = EMAIL_API_BASE.rstrip("/")
|
||||
if not api_base.endswith("/api/public"):
|
||||
api_base = f"{api_base}/api/public"
|
||||
|
||||
url = f"{api_base}/emailList"
|
||||
headers = {
|
||||
"Authorization": EMAIL_API_AUTH,
|
||||
"Content-Type": "application/json"
|
||||
|
||||
Reference in New Issue
Block a user