Rhinox13/chatapi
0
1from __future__ import annotations
2
3import json
4import smtplib
5from email.mime.multipart import MIMEMultipart
6from email.mime.text import MIMEText
7from logging import Logger
8
9from flask import has_request_context, request as flask_request
10
11from ..core import settings
12
13EMAIL_PROVIDER_SMTP = "smtp"
14EMAIL_PROVIDER_RESEND = "resend"
15EMAIL_PROVIDER_BREVO = "brevo"
16EMAIL_PROVIDER_TENCENTCLOUD = "tencentcloud"
17
18_TENCENTCLOUD_SES_DEFAULT_REGION = "ap-guangzhou"
19
20
21def get_available_email_providers() -> list[dict[str, str]]:
22 providers: list[dict[str, str]] = []
23 if settings.smtp_host:
24 providers.append({"value": EMAIL_PROVIDER_SMTP, "label": "SMTP"})
25 if settings.resend_api_key.strip():
26 providers.append({"value": EMAIL_PROVIDER_RESEND, "label": "Resend"})
27 if settings.brevo_api_key.strip():
28 providers.append({"value": EMAIL_PROVIDER_BREVO, "label": "Brevo"})
29 if (
30 settings.tencentcloud_secret_id.strip()
31 and settings.tencentcloud_secret_key.strip()
32 and settings.email_from.strip()
33 and settings.tencentcloud_template_id.strip()
34 ):
35 providers.append({"value": EMAIL_PROVIDER_TENCENTCLOUD, "label": "腾讯云 SES"})
36 return providers
37
38
39def resolve_email_provider(provider: str, available_providers: list[dict[str, str]] | None = None) -> str:
40 options = available_providers if available_providers is not None else get_available_email_providers()
41 selected = provider.strip().lower()
42 if selected and any(option["value"] == selected for option in options):
43 return selected
44 return options[0]["value"] if options else ""
45
46
47def _build_smtp_connection(logger: Logger) -> tuple[smtplib.SMTP | smtplib.SMTP_SSL, str] | tuple[None, str]:
48 host = settings.smtp_host
49 port = settings.smtp_port
50 username = settings.smtp_username
51 password = settings.smtp_password
52 use_tls = settings.smtp_use_tls
53
54 if not host:
55 return None, "SMTP_HOST 未配置"
56
57 try:
58 use_ssl = (port == 465)
59 if use_ssl:
60 server: smtplib.SMTP | smtplib.SMTP_SSL = smtplib.SMTP_SSL(host, port, timeout=10)
61 else:
62 server = smtplib.SMTP(host, port, timeout=10)
63
64 if not use_ssl and use_tls:
65 server.starttls()
66 server.ehlo()
67
68 if username and password:
69 server.login(username, password)
70
71 return server, ""
72 except smtplib.SMTPAuthenticationError:
73 return None, "SMTP 认证失败，请检查用户名和密码"
74 except smtplib.SMTPConnectError:
75 return None, f"无法连接到 SMTP 服务器 {host}:{port}"
76 except smtplib.SMTPException as exc:
77 return None, f"SMTP 错误: {exc}"
78 except Exception as exc:
79 logger.exception("[SMTP] connection failed")
80 return None, f"连接失败: {exc}"
81
82
83def _send_smtp_email(to: str, subject: str, body: str, *, logger: Logger) -> tuple[bool, str]:
84 server, err = _build_smtp_connection(logger)
85 if server is None:
86 return False, err
87
88 from_addr = _default_email_from() or settings.smtp_username
89
90 try:
91 msg = MIMEMultipart()
92 msg["From"] = from_addr
93 msg["To"] = to
94 msg["Subject"] = subject
95 msg.attach(MIMEText(body, "plain", "utf-8"))
96 server.sendmail(from_addr, [to], msg.as_string())
97 return True, "邮件已发送"
98 except smtplib.SMTPException as exc:
99 return False, f"SMTP 错误: {exc}"
100 except Exception as exc:
101 logger.exception("[SMTP] send failed")
102 return False, f"发送失败: {exc}"
103 finally:
104 try:
105 server.quit()
106 except Exception:
107 pass
108
109
110def _read_error_message_from_body(raw: str, fallback_status: int | None = None) -> str:
111 if not raw.strip():
112 return f"HTTP {fallback_status}" if fallback_status is not None else "HTTP error"
113 try:
114 payload = json.loads(raw)
115 except Exception:
116 return raw.strip()
117 if isinstance(payload, dict):
118 response = payload.get("Response")
119 if isinstance(response, dict):
120 error = response.get("Error")
121 if isinstance(error, dict):
122 message = str(error.get("Message", "") or "").strip()
123 code = str(error.get("Code", "") or "").strip()
124 if message:
125 return f"{code}: {message}" if code else message
126 for key in ("error", "message", "detail"):
127 value = payload.get(key)
128 if isinstance(value, str) and value.strip():
129 return value.strip()
130 return raw.strip()
131
132
133def _truncate(text: str, limit: int = 1200) -> str:
134 text = text.strip()
135 if len(text) <= limit:
136 return text
137 return f"{text[:limit]}... [truncated {len(text) - limit} chars]"
138
139
140def _default_email_from() -> str:
141 configured_from = settings.email_from.strip()
142 if configured_from:
143 return configured_from
144
145 if has_request_context():
146 host = (
147 flask_request.headers.get("X-Forwarded-Host")
148 or flask_request.headers.get("Host")
149 or flask_request.host
150 or ""
151 ).split(",")[0].strip()
152 host = host.split(":", 1)[0].strip()
153 if host:
154 return f"noreply@{host}"
155 return "noreply@localhost"
156
157
158def _build_resend_message_payload(from_addr: str, to: str, subject: str, body: str) -> dict[str, str]:
159 return {
160 "from": from_addr,
161 "to": to,
162 "subject": subject,
163 "html": body,
164 }
165
166
167def _send_resend_email(to: str, subject: str, body: str, *, logger: Logger) -> tuple[bool, str]:
168 try:
169 import resend
170 except Exception as exc:
171 logger.exception("[Resend] import failed")
172 return False, f"Resend SDK 未安装或导入失败: {exc}"
173
174 api_key = settings.resend_api_key.strip()
175 if not api_key:
176 return False, "RESEND_API_KEY 未配置"
177
178 from_addr = _default_email_from()
179 logger.info(
180 "[Resend] sending email to=%s from=%s subject=%s",
181 to,
182 from_addr,
183 subject,
184 )
185 resend.api_key = api_key
186 payload = _build_resend_message_payload(from_addr, to, subject, body)
187 try:
188 result = resend.Emails.send(payload)
189 logger.info("[Resend] success response=%s", _truncate(repr(result), 4000))
190 return True, "邮件已发送"
191 except Exception as exc:
192 response = getattr(exc, "response", None)
193 if response is not None:
194 response_body = getattr(response, "text", "") or getattr(response, "body", "") or ""
195 status_code = getattr(response, "status_code", None) or getattr(response, "status", None)
196 request_id = ""
197 headers = getattr(response, "headers", None)
198 if headers is not None:
199 request_id = headers.get("x-request-id") or headers.get("x-resend-id") or ""
200 logger.warning(
201 "[Resend] http_error status=%s request_id=%s body=%s",
202 status_code,
203 request_id,
204 _truncate(str(response_body), 4000),
205 )
206 return False, f"Resend 错误: {_read_error_message_from_body(str(response_body), status_code)}"
207 logger.exception("[Resend] send failed")
208 return False, f"发送失败: {exc}"
209
210
211def _send_brevo_email(to: str, subject: str, body: str, *, logger: Logger) -> tuple[bool, str]:
212 try:
213 import sib_api_v3_sdk
214 from sib_api_v3_sdk.rest import ApiException
215 except Exception as exc:
216 logger.exception("[Brevo] import failed")
217 return False, f"Brevo SDK 未安装或导入失败: {exc}"
218
219 api_key = settings.brevo_api_key.strip()
220 if not api_key:
221 return False, "BREVO_API_KEY 未配置"
222
223 from_email = _default_email_from()
224 from_name = settings.brevo_from_name.strip() or "ChatAPI"
225 logger.info(
226 "[Brevo] sending email to=%s from=%s sender_name=%s subject=%s",
227 to,
228 from_email,
229 from_name,
230 subject,
231 )
232
233 configuration = sib_api_v3_sdk.Configuration()
234 configuration.api_key["api-key"] = api_key
235 api_instance = sib_api_v3_sdk.TransactionalEmailsApi(sib_api_v3_sdk.ApiClient(configuration))
236
237 sender = sib_api_v3_sdk.SendSmtpEmailSender(name=from_name, email=from_email)
238 recipient = sib_api_v3_sdk.SendSmtpEmailTo(email=to)
239 email = sib_api_v3_sdk.SendSmtpEmail(
240 sender=sender,
241 to=[recipient],
242 subject=subject,
243 html_content=body,
244 )
245
246 try:
247 result = api_instance.send_transac_email(email)
248 logger.info("[Brevo] success response=%s", _truncate(repr(result), 4000))
249 return True, "邮件已发送"
250 except ApiException as exc:
251 raw_body = getattr(exc, "body", "") or ""
252 logger.warning(
253 "[Brevo] api_error status=%s reason=%s body=%s",
254 getattr(exc, "status", ""),
255 getattr(exc, "reason", ""),
256 _truncate(str(raw_body), 4000),
257 )
258 return False, f"Brevo 错误: {_read_error_message_from_body(str(raw_body), getattr(exc, 'status', None))}"
259 except Exception as exc:
260 logger.exception("[Brevo] send failed")
261 return False, f"发送失败: {exc}"
262
263
264def _build_tencentcloud_template_data(
265 *,
266 action: str,
267 code: str = "",
268) -> str:
269 return json.dumps(
270 {
271 "action": action,
272 "code": code,
273 },
274 ensure_ascii=False,
275 separators=(",", ":"),
276 )
277
278
279def _send_tencentcloud_email(
280 to: str,
281 subject: str,
282 text_body: str,
283 *,
284 html_body: str | None,
285 trigger_type: int,
286 action: str = "",
287 code: str = "",
288 logger: Logger,
289) -> tuple[bool, str]:
290 secret_id = settings.tencentcloud_secret_id.strip()
291 secret_key = settings.tencentcloud_secret_key.strip()
292 if not secret_id or not secret_key:
293 return False, "TENCENTCLOUD_SECRET_ID / TENCENTCLOUD_SECRET_KEY 未配置"
294
295 from_addr = settings.email_from.strip()
296 if not from_addr:
297 return False, "CHATAPI_EMAIL_FROM 未配置"
298
299 template_id_raw = settings.tencentcloud_template_id.strip()
300 if not template_id_raw:
301 return False, "CHATAPI_TENCENTCLOUD_TEMPLATE_ID 未配置"
302 try:
303 template_id = int(template_id_raw)
304 except ValueError:
305 return False, "CHATAPI_TENCENTCLOUD_TEMPLATE_ID 必须是整数"
306
307 region = settings.tencentcloud_ses_region.strip() or _TENCENTCLOUD_SES_DEFAULT_REGION
308 template_data = _build_tencentcloud_template_data(
309 action=action,
310 code=code,
311 )
312
313 payload: dict[str, object] = {
314 "FromEmailAddress": from_addr,
315 "Destination": [to],
316 "Subject": subject,
317 "ReplyToAddresses": from_addr,
318 "HeaderFrom": from_addr,
319 "TriggerType": trigger_type,
320 "Template": {
321 "TemplateID": template_id,
322 "TemplateData": template_data,
323 },
324 }
325 logger.info(
326 "[TencentCloud SES] sending email to=%s from=%s region=%s subject=%s",
327 to,
328 from_addr,
329 region,
330 subject,
331 )
332 try:
333 from tencentcloud.common import credential
334 from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
335 from tencentcloud.common.profile.client_profile import ClientProfile
336 from tencentcloud.common.profile.http_profile import HttpProfile
337 from tencentcloud.ses.v20201002 import models, ses_client
338 except Exception as exc:
339 logger.exception("[TencentCloud SES] import failed")
340 return False, f"TencentCloud SDK 未安装或导入失败: {exc}"
341
342 try:
343 cred = credential.Credential(secret_id, secret_key)
344 http_profile = HttpProfile()
345 http_profile.endpoint = f"ses.{region}.tencentcloudapi.com"
346 client_profile = ClientProfile()
347 client_profile.httpProfile = http_profile
348 client_profile.signMethod = "TC3-HMAC-SHA256"
349 client = ses_client.SesClient(cred, region, client_profile)
350
351 req = models.SendEmailRequest()
352 req.from_json_string(json.dumps(payload, ensure_ascii=False))
353 response = client.SendEmail(req)
354 response_text = response.to_json_string()
355 logger.info("[TencentCloud SES] success response=%s", _truncate(response_text, 4000))
356 return True, "邮件已发送"
357 except TencentCloudSDKException as exc:
358 request_id = str(getattr(exc, "request_id", "") or getattr(exc, "requestId", "") or "").strip()
359 code = str(getattr(exc, "code", "") or "").strip()
360 message = str(getattr(exc, "message", "") or "").strip() or str(exc)
361 logger.warning(
362 "[TencentCloud SES] api_error code=%s request_id=%s message=%s",
363 code,
364 request_id,
365 _truncate(message, 4000),
366 )
367 if code and message:
368 return False, f"腾讯云 SES 错误: {code}: {message}"
369 return False, f"腾讯云 SES 错误: {message}"
370 except Exception as exc:
371 logger.exception("[TencentCloud SES] send failed")
372 return False, f"发送失败: {exc}"
373
374
375def _send_email(
376 provider: str,
377 to: str,
378 subject: str,
379 text_body: str,
380 *,
381 html_body: str | None = None,
382 trigger_type: int = 1,
383 action: str = "",
384 code: str = "",
385 logger: Logger,
386) -> tuple[bool, str]:
387 selected_provider = resolve_email_provider(provider)
388 if selected_provider == EMAIL_PROVIDER_RESEND:
389 return _send_resend_email(to, subject, html_body or text_body, logger=logger)
390 if selected_provider == EMAIL_PROVIDER_BREVO:
391 return _send_brevo_email(to, subject, html_body or text_body, logger=logger)
392 if selected_provider == EMAIL_PROVIDER_TENCENTCLOUD:
393 return _send_tencentcloud_email(
394 to,
395 subject,
396 text_body,
397 html_body=html_body,
398 trigger_type=trigger_type,
399 action=action,
400 code=code,
401 logger=logger,
402 )
403 if selected_provider == EMAIL_PROVIDER_SMTP:
404 return _send_smtp_email(to, subject, text_body, logger=logger)
405 return False, "未配置可用的邮箱发送方式"
406
407
408def send_test_email(to: str, *, provider: str = "", logger: Logger) -> tuple[bool, str]:
409 body = "这是一封来自 ChatAPI 的测试邮件，说明邮箱发送配置正确。"
410 html = "<p>这是一封来自 ChatAPI 的测试邮件，说明邮箱发送配置正确。</p>"
411 selected_provider = resolve_email_provider(provider)
412 return _send_email(
413 selected_provider,
414 to,
415 "ChatAPI 测试邮件",
416 body,
417 html_body=html,
418 trigger_type=0,
419 action="测试",
420 logger=logger,
421 )
422
423
424def send_verification_email(to: str, code: str, *, provider: str = "", logger: Logger) -> tuple[bool, str]:
425 subject = "ChatAPI 邮箱验证码"
426 body = f"您的验证码是：{code}\n\n验证码 5 分钟内有效，请勿泄露给他人。"
427 html = f"<p>您的验证码是：<strong>{code}</strong></p><p>验证码 5 分钟内有效，请勿泄露给他人。</p>"
428 selected_provider = resolve_email_provider(provider)
429 return _send_email(
430 selected_provider,
431 to,
432 subject,
433 body,
434 html_body=html,
435 trigger_type=1,
436 action="注册",
437 code=code,
438 logger=logger,
439 )
440