mygo-kyx/codexTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(telegram_bot): Improve Cloud Mail domain management and API response handling

Browse Source 
- Extract message object using get_message() helper in _cloudmail_add_domain and _cloudmail_del_domain methods for consistency
- Replace direct update.message.reply_text() calls with message.reply_text() throughout domain management functions
- Add comprehensive HTTP status code validation in API connection check
- Implement empty response detection to handle edge cases where API returns blank content
- Add JSON parsing error handling with fallback error message for non-JSON responses
- Improve error messaging to include HTTP status codes and API error codes for better debugging
- Enhance robustness of email service API validation with multiple safety checks
This commit is contained in:
kyx236
2026-01-30 15:30:16 +08:00
parent f39dff8ee6
commit 1c17015669
1 changed files with 24 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
telegram_bot.py
View File

@@ -3077,13 +3077,14 @@ class ProvisionerBot:
async def _cloudmail_add_domain(self, update: Update, domain: str):
"""添加 Cloud Mail 域名"""
message = get_message(update)
try:
import tomli_w
from config import CONFIG_FILE, EMAIL_DOMAINS
import tomllib
if domain in EMAIL_DOMAINS:
await update.message.reply_text(f"⚠️ 域名 {domain} 已存在")
await message.reply_text(f"⚠️ 域名 {domain} 已存在")
return
# 读取配置
@@ -3107,21 +3108,22 @@ class ProvisionerBot:
from config import reload_config
reload_config()
await update.message.reply_text(f"✅ 已添加域名: {domain}")
await message.reply_text(f"✅ 已添加域名: {domain}")
except ImportError:
await update.message.reply_text("❌ 缺少 tomli_w 依赖\n请运行: uv add tomli_w")
await message.reply_text("❌ 缺少 tomli_w 依赖\n请运行: uv add tomli_w")
except Exception as e:
await update.message.reply_text(f"❌ 添加失败: {e}")
await message.reply_text(f"❌ 添加失败: {e}")
async def _cloudmail_del_domain(self, update: Update, domain: str):
"""删除 Cloud Mail 域名"""
message = get_message(update)
try:
import tomli_w
from config import CONFIG_FILE, EMAIL_DOMAINS
import tomllib
if domain not in EMAIL_DOMAINS:
await update.message.reply_text(f"⚠️ 域名 {domain} 不存在")
await message.reply_text(f"⚠️ 域名 {domain} 不存在")
return
# 读取配置
@@ -3139,11 +3141,11 @@ class ProvisionerBot:
from config import reload_config
reload_config()
await update.message.reply_text(f"✅ 已删除域名: {domain}")
await message.reply_text(f"✅ 已删除域名: {domain}")
except ImportError:
await update.message.reply_text("❌ 缺少 tomli_w 依赖\n请运行: uv add tomli_w")
await message.reply_text("❌ 缺少 tomli_w 依赖\n请运行: uv add tomli_w")
except Exception as e:
await update.message.reply_text(f"❌ 删除失败: {e}")
await message.reply_text(f"❌ 删除失败: {e}")
@admin_only
async def cmd_test_email(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
@@ -4516,12 +4518,24 @@ class ProvisionerBot:
}
response = requests.post(url, headers=headers, json=payload, timeout=10)
data = response.json()
# 检查响应状态码
if response.status_code != 200:
return False, f"HTTP {response.status_code}: {response.text[:100]}"
# 检查响应内容
if not response.text or not response.text.strip():
return False, "API 返回空响应"
try:
data = response.json()
except Exception:
return False, f"API 返回非 JSON 格式: {response.text[:100]}"
if data.get("code") == 200:
return True, "API 连接正常"
else:
return False, data.get("message", "未知错误")
return False, data.get("message", f"错误码: {data.get('code')}")
return False, "未知的邮箱服务类型"