mmc/openai_pool_orchestrator-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: support custom DuckMail domains

Browse Source
This commit is contained in:
root
2026-03-18 22:43:16 +08:00
parent b38cb7b27c
commit d31f62891e
2 changed files with 21 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
openai_pool_orchestrator/mail_providers.py
View File

@@ -398,9 +398,10 @@ class MoeMailProvider(MailProvider):
# ==================== DuckMail ====================
class DuckMailProvider(MailProvider):
def __init__(self, api_base: str = "https://api.duckmail.sbs", bearer_token: str = ""):
def __init__(self, api_base: str = "https://api.duckmail.sbs", bearer_token: str = "", domain: str = ""):
self.api_base = api_base.rstrip("/")
self.bearer_token = bearer_token
self.domain = str(domain).strip()
def _auth_headers(self, token: str = "") -> Dict[str, str]:
h: Dict[str, str] = {"Accept": "application/json"}
@@ -419,17 +420,19 @@ class DuckMailProvider(MailProvider):
headers["Authorization"] = f"Bearer {self.bearer_token}"
try:
domains_resp = session.get(f"{self.api_base}/domains", headers={"Accept": "application/json"}, timeout=15, verify=False)
if domains_resp.status_code != 200:
return "", ""
data = domains_resp.json()
items = data if isinstance(data, list) else (data.get("hydra:member") or [])
domains = [str(i.get("domain") or "") for i in items if isinstance(i, dict) and i.get("domain") and i.get("isActive", True)]
if not domains:
return "", ""
# 优先使用 duckmail.sbs 主域名，避免临时域名被 OpenAI 封禁
_preferred = [d for d in domains if "duckmail" in d.lower()]
domain = random.choice(_preferred) if _preferred else random.choice(domains)
domain = self.domain
if not domain:
domains_resp = session.get(f"{self.api_base}/domains", headers={"Accept": "application/json"}, timeout=15, verify=False)
if domains_resp.status_code != 200:
return "", ""
data = domains_resp.json()
items = data if isinstance(data, list) else (data.get("hydra:member") or [])
domains = [str(i.get("domain") or "") for i in items if isinstance(i, dict) and i.get("domain") and i.get("isActive", True)]
if not domains:
return "", ""
# 优先使用 duckmail.sbs 主域名，避免临时域名被 OpenAI 封禁
_preferred = [d for d in domains if "duckmail" in d.lower()]
domain = random.choice(_preferred) if _preferred else random.choice(domains)
local = f"oc{secrets.token_hex(5)}"
email = f"{local}@{domain}"
@@ -793,6 +796,7 @@ def create_provider_by_name(provider_type: str, mail_cfg: Dict[str, Any]) -> Mai
return DuckMailProvider(
api_base=api_base or "https://api.duckmail.sbs",
bearer_token=str(mail_cfg.get("bearer_token", "")).strip(),
domain=str(mail_cfg.get("domain", "")).strip(),
)
elif provider_type == "cloudflare_temp_email":
return CloudflareTempEmailProvider(

5
openai_pool_orchestrator/static/index.html
View File

@@ -632,6 +632,11 @@
<input type="text" data-key="api_base" placeholder="https://api.duckmail.sbs" autocomplete="off"
spellcheck="false" />
</div>
<div class="config-field">
<label>分配域名</label>
<input type="text" data-key="domain" placeholder="example.com" autocomplete="off"
spellcheck="false" />
</div>
<div class="config-field">
<label>Bearer Token</label>
<input type="password" data-key="bearer_token" placeholder="DuckMail Bearer Token"