diff --git a/backend/package/yuxi/channel/extensions/email_smtp/__init__.py b/backend/package/yuxi/channel/extensions/email_smtp/__init__.py new file mode 100644 index 00000000..a6ed15c2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/__init__.py @@ -0,0 +1,4 @@ +from yuxi.channel.extensions.email_smtp.plugin import EmailSmtpPlugin +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +ChannelPluginRegistry.register(EmailSmtpPlugin()) diff --git a/backend/package/yuxi/channel/extensions/email_smtp/attachment.py b/backend/package/yuxi/channel/extensions/email_smtp/attachment.py new file mode 100644 index 00000000..3bd7f940 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/attachment.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import os + +ALLOWED_MIME_TYPES = { + "application/pdf", + "text/plain", + "text/csv", + "image/jpeg", + "image/png", + "image/gif", + "image/webp", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/msword", + "application/vnd.ms-excel", +} + +BLOCKED_EXTENSIONS = { + ".exe", + ".dll", + ".bat", + ".cmd", + ".ps1", + ".vbs", + ".js", + ".scr", + ".pif", + ".msi", + ".com", +} + +MAX_TOTAL_SIZE = 25 * 1024 * 1024 +MAX_PER_ATTACHMENT = 10 * 1024 * 1024 +MAX_TEXT_FOR_AI = 200 * 1024 + + +def is_attachment_safe(filename: str, content_type: str, size: int) -> tuple[bool, str]: + ext = os.path.splitext(filename)[1].lower() + if ext in BLOCKED_EXTENSIONS: + return False, f"禁止的文件类型: {ext}" + if size > MAX_PER_ATTACHMENT: + return False, f"附件过大 ({size} bytes > {MAX_PER_ATTACHMENT})" + if content_type not in ALLOWED_MIME_TYPES: + return False, f"不支持的 MIME 类型: {content_type}" + return True, "ok" \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/email_smtp/bounce_handler.py b/backend/package/yuxi/channel/extensions/email_smtp/bounce_handler.py new file mode 100644 index 00000000..2ccaccc8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/bounce_handler.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +import email +import logging +import re +from dataclasses import dataclass +from email import policy + +from .types import ParsedEmail + +logger = logging.getLogger(__name__) + +BOUNCE_INDICATORS = { + "auto-submitted": ["auto-generated", "auto-replied"], + "precedence": ["bulk", "auto_reply", "list"], + "x-failed-recipients": None, +} + +DSN_CONTENT_TYPES = { + "multipart/report", + "message/delivery-status", + "message/disposition-notification", +} + +BOUNCE_SUBJECT_PATTERNS = [ + re.compile(r"^\s*undeliver(ed|able)", re.IGNORECASE), + re.compile(r"^\s*delivery\s+(failure|failed|status|report)", re.IGNORECASE), + re.compile(r"^\s*returned\s+mail", re.IGNORECASE), + re.compile(r"^\s*bounce", re.IGNORECASE), + re.compile(r"^\s*mail\s+delivery\s+(failure|failed)", re.IGNORECASE), + re.compile(r"^\s*failure\s+notice", re.IGNORECASE), + re.compile(r"^\s*mail\s+system\s+error", re.IGNORECASE), +] + +BOUNCE_FROM_PATTERNS = [ + re.compile(r"mailer-daemon", re.IGNORECASE), + re.compile(r"postmaster", re.IGNORECASE), + re.compile(r"mail\s+delivery\s+subsystem", re.IGNORECASE), + re.compile(r"\bdsn\b", re.IGNORECASE), +] + +SMTP_STATUS_CODES = { + "4.0.0": ("temporary_failure", True), + "4.1.1": ("bad_destination_mailbox", True), + "4.1.2": ("bad_destination_system", True), + "4.1.8": ("bad_sender_address", True), + "4.2.0": ("mailbox_unavailable", True), + "4.2.2": ("mailbox_full", True), + "4.3.0": ("other_mail_system", True), + "4.3.1": ("mail_system_full", True), + "4.3.2": ("system_not_accepting", True), + "4.4.0": ("routing_error", True), + "4.4.1": ("no_answer_from_host", True), + "4.4.2": ("bad_connection", True), + "4.4.5": ("mail_loop", True), + "4.6.0": ("other_protocol_status", True), + "4.7.0": ("security_error", True), + "4.7.1": ("delivery_not_authorized", True), + "5.0.0": ("permanent_failure", False), + "5.1.0": ("bad_destination_address", False), + "5.1.1": ("bad_destination_mailbox", False), + "5.1.2": ("bad_destination_system", False), + "5.1.3": ("bad_destination_mailbox_syntax", False), + "5.1.4": ("destination_ambiguous", False), + "5.1.6": ("destination_moved", False), + "5.1.7": ("bad_sender_address_syntax", False), + "5.1.8": ("bad_sender_address", False), + "5.2.0": ("mailbox_unavailable", False), + "5.2.1": ("mailbox_disabled", False), + "5.2.2": ("mailbox_full", False), + "5.2.3": ("message_too_long", False), + "5.2.4": ("mailing_list_expansion_problem", False), + "5.3.0": ("other_mail_system", False), + "5.3.1": ("mail_system_full", False), + "5.3.2": ("system_not_accepting", False), + "5.3.3": ("system_unable_to_process", False), + "5.3.4": ("message_too_big", False), + "5.4.0": ("routing_error", False), + "5.4.1": ("no_answer_from_host", False), + "5.4.2": ("bad_connection", False), + "5.4.3": ("routing_server_failure", False), + "5.4.4": ("unable_to_route", False), + "5.4.6": ("routing_loop_detected", False), + "5.4.7": ("delivery_time_expired", False), + "5.5.0": ("other_protocol_status", False), + "5.5.1": ("invalid_command", False), + "5.5.2": ("syntax_error", False), + "5.5.3": ("too_many_recipients", False), + "5.5.4": ("invalid_command_arguments", False), + "5.5.5": ("wrong_protocol_version", False), + "5.6.0": ("other_media_error", False), + "5.6.1": ("media_not_supported", False), + "5.6.2": ("conversion_required_and_prohibited", False), + "5.6.3": ("conversion_required_but_not_supported", False), + "5.6.4": ("conversion_with_loss_performed", False), + "5.6.5": ("conversion_failed", False), + "5.7.0": ("security_error", False), + "5.7.1": ("delivery_not_authorized", False), + "5.7.2": ("list_expansion_prohibited", False), + "5.7.3": ("security_conversion_required", False), + "5.7.4": ("security_features_not_supported", False), + "5.7.5": ("cryptographic_failure", False), + "5.7.6": ("cryptographic_algorithm_not_supported", False), + "5.7.7": ("message_integrity_failure", False), + "5.7.8": ("trust_anchor_not_found", False), + "5.7.9": ("signature_error", False), +} + + +@dataclass +class BounceInfo: + is_bounce: bool = False + bounce_type: str = "" + original_recipient: str = "" + original_message_id: str = "" + status_code: str = "" + status_description: str = "" + retryable: bool = False + diagnostic_info: str = "" + bounce_reason: str = "" + + +class EmailBounceHandler: + def analyze(self, parsed_email: ParsedEmail) -> BounceInfo: + info = BounceInfo() + + if self._is_bounce_by_headers(parsed_email.raw_bytes): + info.is_bounce = True + info.bounce_type = "bounce" + info.bounce_reason = "bounce_headers_detected" + elif self._is_bounce_by_subject(parsed_email.subject): + info.is_bounce = True + info.bounce_type = "bounce" + info.bounce_reason = "bounce_subject_detected" + elif self._is_bounce_by_sender(parsed_email.from_email): + info.is_bounce = True + info.bounce_type = "bounce" + info.bounce_reason = "bounce_sender_detected" + + if info.is_bounce: + self._extract_dsn_info(parsed_email.raw_bytes, info) + logger.info( + "Bounce detected: type=%s recipient=%s code=%s retryable=%s reason=%s", + info.bounce_type, + info.original_recipient, + info.status_code, + info.retryable, + info.bounce_reason, + ) + + return info + + def _is_bounce_by_headers(self, raw_bytes: bytes) -> bool: + try: + msg = email.message_from_bytes(raw_bytes, policy=policy.default) + except Exception: + return False + + auto_submitted = str(msg.get("Auto-Submitted", "")).lower() + if auto_submitted in BOUNCE_INDICATORS["auto-submitted"]: + return True + + precedence = str(msg.get("Precedence", "")).lower() + if precedence in BOUNCE_INDICATORS["precedence"]: + return True + + if msg.get("X-Failed-Recipients"): + return True + + content_type = msg.get_content_type() + if content_type in DSN_CONTENT_TYPES: + return True + + return False + + def _is_bounce_by_subject(self, subject: str) -> bool: + if not subject: + return False + for pattern in BOUNCE_SUBJECT_PATTERNS: + if pattern.search(subject): + return True + return False + + def _is_bounce_by_sender(self, from_email: str) -> bool: + if not from_email: + return False + for pattern in BOUNCE_FROM_PATTERNS: + if pattern.search(from_email): + return True + return False + + def _extract_dsn_info(self, raw_bytes: bytes, info: BounceInfo) -> None: + try: + msg = email.message_from_bytes(raw_bytes, policy=policy.default) + except Exception: + return + + if msg.is_multipart(): + for part in msg.walk(): + content_type = part.get_content_type() + if content_type == "message/delivery-status": + self._parse_delivery_status(part, info) + elif content_type == "message/rfc822": + self._parse_original_message(part, info) + elif content_type == "text/plain": + payload = part.get_payload(decode=True) + if payload: + self._extract_recipient_from_text(payload.decode("utf-8", errors="replace"), info) + + if not info.original_message_id: + info.original_message_id = str(msg.get("In-Reply-To", "")).strip("<>") + + if not info.original_recipient: + failed_recipients = msg.get("X-Failed-Recipients") + if failed_recipients: + info.original_recipient = failed_recipients.split(",")[0].strip() + + if info.status_code: + desc, retryable = SMTP_STATUS_CODES.get(info.status_code, ("unknown", False)) + info.status_description = desc + info.retryable = retryable + + def _parse_delivery_status(self, part, info: BounceInfo) -> None: + payload = part.get_payload(decode=False) + if isinstance(payload, list): + for sub_msg in payload: + text = sub_msg.as_string() + self._parse_delivery_status_text(text, info) + elif payload: + text = payload.decode("utf-8", errors="replace") if isinstance(payload, bytes) else str(payload) + self._parse_delivery_status_text(text, info) + + def _parse_delivery_status_text(self, text: str, info: BounceInfo) -> None: + for line in text.splitlines(): + line = line.strip() + if line.lower().startswith("final-recipient:"): + match = re.search(r"rfc822;\s*(.+)", line, re.IGNORECASE) + if match: + info.original_recipient = match.group(1).strip() + elif line.lower().startswith("original-recipient:"): + match = re.search(r"rfc822;\s*(.+)", line, re.IGNORECASE) + if match: + info.original_recipient = match.group(1).strip() + elif line.lower().startswith("status:"): + info.status_code = line.split(":", 1)[1].strip() + elif line.lower().startswith("diagnostic-code:"): + info.diagnostic_info = line.split(":", 1)[1].strip() + elif line.lower().startswith("action:"): + action = line.split(":", 1)[1].strip().lower() + if action == "failed": + info.retryable = False + elif action in ("delayed", "transient"): + info.retryable = True + + def _parse_original_message(self, part, info: BounceInfo) -> None: + try: + original_msg = email.message_from_bytes(part.get_payload(decode=True) or b"", policy=policy.default) + info.original_message_id = str(original_msg.get("Message-ID", "")).strip("<>") + if not info.original_recipient: + to_header = str(original_msg.get("To", "")) + if to_header: + info.original_recipient = to_header.strip() + except Exception: + pass + + def _extract_recipient_from_text(self, text: str, info: BounceInfo) -> None: + if info.original_recipient: + return + patterns = [ + re.compile(r"[<\[]([^>@\s]+@[^>\]\s]+)[>\]]"), + re.compile(r"\b([a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})\b"), + ] + for pattern in patterns: + match = pattern.search(text) + if match: + info.original_recipient = match.group(1).strip() + return diff --git a/backend/package/yuxi/channel/extensions/email_smtp/charset_utils.py b/backend/package/yuxi/channel/extensions/email_smtp/charset_utils.py new file mode 100644 index 00000000..78f7f1f6 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/charset_utils.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import email.header + +from charset_normalizer import from_bytes + + +def decode_mime_header(raw: str) -> str: + if not raw: + return "" + parts = email.header.decode_header(raw) + result: list[str] = [] + for decoded_part, charset in parts: + if isinstance(decoded_part, bytes): + if charset: + try: + result.append(decoded_part.decode(charset)) + except (UnicodeDecodeError, LookupError): + detected = from_bytes(decoded_part).best() + result.append(str(detected) if detected else decoded_part.decode("utf-8", errors="replace")) + else: + detected = from_bytes(decoded_part).best() + result.append(str(detected) if detected else decoded_part.decode("utf-8", errors="replace")) + else: + result.append(decoded_part) + return "".join(result) + + +def safe_decode_payload(payload: bytes, declared_charset: str | None) -> str: + if not payload: + return "" + + if not declared_charset: + detected = from_bytes(payload).best() + return str(detected) if detected else payload.decode("utf-8", errors="replace") + + try: + return payload.decode(declared_charset) + except (UnicodeDecodeError, LookupError): + if declared_charset.lower() in ("gb2312", "gbk", "gb18030"): + for enc in ("gb18030", "gbk", "gb2312"): + try: + return payload.decode(enc) + except UnicodeDecodeError: + continue + + detected = from_bytes(payload).best() + return str(detected) if detected else payload.decode("utf-8", errors="replace") diff --git a/backend/package/yuxi/channel/extensions/email_smtp/config.py b/backend/package/yuxi/channel/extensions/email_smtp/config.py new file mode 100644 index 00000000..3c4afe70 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/config.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import logging +import os +import secrets +from typing import Any + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from .types import EmailAccount, ImapConfig, SmtpConfig + +logger = logging.getLogger(__name__) + + +class EmailCredentialVault: + def __init__(self, master_key_path: str = "saves/.secret/email_vault.key"): + self._master_key = self._load_or_create_key(master_key_path) + + def encrypt(self, plaintext: str) -> bytes: + nonce = secrets.token_bytes(12) + aesgcm = AESGCM(self._master_key) + ct = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None) + return nonce + ct + + def decrypt(self, ciphertext: bytes) -> str: + nonce, ct = ciphertext[:12], ciphertext[12:] + aesgcm = AESGCM(self._master_key) + return aesgcm.decrypt(nonce, ct, None).decode("utf-8") + + def _load_or_create_key(self, path: str) -> bytes: + if os.path.exists(path): + with open(path, "rb") as f: + return f.read() + key = AESGCM.generate_key(bit_length=256) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as f: + f.write(key) + return key + + +class EmailSmtpConfigAdapter: + def __init__(self): + self._vault = EmailCredentialVault() + + def list_account_ids(self, config: dict) -> list[str]: + accounts = config.get("accounts", {}) + if not accounts: + return ["default"] + return list(accounts.keys()) + + def resolve_account(self, account_id: str, config: dict | None = None) -> dict: + config = config or self._load_config() + accounts = config.get("accounts", {}) + acct = accounts.get(account_id, {}) + + email = acct.get("email_address", config.get("base_email_address", "")) + return { + "account_id": account_id, + "email_address": email, + "display_name": acct.get("display_name", config.get("base_display_name", "AI \u5ba2\u670d")), + "smtp_host": acct.get("smtp_host", config.get("base_smtp_host", "")), + "smtp_port": acct.get("smtp_port", config.get("base_smtp_port", 465)), + "smtp_use_tls": acct.get("smtp_use_tls", True), + "smtp_username": acct.get("smtp_username", email), + "smtp_password": self._resolve_secret_with_vault(acct, "smtp_password"), + "imap_host": acct.get("imap_host", config.get("base_imap_host", "")), + "imap_port": acct.get("imap_port", config.get("base_imap_port", 993)), + "imap_use_ssl": acct.get("imap_use_ssl", True), + "imap_username": acct.get("imap_username", email), + "imap_password": self._resolve_secret_with_vault(acct, "imap_password"), + "idle_timeout_secs": acct.get("idle_timeout_secs", 0), + "poll_fallback_secs": acct.get("poll_fallback_secs", 60), + "max_fetch_per_cycle": acct.get("max_fetch_per_cycle", 50), + "dkim_selector": acct.get("dkim_selector", ""), + "dkim_private_key": self._resolve_secret_with_vault(acct, "dkim_private_key"), + "dkim_domain": acct.get("dkim_domain", ""), + "enabled": acct.get("enabled", True), + "dm_policy": acct.get("dm_policy", config.get("default_dm_policy", "open")), + "oauth2_enabled": acct.get("oauth2_enabled", False), + "oauth2_client_id": acct.get("oauth2_client_id", ""), + "oauth2_client_secret": acct.get("oauth2_client_secret", ""), + "oauth2_refresh_token": acct.get("oauth2_refresh_token", ""), + "oauth2_access_token": acct.get("oauth2_access_token", ""), + "oauth2_token_expiry": acct.get("oauth2_token_expiry", 0.0), + } + + def is_configured(self, account: dict) -> bool: + return bool(account.get("email_address") and account.get("smtp_host") and account.get("imap_host")) + + def is_enabled(self, account: dict) -> bool: + return account.get("enabled", True) + + def disabled_reason(self, account: dict) -> str: + if not self.is_enabled(account): + return "\u5df2\u7981\u7528" + if not self.is_configured(account): + return "\u914d\u7f6e\u4e0d\u5b8c\u6574" + return "" + + def describe_account(self, account: dict) -> dict: + return { + "account_id": account.get("account_id", ""), + "email_address": account.get("email_address", ""), + } + + @staticmethod + def _resolve_secret(acct: dict, key: str) -> str: + raw = acct.get(key, "") + if not raw: + env_var = acct.get(f"{key}_env", "") + if env_var: + raw = os.getenv(env_var, "") + return raw + + def _resolve_secret_with_vault(self, acct: dict, key: str) -> str: + raw = self._resolve_secret(acct, key) + if not raw: + return "" + encrypted_key = f"{key}_encrypted" + if acct.get(encrypted_key): + try: + ciphertext = bytes.fromhex(raw) + return self._vault.decrypt(ciphertext) + except Exception: + logger.warning("Failed to decrypt %s, using raw value", key) + return raw + + @staticmethod + def make_email_account(account_dict: dict) -> EmailAccount: + smtp = SmtpConfig( + host=account_dict.get("smtp_host", ""), + port=account_dict.get("smtp_port", 465), + use_tls=account_dict.get("smtp_use_tls", True), + starttls=account_dict.get("starttls", False), + username=account_dict.get("smtp_username", ""), + password=account_dict.get("smtp_password", ""), + sender_display_name=account_dict.get("display_name", "AI \u5ba2\u670d"), + ) + imap = ImapConfig( + host=account_dict.get("imap_host", ""), + port=account_dict.get("imap_port", 993), + use_ssl=account_dict.get("imap_use_ssl", True), + username=account_dict.get("imap_username", ""), + password=account_dict.get("imap_password", ""), + idle_timeout_secs=account_dict.get("idle_timeout_secs", 0), + poll_fallback_secs=account_dict.get("poll_fallback_secs", 60), + max_fetch_per_cycle=account_dict.get("max_fetch_per_cycle", 50), + ) + return EmailAccount( + account_id=account_dict.get("account_id", "default"), + email_address=account_dict.get("email_address", ""), + display_name=account_dict.get("display_name", "AI \u5ba2\u670d"), + smtp=smtp, + imap=imap, + dkim_selector=account_dict.get("dkim_selector", ""), + dkim_private_key=account_dict.get("dkim_private_key", ""), + dkim_domain=account_dict.get("dkim_domain", ""), + enabled=account_dict.get("enabled", True), + dm_policy=account_dict.get("dm_policy", "open"), + oauth2_enabled=account_dict.get("oauth2_enabled", False), + oauth2_client_id=account_dict.get("oauth2_client_id", ""), + oauth2_client_secret=account_dict.get("oauth2_client_secret", ""), + oauth2_refresh_token=account_dict.get("oauth2_refresh_token", ""), + oauth2_access_token=account_dict.get("oauth2_access_token", ""), + oauth2_token_expiry=account_dict.get("oauth2_token_expiry", 0.0), + ) + + def _load_config(self) -> dict[str, Any]: + try: + from yuxi.config import get_channel_config + return get_channel_config("email-smtp") or {} + except ImportError: + return {} diff --git a/backend/package/yuxi/channel/extensions/email_smtp/dedupe.py b/backend/package/yuxi/channel/extensions/email_smtp/dedupe.py new file mode 100644 index 00000000..4ae61201 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/dedupe.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import hashlib +import time +from collections import OrderedDict + + +class EmailDeduplicator: + def __init__(self, max_entries: int = 100_000, ttl_seconds: int = 259200): + self._seen: OrderedDict[str, float] = OrderedDict() + self._max_entries = max_entries + self._ttl = ttl_seconds + + def is_duplicate(self, message_id: str, uid: str = "") -> bool: + key = self._make_key(message_id, uid) + self._expire() + return key in self._seen + + def mark(self, message_id: str, uid: str = ""): + key = self._make_key(message_id, uid) + self._expire() + self._seen[key] = time.monotonic() + if len(self._seen) > self._max_entries: + self._seen.popitem(last=False) + + def _make_key(self, message_id: str, uid: str) -> str: + raw = f"{message_id}|{uid}".strip().lower() + return hashlib.sha256(raw.encode()).hexdigest()[:32] + + def _expire(self): + now = time.monotonic() + while self._seen: + key, ts = next(iter(self._seen.items())) + if now - ts > self._ttl: + self._seen.popitem(last=False) + else: + break + + def handle_uidvalidity_change(self): + self._seen.clear() diff --git a/backend/package/yuxi/channel/extensions/email_smtp/errors.py b/backend/package/yuxi/channel/extensions/email_smtp/errors.py new file mode 100644 index 00000000..0adf48a8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/errors.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class EmailSmtpError(Exception): + message: str + code: str = "" + retryable: bool = False + permanent: bool = False + + +class ImapAuthError(EmailSmtpError): + pass + + +class ImapConnectionError(EmailSmtpError): + pass + + +class SmtpAuthError(EmailSmtpError): + pass + + +class SmtpSendError(EmailSmtpError): + pass + + +class EmailParseError(EmailSmtpError): + pass + + +SMTP_ERROR_CLASSIFICATION: dict[int, tuple[str, bool, bool]] = { + 421: ("SERVICE_UNAVAILABLE", True, False), + 450: ("MAILBOX_UNAVAILABLE", True, False), + 451: ("LOCAL_ERROR", True, False), + 452: ("INSUFFICIENT_STORAGE", True, False), + 500: ("SYNTAX_ERROR", False, True), + 501: ("PARAM_SYNTAX_ERROR", False, True), + 502: ("COMMAND_NOT_IMPLEMENTED", False, True), + 503: ("BAD_SEQUENCE", False, True), + 504: ("PARAM_NOT_IMPLEMENTED", False, True), + 530: ("AUTH_REQUIRED", False, True), + 550: ("MAILBOX_NOT_FOUND", False, True), + 551: ("USER_NOT_LOCAL", False, True), + 552: ("STORAGE_EXCEEDED", False, True), + 553: ("MAILBOX_NAME_INVALID", False, True), + 554: ("TRANSACTION_FAILED", False, True), +} + + +def classify_smtp_error(code: int, message: str) -> EmailSmtpError: + if code in SMTP_ERROR_CLASSIFICATION: + desc, retryable, permanent = SMTP_ERROR_CLASSIFICATION[code] + return SmtpSendError(message=f"[{code}] {desc}: {message}", retryable=retryable, permanent=permanent) + if 400 <= code < 500: + return SmtpSendError(message=f"[{code}] {message}", retryable=True, permanent=False) + return SmtpSendError(message=f"[{code}] {message}", retryable=False, permanent=True) diff --git a/backend/package/yuxi/channel/extensions/email_smtp/format.py b/backend/package/yuxi/channel/extensions/email_smtp/format.py new file mode 100644 index 00000000..4a039dd8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/format.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import html +import re + +from bs4 import BeautifulSoup + + +def strip_html_to_text(html_content: str) -> str: + if not html_content: + return "" + soup = BeautifulSoup(html_content, "lxml") + for tag in soup(["style", "script", "nav", "header", "footer"]): + tag.decompose() + text = soup.get_text(separator="\n") + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +_EMAIL_TABLE_STYLE = "background:#ffffff;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;" +_EMAIL_FOOTER = "此邮件由 ForcePilot AI 自动生成" + + +def _build_paragraph_html(text: str) -> str: + text = text.strip() + if not text: + return "" + text = text.replace("\n", "
") + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub( + r"`([^`]+)`", + ('\\1'), + text, + ) + return f'

{text}

' + + +def build_email_html(markdown_text: str) -> str: + escaped = html.escape(markdown_text) + paragraphs = escaped.split("\n\n") + body_html = "".join(_build_paragraph_html(p) for p in paragraphs) + + return ( + "\n" + '\n' + '' + '\n' + '\n' + f'\n' + " \n" + ' \n" + " \n" + "
\n' + f" {body_html}\n" + '
\n' + f'

{_EMAIL_FOOTER}

\n' + "
\n" + "\n" + "" + ) + + +def build_email_system_prompt() -> str: + return ( + "你是一个邮件客服 AI。回答时请遵循以下规范：\n" + "1. 使用正式但友好的语气，避免过于口语化\n" + "2. 不要使用 Markdown 格式（邮件客户端可能不支持）\n" + "3. 回复开头不需要问候语，直接回答问题\n" + "4. 保持回复简洁，每次聚焦于用户的核心问题\n" + "5. 不要使用表情符号\n" + "6. 段落之间用空行分隔，使用 **粗体** 强调关键信息" + ) diff --git a/backend/package/yuxi/channel/extensions/email_smtp/gateway.py b/backend/package/yuxi/channel/extensions/email_smtp/gateway.py new file mode 100644 index 00000000..d026c5a3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/gateway.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import asyncio +import logging + +from yuxi.channel.context import ChannelContext + +from .bounce_handler import EmailBounceHandler +from .config import EmailSmtpConfigAdapter +from .dedupe import EmailDeduplicator +from .imap_client import ImapClient +from .monitor import EmailSmtpMonitor +from .outbound import EmailSmtpOutboundAdapter +from .smtp_client import SmtpClientManager +from .types import EmailAccount, ImapConnectionState + +logger = logging.getLogger(__name__) + + +class EmailSmtpGatewayAdapter: + def __init__(self, outbound: EmailSmtpOutboundAdapter | None = None, dedupe: EmailDeduplicator | None = None): + self._active_imap_clients: dict[str, ImapClient] = {} + self._smtp_managers: dict[str, SmtpClientManager] = {} + self._imap_states: dict[str, ImapConnectionState] = {} + self._config_adapter = EmailSmtpConfigAdapter() + self._monitor = EmailSmtpMonitor() + self._deduplicator = dedupe or EmailDeduplicator() + self._bounce_handler = EmailBounceHandler() + self._outbound = outbound + + async def start(self, ctx: ChannelContext) -> object: + account_dict = self._config_adapter.resolve_account(ctx.account_id, ctx.config) + account = EmailSmtpConfigAdapter.make_email_account(account_dict) + account_id = account.account_id + + self._imap_states[account_id] = ImapConnectionState() + + smtp_config = account.smtp + smtp_mgr = SmtpClientManager(smtp_config) + self._smtp_managers[account_id] = smtp_mgr + + if self._outbound: + self._outbound.set_smtp_managers(self._smtp_managers) + self._outbound.set_imap_clients(self._active_imap_clients) + + imap_client = ImapClient( + host=account.imap.host, + port=account.imap.port, + username=account.imap.username, + password=account.imap.password, + use_ssl=account.imap.use_ssl, + idle_timeout_secs=account.imap.idle_timeout_secs, + poll_fallback_secs=account.imap.poll_fallback_secs, + max_fetch_per_cycle=account.imap.max_fetch_per_cycle, + ) + self._active_imap_clients[account_id] = imap_client + + reconnect_delay = 1 + + while not ctx.cancel_event.is_set(): + try: + if account.oauth2_enabled and account.oauth2_refresh_token: + from .oauth2 import EmailOAuth2Manager + + provider = account.email_address.split("@")[-1].split(".")[0] + oauth2_mgr = EmailOAuth2Manager( + provider=provider, + client_id=account.oauth2_client_id, + client_secret=account.oauth2_client_secret, + ) + access_token = await oauth2_mgr.get_access_token(account.oauth2_refresh_token) + await imap_client.connect_oauth2(access_token) + else: + await imap_client.connect() + self._imap_states[account_id].connected = True + self._imap_states[account_id].reconnect_count = 0 + self._imap_states[account_id].uid_validity = imap_client.uid_validity + reconnect_delay = 1 + + if imap_client.uid_validity != self._imap_states[account_id].uid_validity: + self._deduplicator.handle_uidvalidity_change() + self._imap_states[account_id].uid_validity = imap_client.uid_validity + + async for parsed_email in imap_client.listen(ctx.cancel_event): + await self._handle_inbound(parsed_email, account, ctx) + + except asyncio.CancelledError: + break + except Exception as e: + self._imap_states[account_id].connected = False + self._imap_states[account_id].last_error = str(e) + logger.error( + "IMAP connection error for %s: %s, retrying in %ds", + account_id, + e, + reconnect_delay, + ) + await asyncio.sleep(reconnect_delay) + reconnect_delay = min(reconnect_delay * 2, 300) + finally: + try: + await imap_client.disconnect() + except Exception: + pass + self._imap_states[account_id].connected = False + + return ctx.queue + + async def stop(self, ctx: ChannelContext) -> None: + account_id = ctx.account_id + client = self._active_imap_clients.pop(account_id, None) + if client: + await client.disconnect() + smtp_mgr = self._smtp_managers.pop(account_id, None) + if smtp_mgr: + await smtp_mgr.close() + self._imap_states.pop(account_id, None) + + async def _handle_inbound(self, parsed_email, account: EmailAccount, ctx: ChannelContext): + if self._deduplicator.is_duplicate(parsed_email.message_id, parsed_email.uid): + return + + auto_submitted = parsed_email.auto_submitted or "" + if auto_submitted.lower() in ("auto-replied", "auto-generated"): + logger.debug("Skipping auto-submitted email: %s", parsed_email.message_id) + return + + bounce_info = self._bounce_handler.analyze(parsed_email) + if bounce_info.is_bounce: + logger.warning( + "Bounce email detected from=%s recipient=%s code=%s retryable=%s", + parsed_email.from_email, + bounce_info.original_recipient, + bounce_info.status_code, + bounce_info.retryable, + ) + self._deduplicator.mark(parsed_email.message_id, parsed_email.uid) + return + + self._deduplicator.mark(parsed_email.message_id, parsed_email.uid) + + unified = self._monitor.parse_to_unified(parsed_email, account) + + if self._outbound: + self._outbound.set_last_message_metadata(unified.metadata or {}) + + if ctx.queue is not None: + await ctx.queue.put(unified) + + def get_imap_client(self, account_id: str) -> ImapClient | None: + return self._active_imap_clients.get(account_id) + + def get_smtp_manager(self, account_id: str) -> SmtpClientManager | None: + return self._smtp_managers.get(account_id) + + def get_imap_state(self, account_id: str) -> ImapConnectionState | None: + return self._imap_states.get(account_id) diff --git a/backend/package/yuxi/channel/extensions/email_smtp/imap_client.py b/backend/package/yuxi/channel/extensions/email_smtp/imap_client.py new file mode 100644 index 00000000..7c5b429a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/imap_client.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import asyncio +import logging +import time +from email.utils import parsedate_to_datetime + +from imap_tools import AND, A, MailBox + +from .charset_utils import decode_mime_header +from .mime_parser import parse_mime_email +from .types import ParsedEmail + +logger = logging.getLogger(__name__) + +PROVIDER_IDLE_TIMEOUTS: dict[str, int] = { + "gmail.com": 25 * 60, + "googlemail.com": 25 * 60, + "outlook.com": 25 * 60, + "office365.com": 25 * 60, + "hotmail.com": 25 * 60, + "live.com": 25 * 60, + "qq.com": 8 * 60, + "foxmail.com": 8 * 60, + "163.com": 25 * 60, + "126.com": 25 * 60, + "yeah.net": 25 * 60, + "aliyun.com": 25 * 60, +} + + +class ImapClient: + def __init__( + self, + host: str, + port: int, + username: str, + password: str, + use_ssl: bool = True, + idle_timeout_secs: int = 0, + poll_fallback_secs: int = 60, + max_fetch_per_cycle: int = 50, + ): + self._host = host + self._port = port + self._username = username + self._password = password + self._use_ssl = use_ssl + self._idle_timeout = idle_timeout_secs or self._detect_idle_timeout() + self._poll_fallback = poll_fallback_secs + self._max_fetch = max_fetch_per_cycle + self._mailbox: MailBox | None = None + self._last_uid: str = "" + self._uid_validity: int = 0 + + def _detect_idle_timeout(self) -> int: + domain = self._username.split("@")[-1].lower() if "@" in self._username else "" + return PROVIDER_IDLE_TIMEOUTS.get(domain, 20 * 60) + + async def connect(self): + self._mailbox = await asyncio.to_thread(MailBox, self._host, self._port) + await asyncio.to_thread(self._mailbox.login, self._username, self._password, initial_folder="INBOX") + folder_info = await asyncio.to_thread(self._mailbox.folder.status, "INBOX") + self._uid_validity = folder_info.get("UIDVALIDITY", 0) + + async def connect_oauth2(self, access_token: str): + from .oauth2 import EmailOAuth2Manager + + xoauth2 = EmailOAuth2Manager.build_xoauth2_string(self._username, access_token) + self._mailbox = await asyncio.to_thread(MailBox, self._host, self._port) + await asyncio.to_thread(self._mailbox.login, self._username, xoauth2, initial_folder="INBOX") + folder_info = await asyncio.to_thread(self._mailbox.folder.status, "INBOX") + self._uid_validity = folder_info.get("UIDVALIDITY", 0) + + async def disconnect(self): + if self._mailbox: + try: + await asyncio.to_thread(self._mailbox.logout) + except Exception: + pass + self._mailbox = None + + async def append_to_sent(self, raw_message: bytes) -> None: + if self._mailbox: + try: + await asyncio.to_thread( + self._mailbox.append, + raw_message, + "Sent", + "\\Seen", + ) + except Exception as e: + logger.warning("Failed to append to Sent folder: %s", e) + + async def listen(self, cancel_event: asyncio.Event): + while not cancel_event.is_set(): + async for email in self._process_backlog(): + yield email + + try: + idle_start = time.monotonic() + responses = await asyncio.to_thread(self._mailbox.idle.wait, timeout=min(self._idle_timeout, 300)) + if responses: + async for email in self._process_backlog(): + yield email + elapsed = time.monotonic() - idle_start + if elapsed >= self._idle_timeout * 0.9: + break + except Exception: + await asyncio.sleep(self._poll_fallback) + async for email in self._process_backlog(): + yield email + break + + async def _process_backlog(self): + criteria = A(seen=False) + if self._last_uid: + criteria = AND(seen=False, uid=f"{self._last_uid}:*") + + messages = await asyncio.to_thread( + lambda: list(self._mailbox.fetch(criteria, mark_seen=True, limit=self._max_fetch, reverse=True)) + ) + + for msg in messages: + parsed = await asyncio.to_thread(self._parse_message, msg) + if parsed: + self._last_uid = msg.uid + yield parsed + + def _parse_message(self, msg) -> ParsedEmail | None: + try: + parsed = parse_mime_email(msg.obj) + email = ParsedEmail( + message_id=msg.obj.get("Message-ID", "").strip("<>"), + uid=msg.uid, + uid_validity=self._uid_validity, + subject=decode_mime_header(msg.obj.get("Subject", "")), + from_display_name=parsed["from_display_name"], + from_email=parsed["from_email"], + to_list=parsed.get("to_list", []), + cc_list=parsed.get("cc_list", []), + in_reply_to=msg.obj.get("In-Reply-To", "").strip("<>") or None, + references=self._parse_references(msg.obj.get("References", "")), + body_text=parsed.get("body_text", ""), + body_html=parsed.get("body_html", ""), + date=parsedate_to_datetime(msg.obj.get("Date")) if msg.obj.get("Date") else None, + attachments=parsed.get("attachments", []), + raw_bytes=msg.obj.as_bytes(), + gm_thrid=msg.obj.get("X-GM-THRID"), + auto_submitted=parsed.get("auto_submitted", ""), + ) + return email + except Exception as e: + logger.warning("Failed to parse email message: %s", e) + return None + + @staticmethod + def _parse_references(raw: str) -> list[str]: + if not raw: + return [] + return [ref.strip("<>").strip() for ref in raw.split() if ref.strip("<>").strip()] + + @property + def uid_validity(self) -> int: + return self._uid_validity diff --git a/backend/package/yuxi/channel/extensions/email_smtp/mime_parser.py b/backend/package/yuxi/channel/extensions/email_smtp/mime_parser.py new file mode 100644 index 00000000..0562b3a9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/mime_parser.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import email +from email import policy +from email.utils import getaddresses, parseaddr + +from .charset_utils import decode_mime_header, safe_decode_payload +from .reply_stripper import strip_quotation +from .types import EmailAttachment + + +def parse_mime_email(raw_msg: bytes) -> dict: + msg = email.message_from_bytes(raw_msg, policy=policy.default) + + from_display_name, from_email = parseaddr(decode_mime_header(str(msg["From"]))) + to_list = _parse_address_list(str(msg["To"])) + cc_list = _parse_address_list(str(msg["Cc"])) + + body_text = "" + body_html = "" + attachments: list[EmailAttachment] = [] + + if msg.is_multipart(): + for part in msg.walk(): + content_type = part.get_content_type() + disposition = part.get_content_disposition() + + if content_type == "text/plain" and disposition != "attachment": + payload = part.get_payload(decode=True) + charset = part.get_content_charset() + body_text += safe_decode_payload(payload or b"", charset) + + elif content_type == "text/html" and disposition != "attachment": + payload = part.get_payload(decode=True) + charset = part.get_content_charset() + body_html += safe_decode_payload(payload or b"", charset) + + elif disposition == "attachment" or ( + disposition is None and part.get_content_maintype() not in ("text", "multipart") + ): + filename = part.get_filename() + if filename: + filename = decode_mime_header(filename) + payload = part.get_payload(decode=True) + attachments.append( + EmailAttachment( + filename=filename or "unnamed", + content_type=content_type, + data=payload or b"", + size=len(payload) if payload else 0, + content_id=part.get("Content-ID", "").strip("<>"), + ) + ) + else: + payload = msg.get_payload(decode=True) + charset = msg.get_content_charset() + content_type = msg.get_content_type() + if content_type == "text/html": + body_html = safe_decode_payload(payload or b"", charset) + else: + body_text = safe_decode_payload(payload or b"", charset) + + if not body_text and body_html: + import html2text + + body_text = html2text.html2text(body_html) + + body_text = strip_quotation(body_text) + + return { + "from_display_name": from_display_name, + "from_email": from_email, + "to_list": to_list, + "cc_list": cc_list, + "body_text": body_text.strip(), + "body_html": body_html, + "attachments": attachments, + "auto_submitted": str(msg.get("Auto-Submitted", "")), + } + + +def _parse_address_list(raw: str) -> list[dict]: + if not raw: + return [] + decoded = decode_mime_header(raw) + addresses = getaddresses([decoded]) + return [{"display_name": name.strip() or None, "email": addr.lower().strip()} for name, addr in addresses if addr] diff --git a/backend/package/yuxi/channel/extensions/email_smtp/monitor.py b/backend/package/yuxi/channel/extensions/email_smtp/monitor.py new file mode 100644 index 00000000..275580af --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/monitor.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from yuxi.channel.message.models import PeerInfo, UnifiedMessage +from yuxi.channel.routing.models import PeerKind + +from .types import EmailAccount, ParsedEmail + + +class EmailSmtpMonitor: + def parse_to_unified(self, email: ParsedEmail, account: EmailAccount) -> UnifiedMessage: + is_group = len(email.to_list) > 1 or len(email.cc_list) > 0 + + thread_id = email.in_reply_to or email.message_id + if email.references: + for ref in reversed(email.references): + if ref: + thread_id = ref + break + + peer = PeerInfo( + kind=PeerKind.DIRECT if not is_group else PeerKind.GROUP, + id=email.from_email, + display_name=email.from_display_name or email.from_email, + ) + + return UnifiedMessage( + msg_id=f"email-smtp:{email.message_id}", + channel_type="email-smtp", + account_id=account.account_id, + sender=peer, + content=email.body_text, + reply_to_id=email.in_reply_to, + message_thread_id=thread_id, + media_urls=[att.content_id or att.filename for att in email.attachments], + metadata={ + "subject": email.subject, + "to": [t["email"] for t in email.to_list], + "cc": [t["email"] for t in email.cc_list], + "date": email.date.isoformat() if email.date else None, + "gm_thrid": email.gm_thrid, + "uid": email.uid, + "uid_validity": email.uid_validity, + }, + raw_payload={"raw_bytes": email.raw_bytes}, + ) diff --git a/backend/package/yuxi/channel/extensions/email_smtp/oauth2.py b/backend/package/yuxi/channel/extensions/email_smtp/oauth2.py new file mode 100644 index 00000000..9cc4d98f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/oauth2.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass + +import aiohttp + + +@dataclass +class OAuth2Token: + access_token: str + refresh_token: str + expires_at: float + token_type: str = "Bearer" + + @property + def is_expired(self) -> bool: + return time.time() > self.expires_at - 60 + + +class EmailOAuth2Manager: + MICROSOFT_TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token" + GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" + + def __init__(self, provider: str, client_id: str, client_secret: str): + self._provider = provider + self._client_id = client_id + self._client_secret = client_secret + self._token: OAuth2Token | None = None + + def _token_url(self) -> str: + if self._provider in ("outlook", "office365", "hotmail", "live"): + return self.MICROSOFT_TOKEN_URL + return self.GOOGLE_TOKEN_URL + + async def get_access_token(self, refresh_token: str) -> str: + if self._token and not self._token.is_expired: + return self._token.access_token + + data = { + "client_id": self._client_id, + "client_secret": self._client_secret, + "refresh_token": refresh_token, + "grant_type": "refresh_token", + } + async with aiohttp.ClientSession() as session: + async with session.post( + self._token_url(), data=data, timeout=aiohttp.ClientTimeout(total=15) + ) as resp: + if resp.status != 200: + raise RuntimeError(f"OAuth2 token refresh failed: {await resp.text()}") + token_data = await resp.json() + self._token = OAuth2Token( + access_token=token_data["access_token"], + refresh_token=token_data.get("refresh_token", refresh_token), + expires_at=time.time() + token_data.get("expires_in", 3600), + ) + return self._token.access_token + + def build_xoauth2_string(self, username: str, access_token: str) -> str: + return f"user={username}\x01auth=Bearer {access_token}\x01\x01" diff --git a/backend/package/yuxi/channel/extensions/email_smtp/outbound.py b/backend/package/yuxi/channel/extensions/email_smtp/outbound.py new file mode 100644 index 00000000..0d9c3a3f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/outbound.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import logging +import os +from email.mime.base import MIMEBase +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +from email import encoders +from email.utils import formataddr, formatdate, make_msgid + +import aiohttp + +from .attachment import is_attachment_safe +from .config import EmailSmtpConfigAdapter +from .format import build_email_html + +logger = logging.getLogger(__name__) + +REPLY_SEPARATOR_HTML = ( + '
--reply-above-this--
' +) + + +class EmailSmtpOutboundAdapter: + def __init__(self, smtp_managers: dict | None = None, imap_clients: dict | None = None): + self._smtp_managers: dict = smtp_managers or {} + self._imap_clients: dict = imap_clients or {} + self._config_adapter = EmailSmtpConfigAdapter() + self._last_message_metadata: dict = {} + + def set_smtp_managers(self, managers: dict): + self._smtp_managers = managers + + def set_imap_clients(self, clients: dict): + self._imap_clients = clients + + def set_last_message_metadata(self, metadata: dict): + self._last_message_metadata = metadata + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + cc_list: list[str] | None = None, + bcc_list: list[str] | None = None, + priority: str | None = None, + ) -> None: + aid = account_id or "default" + account_dict = self._config_adapter.resolve_account(aid) + account = EmailSmtpConfigAdapter.make_email_account(account_dict) + + smtp_mgr = self._smtp_managers.get(aid) + if not smtp_mgr: + logger.error("SMTP manager not found for account %s", aid) + return + + html_body = build_email_html(content) + msg_id = make_msgid(domain=account.email_address.split("@")[-1]) + + msg = MIMEMultipart("alternative") + msg["From"] = formataddr((account.display_name, account.email_address)) + msg["To"] = target_id + if cc_list: + msg["Cc"] = ", ".join(cc_list) + msg["Subject"] = self._resolve_subject() + msg["Date"] = formatdate(localtime=True) + msg["Message-ID"] = msg_id + msg["Auto-Submitted"] = "auto-replied" + msg["X-Mailer"] = "ForcePilot AI/1.0" + + if priority: + self._set_priority_headers(msg, priority) + + if reply_to_id: + msg["In-Reply-To"] = f"<{reply_to_id}>" + msg["References"] = f"<{reply_to_id}>" + + msg.attach(MIMEText(content, "plain", "utf-8")) + msg.attach(MIMEText(html_body + REPLY_SEPARATOR_HTML, "html", "utf-8")) + + all_recipients = [target_id] + if cc_list: + all_recipients.extend(cc_list) + if bcc_list: + all_recipients.extend(bcc_list) + + result = await smtp_mgr.send_message(msg, account, recipients=all_recipients) + if result.success: + try: + imap_client = self._imap_clients.get(aid) + if imap_client: + await imap_client.append_to_sent(msg.as_bytes()) + except Exception: + logger.warning("Failed to append to Sent folder", exc_info=True) + else: + logger.error("SMTP send failed: %s", result.error) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + aid = account_id or "default" + account_dict = self._config_adapter.resolve_account(aid) + account = EmailSmtpConfigAdapter.make_email_account(account_dict) + + smtp_mgr = self._smtp_managers.get(aid) + if not smtp_mgr: + logger.error("SMTP manager not found for account %s", aid) + return + + content_type_map = { + "image": ("image/png", ".png"), + "file": ("application/octet-stream", ""), + "pdf": ("application/pdf", ".pdf"), + } + mime_main, default_ext = content_type_map.get(media_type, ("application/octet-stream", "")) + + try: + async with aiohttp.ClientSession() as session: + async with session.get(media_url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status != 200: + logger.error("Failed to fetch media from %s: HTTP %s", media_url, resp.status) + return + file_data = await resp.read() + except Exception as e: + logger.error("Failed to download media from %s: %s", media_url, e) + return + + filename = os.path.basename(media_url) or f"attachment{default_ext}" + safe, reason = is_attachment_safe( + filename=filename, + content_type=mime_main, + size=len(file_data), + ) + if not safe: + logger.warning("Media rejected by safety filter: %s", reason) + return + + msg = MIMEMultipart("mixed") + msg["From"] = formataddr((account.display_name, account.email_address)) + msg["To"] = target_id + msg["Subject"] = self._resolve_subject() + msg["Date"] = formatdate(localtime=True) + msg["Message-ID"] = make_msgid(domain=account.email_address.split("@")[-1]) + msg["Auto-Submitted"] = "auto-replied" + msg["X-Mailer"] = "ForcePilot AI/1.0" + + if reply_to_id: + msg["In-Reply-To"] = f"<{reply_to_id}>" + msg["References"] = f"<{reply_to_id}>" + + body = MIMEMultipart("alternative") + plain_content = f"[附件] {media_type}" + body.attach(MIMEText(plain_content, "plain", "utf-8")) + msg.attach(body) + + attachment_part = MIMEBase(*mime_main.split("/", 1)) + attachment_part.set_payload(file_data) + encoders.encode_base64(attachment_part) + attachment_part.add_header("Content-Disposition", "attachment", filename=filename) + msg.attach(attachment_part) + + result = await smtp_mgr.send_message(msg, account) + if result.success: + try: + imap_client = self._imap_clients.get(aid) + if imap_client: + await imap_client.append_to_sent(msg.as_bytes()) + except Exception: + logger.warning("Failed to append media to Sent folder", exc_info=True) + else: + logger.error("SMTP media send failed: %s", result.error) + + def _resolve_subject(self) -> str: + original_subject = self._last_message_metadata.get("subject", "") + if original_subject and not original_subject.lower().startswith("re:"): + return f"Re: {original_subject}" + return original_subject or "Re: \u6765\u81ea AI \u5ba2\u670d\u7684\u56de\u590d" + + @staticmethod + def _set_priority_headers(msg: MIMEMultipart, priority: str) -> None: + priority_map = { + "high": {"X-Priority": "1", "X-MSMail-Priority": "High", "Importance": "high"}, + "normal": {"X-Priority": "3", "X-MSMail-Priority": "Normal", "Importance": "normal"}, + "low": {"X-Priority": "5", "X-MSMail-Priority": "Low", "Importance": "low"}, + } + headers = priority_map.get(priority.lower(), priority_map["normal"]) + for key, value in headers.items(): + msg[key] = value diff --git a/backend/package/yuxi/channel/extensions/email_smtp/pairing.py b/backend/package/yuxi/channel/extensions/email_smtp/pairing.py new file mode 100644 index 00000000..65544f21 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/pairing.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import random +import time + + +class EmailSmtpPairingAdapter: + id_label = "email_address" + + def __init__(self): + self._codes: dict[str, dict] = {} + self._code_ttl = 600 + + def generate_code(self, peer_id: str) -> str: + code = str(random.randint(100000, 999999)) + self._codes[peer_id] = {"code": code, "ts": time.monotonic()} + return code + + def verify_code(self, peer_id: str, code: str) -> bool: + entry = self._codes.get(peer_id) + if not entry: + return False + if time.monotonic() - entry["ts"] > self._code_ttl: + del self._codes[peer_id] + return False + return entry["code"] == code + + def normalize_allow_entry(self, entry: str) -> str: + return entry.strip().lower() \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/email_smtp/plugin.json b/backend/package/yuxi/channel/extensions/email_smtp/plugin.json new file mode 100644 index 00000000..463c4ca4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/plugin.json @@ -0,0 +1,32 @@ +{ + "id": "email-smtp", + "name": "\u90ae\u4ef6 (SMTP/IMAP)", + "version": "1.0.0", + "label": "Email", + "aliases": ["email", "smtp", "imap", "mail"], + "description": "SMTP/IMAP \u90ae\u4ef6\u6e20\u9053\u63d2\u4ef6\uff0c\u652f\u6301 IMAP IDLE \u5b9e\u65f6\u6536\u4ef6 + SMTP \u53d1\u4ef6", + "author": "ForcePilot", + "order": 80, + "dependencies": ["aiosmtplib", "imap-tools", "aiohttp", "beautifulsoup4", "html2text", "charset-normalizer"], + "enabled": true, + "python_requires": ">=3.12", + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text"], + "reactions": false, + "typing_indicator": false, + "threads": true, + "edit": false, + "unsend": false, + "reply": true, + "media": true, + "native_commands": false, + "polls": false, + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "block_streaming_chunk_min_chars": 800, + "block_streaming_chunk_max_chars": 1200, + "block_streaming_chunk_break_preference": "paragraph" + } +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/email_smtp/plugin.py b/backend/package/yuxi/channel/extensions/email_smtp/plugin.py new file mode 100644 index 00000000..9bb7bc4b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/plugin.py @@ -0,0 +1,273 @@ +from __future__ import annotations + +import asyncio +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.protocols import SessionResolution + +from .config import EmailSmtpConfigAdapter +from .dedupe import EmailDeduplicator +from .format import build_email_system_prompt +from .gateway import EmailSmtpGatewayAdapter +from .outbound import EmailSmtpOutboundAdapter +from .pairing import EmailSmtpPairingAdapter +from .security import EmailSmtpSecurityAdapter +from .session import EmailSmtpSessionAdapter +from .status import EmailSmtpStatusAdapter +from .streaming import EmailSmtpStreamingAdapter + +logger = logging.getLogger(__name__) + + +class EmailSmtpPlugin(BaseChannelPlugin): + id = "email-smtp" + name = "\u90ae\u4ef6 (SMTP/IMAP)" + order = 80 + label = "\u90ae\u4ef6" + aliases = ["email", "smtp", "imap", "mail"] + + def __init__(self): + self._config = EmailSmtpConfigAdapter() + self._outbound = EmailSmtpOutboundAdapter() + self._dedupe = EmailDeduplicator() + self._gateway = EmailSmtpGatewayAdapter(self._outbound, dedupe=self._dedupe) + self._security = EmailSmtpSecurityAdapter() + self._pairing = EmailSmtpPairingAdapter() + self._session = EmailSmtpSessionAdapter() + self._streaming = EmailSmtpStreamingAdapter(self._outbound) + self._status = EmailSmtpStatusAdapter() + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text"], + reactions=False, + typing_indicator=False, + threads=True, + edit=False, + unsend=False, + reply=True, + media=True, + native_commands=False, + polls=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=800, + block_streaming_chunk_max_chars=1200, + block_streaming_chunk_break_preference="paragraph", + ) + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + return self._config.resolve_account(account_id) + + def is_configured(self, account: dict) -> bool: + return self._config.is_configured(account) + + def is_enabled(self, account: dict) -> bool: + return self._config.is_enabled(account) + + def disabled_reason(self, account: dict) -> str: + return self._config.disabled_reason(account) + + def describe_account(self, account: dict) -> dict: + return self._config.describe_account(account) + + async def start(self, ctx) -> object: + return await self._gateway.start(ctx) + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + cc_list: list[str] | None = None, + bcc_list: list[str] | None = None, + ) -> None: + await self._outbound.send_text( + target_id, + content, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + cc_list=cc_list, + bcc_list=bcc_list, + ) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_media( + target_id, + media_url, + media_type, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + async def probe(self, account: dict) -> bool: + return await self._status.probe(account) + + def build_summary(self, snapshot: object) -> dict: + return self._status.build_summary(snapshot) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + return self._security.check_allowlist(peer_id) + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_dm_policy() + + async def generate_code(self, peer_id: str) -> str: + return self._pairing.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + return self._pairing.verify_code(peer_id, code) + + def normalize_allow_entry(self, entry: str) -> str: + return self._pairing.normalize_allow_entry(entry) + + def is_duplicate(self, key: str) -> bool: + return self._dedupe.is_duplicate(key) + + def mark_seen(self, key: str) -> None: + self._dedupe.mark(key) + + def build_system_prompt(self, context) -> str | None: + return build_email_system_prompt() + + def build_context_note(self, context) -> str: + return "" + + def extract_thread_id(self, msg: object) -> str | None: + if hasattr(msg, "message_thread_id"): + return msg.message_thread_id + return None + + def resolve_session(self, msg: object) -> SessionResolution: + from yuxi.channel.message.models import PeerKind + + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + if msg.sender.kind == PeerKind.DIRECT: + session = self._session.resolve_session(msg) + return SessionResolution( + kind="direct", + conversation_id=msg.sender.id, + thread_id=session.thread_id, + label=session.label, + ) + gid = msg.group.id if hasattr(msg, "group") and msg.group and msg.group.id else "unknown" + return SessionResolution(kind="group", conversation_id=gid) + + @property + def streaming_mode(self) -> str: + return self._streaming.streaming_mode + + @property + def preview_stream_throttle_ms(self) -> int: + return self._streaming.preview_stream_throttle_ms + + @property + def block_streaming_enabled(self) -> bool: + return self._streaming.block_streaming_enabled + + @property + def block_streaming_chunk_min_chars(self) -> int: + return self._streaming.block_streaming_chunk_min_chars + + @property + def block_streaming_chunk_max_chars(self) -> int: + return self._streaming.block_streaming_chunk_max_chars + + @property + def block_streaming_chunk_break_preference(self) -> str: + return self._streaming.block_streaming_chunk_break_preference + + @property + def block_streaming_coalesce_defaults(self) -> dict | None: + return self._streaming.block_streaming_coalesce_defaults + + def create_block_chunker(self) -> object: + return self._streaming.create_block_chunker() + + def create_draft_stream_session(self, target_id: str) -> object: + return self._streaming.create_draft_stream_session(target_id) + + @property + def channel_format_instructions(self) -> str | None: + return ( + "You are sending email replies via SMTP. " + "Use plain paragraphs separated by blank lines. " + "Use **bold** for emphasis. Do not use Markdown tables or code blocks. " + "Each reply should be concise and focused on the user's question. " + "Do not include greetings like 'Hello' or 'Dear' at the start. " + "Sign-off is optional - the system will add its own footer." + ) + + @property + def ping_interval_ms(self) -> int: + return 5 * 60 * 1000 + + async def on_ping(self, account: dict) -> bool: + try: + imap_client = self._gateway.get_imap_client(account.get("account_id", "default")) + if imap_client and imap_client._mailbox: + await asyncio.to_thread(imap_client._mailbox.noop) + return True + except Exception: + return False + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + client = self._gateway.get_imap_client(account_id) + if client: + await client.disconnect() + + async def on_account_removed(self, account_id: str) -> None: + client = self._gateway.get_imap_client(account_id) + if client: + await client.disconnect() + + async def on_message_received(self, msg: object) -> object | None: + return msg + + async def on_message_sending(self, msg: object, content: str) -> str | None: + logger.debug("Sending to %s: %s...", getattr(msg, "target_id", "unknown"), content[:100]) + return content + + async def before_send_attempt(self, target_id: str, content: str) -> None: + logger.debug("Before send attempt to %s", target_id) + + async def after_send_success(self, target_id: str, receipt: object) -> None: + logger.info("Sent successfully to %s", target_id) + + async def after_send_failure(self, target_id: str, error: str) -> None: + logger.error("Send failed to %s: %s", target_id, error) + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "accounts": { + "type": "object", + "description": "\u591a\u8d26\u6237\u914d\u7f6e", + }, + }, + } diff --git a/backend/package/yuxi/channel/extensions/email_smtp/reply_stripper.py b/backend/package/yuxi/channel/extensions/email_smtp/reply_stripper.py new file mode 100644 index 00000000..12049553 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/reply_stripper.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import re + +from email_reply_parser import EmailReplyParser + +QUOTATION_PATTERNS: list[tuple[re.Pattern, bool]] = [ + (re.compile(r"^\s*-{2,}\s*reply.above.this.line\s*-{2,}", re.MULTILINE | re.IGNORECASE), True), + (re.compile(r"^#{1,3}[-\s]*Please type your reply above this line", re.MULTILINE | re.IGNORECASE), True), + (re.compile(r"^-{3,}\s*原始邮件\s*-{3,}", re.MULTILINE), True), + (re.compile(r"^-{3,}\s*Original Message\s*-{3,}", re.MULTILINE), True), + (re.compile(r"^在\s*\d{4}[年/-]\d{1,2}[月/-]\d{1,2}[日\s]", re.MULTILINE), True), + (re.compile(r"^发件人[：:].*$", re.MULTILINE), False), + (re.compile(r"^On\s.+\s<(.*?)>\s+wrote\s*:", re.MULTILINE), True), + (re.compile(r"^On\s.+,\sat\s.+,\s.+\s+wrote\s*:", re.MULTILINE), True), + (re.compile(r"^_{20,}", re.MULTILINE), True), + (re.compile(r"^From:\s.+[\r\n]+Sent:\s.+[\r\n]+To:\s.+[\r\n]+Subject:", re.MULTILINE), True), + (re.compile(r"^Le\s+.*a\s+écrit\s*:", re.MULTILINE), True), + (re.compile(r"^Am\s+.*schrieb\s+.*:", re.MULTILINE), True), +] + + +def strip_quotation(body: str) -> str: + if not body or len(body) < 20: + return body + + result = EmailReplyParser.parse_reply(body) + if _is_adequately_stripped(body, result): + return result.strip() + + for pattern, cut_before in QUOTATION_PATTERNS: + match = pattern.search(body) + if match: + if cut_before: + return body[:match.start()].strip() + start = match.end() + next_match = pattern.search(body, start) + if next_match: + return body[match.end():next_match.start()].strip() + return body[match.end():].strip() + + return body.strip() + + +def _is_adequately_stripped(original: str, stripped: str) -> bool: + if not stripped or len(stripped) < 5: + return False + if stripped == original: + return False + ratio = len(stripped) / max(len(original), 1) + if ratio < 0.3: + return True + if ">" in stripped and len(stripped) > len(original) * 0.5: + return False + return True \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/email_smtp/security.py b/backend/package/yuxi/channel/extensions/email_smtp/security.py new file mode 100644 index 00000000..ea72675d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/security.py @@ -0,0 +1,31 @@ +from __future__ import annotations + + +class EmailSmtpSecurityAdapter: + def resolve_dm_policy(self, config: dict | None = None) -> dict: + return {"mode": "open", "allow_from": []} + + def check_allowlist(self, sender_email: str) -> bool: + config = self._load_config() + policy = config.get("dm_policy", "open") + if policy == "open": + return True + if policy == "disabled": + return False + if policy == "allowlist": + allowlist = config.get("allowlist", []) + return sender_email.lower() in [a.lower() for a in allowlist] + if policy == "pairing": + return True + return False + + def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]: + return [] + + @staticmethod + def _load_config() -> dict: + try: + from yuxi.config import get_channel_config + return get_channel_config("email-smtp") or {} + except ImportError: + return {} diff --git a/backend/package/yuxi/channel/extensions/email_smtp/session.py b/backend/package/yuxi/channel/extensions/email_smtp/session.py new file mode 100644 index 00000000..e7d1c352 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/session.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from yuxi.channel.message.models import UnifiedMessage +from yuxi.channel.protocols import SessionResolution + + +class EmailSmtpSessionAdapter: + def __init__(self): + self._thread_sessions: dict[str, dict] = {} + + def resolve_session(self, msg: UnifiedMessage, account_id: str = "") -> SessionResolution: + thread_id = msg.message_thread_id or msg.msg_id + reply_to_id = msg.reply_to_id + sender_id = msg.sender.id if msg.sender else "unknown" + + if thread_id not in self._thread_sessions: + session_id = f"email:{account_id}:{thread_id}" + self._thread_sessions[thread_id] = { + "session_id": session_id, + "root_message_id": thread_id, + "branch_sessions": {}, + } + + thread = self._thread_sessions[thread_id] + + if reply_to_id and reply_to_id in thread.get("branch_sessions", {}): + session_id = thread["branch_sessions"][reply_to_id] + elif reply_to_id and reply_to_id != thread.get("root_message_id"): + session_id = f"email:{account_id}:{thread_id}:branch:{reply_to_id}" + thread.setdefault("branch_sessions", {})[reply_to_id] = session_id + else: + session_id = thread["session_id"] + + return SessionResolution( + kind="direct", + conversation_id=sender_id, + thread_id=session_id, + label=msg.metadata.get("subject", "") if msg.metadata else "", + ) + + def extract_thread_id(self, msg: UnifiedMessage) -> str | None: + return msg.message_thread_id or msg.msg_id diff --git a/backend/package/yuxi/channel/extensions/email_smtp/smtp_client.py b/backend/package/yuxi/channel/extensions/email_smtp/smtp_client.py new file mode 100644 index 00000000..7b8b0331 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/smtp_client.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import asyncio +import logging +import time +from email.mime.multipart import MIMEMultipart + +import aiosmtplib + +from .types import EmailAccount, OutboundResult + +logger = logging.getLogger(__name__) + + +class RateLimiter: + def __init__(self, max_per_hour: int = 60, burst: int = 10): + self._rate = max_per_hour / 3600.0 + self._burst = burst + self._tokens = float(burst) + self._last_refill = time.monotonic() + + async def acquire(self): + while True: + now = time.monotonic() + elapsed = now - self._last_refill + self._tokens = min(self._burst, self._tokens + elapsed * self._rate) + self._last_refill = now + if self._tokens >= 1: + self._tokens -= 1 + return + await asyncio.sleep(0.5) + + +class SmtpClientManager: + def __init__(self, smtp_config): + self._config = smtp_config + self._rate_limiter = RateLimiter(max_per_hour=60, burst=10) + self._last_send_per_recipient: dict[str, float] = {} + self._pool: list[aiosmtplib.SMTP] = [] + self._pool_size = 3 + self._pool_lock = asyncio.Lock() + + async def _acquire_connection(self) -> aiosmtplib.SMTP: + async with self._pool_lock: + while self._pool: + conn = self._pool.pop() + try: + await conn.noop() + return conn + except Exception: + pass + if self._config.starttls: + conn = aiosmtplib.SMTP( + hostname=self._config.host, + port=self._config.port, + use_tls=False, + ) + await conn.connect() + await conn.starttls() + else: + conn = aiosmtplib.SMTP( + hostname=self._config.host, + port=self._config.port, + use_tls=self._config.use_tls, + ) + await conn.connect() + return conn + + async def _release_connection(self, conn: aiosmtplib.SMTP): + async with self._pool_lock: + if len(self._pool) < self._pool_size: + self._pool.append(conn) + else: + try: + await conn.quit() + except Exception: + pass + + async def _authenticate(self, conn: aiosmtplib.SMTP, account: EmailAccount): + if account.oauth2_enabled and account.oauth2_refresh_token: + from .oauth2 import EmailOAuth2Manager + + provider = account.email_address.split("@")[-1].split(".")[0] + oauth2_mgr = EmailOAuth2Manager( + provider=provider, + client_id=account.oauth2_client_id, + client_secret=account.oauth2_client_secret, + ) + access_token = await oauth2_mgr.get_access_token(account.oauth2_refresh_token) + xoauth2 = oauth2_mgr.build_xoauth2_string(account.email_address, access_token) + await conn.login(account.email_address, xoauth2) + elif self._config.username and self._config.password: + await conn.login(self._config.username, self._config.password) + + async def send_message( + self, msg: MIMEMultipart, account: EmailAccount, recipients: list[str] | None = None + ) -> OutboundResult: + await self._rate_limiter.acquire() + + to_addr = msg["To"] + if to_addr: + now = time.monotonic() + last = self._last_send_per_recipient.get(to_addr, 0) + if now - last < 30: + await asyncio.sleep(30 - (now - last)) + self._last_send_per_recipient[to_addr] = time.monotonic() + + if account.dkim_private_key and account.dkim_selector and account.dkim_domain: + msg = self._dkim_sign(msg, account) + + conn = await self._acquire_connection() + try: + await self._authenticate(conn, account) + result = await conn.send_message(msg, recipients=recipients) + queue_id = result.get("message", "").split()[-1] if result.get("message") else None + + return OutboundResult( + success=True, + message_id=msg["Message-ID"], + queue_id=queue_id, + ) + except aiosmtplib.SMTPAuthenticationError as e: + return OutboundResult(success=False, error=str(e), retryable=False) + except aiosmtplib.SMTPConnectError as e: + return OutboundResult(success=False, error=str(e), retryable=True) + except aiosmtplib.SMTPResponseException as e: + retryable = 400 <= e.code < 500 + return OutboundResult(success=False, error=str(e), retryable=retryable) + except Exception as e: + return OutboundResult(success=False, error=str(e), retryable=True) + finally: + await self._release_connection(conn) + + def _dkim_sign(self, msg: MIMEMultipart, account: EmailAccount) -> MIMEMultipart: + try: + import dkim + + msg_bytes = msg.as_bytes() + sig = dkim.sign( + message=msg_bytes, + selector=account.dkim_selector.encode(), + domain=account.dkim_domain.encode(), + privkey=account.dkim_private_key.encode(), + include_headers=[ + b"from", + b"to", + b"subject", + b"date", + b"message-id", + b"in-reply-to", + b"references", + ], + canonicalize=(b"relaxed", b"relaxed"), + ) + signed_bytes = sig + msg_bytes + from email import message_from_bytes, policy + + return message_from_bytes(signed_bytes, policy=policy.default) + except Exception: + return msg + + async def close(self): + async with self._pool_lock: + for conn in self._pool: + try: + await conn.quit() + except Exception: + pass + self._pool.clear() + + +async def probe_smtp(host: str, port: int, use_tls: bool = True) -> bool: + smtp = aiosmtplib.SMTP(hostname=host, port=port, use_tls=use_tls) + try: + await asyncio.wait_for(smtp.connect(), timeout=10.0) + await smtp.ehlo() + return True + except Exception: + return False + finally: + try: + await smtp.quit() + except Exception: + pass diff --git a/backend/package/yuxi/channel/extensions/email_smtp/status.py b/backend/package/yuxi/channel/extensions/email_smtp/status.py new file mode 100644 index 00000000..8eaeb8a9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/status.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime, UTC + + +from .config import EmailSmtpConfigAdapter +from .smtp_client import probe_smtp +from .types import EmailAccount + +logger = logging.getLogger(__name__) + + +class EmailSmtpStatusAdapter: + def __init__(self): + self._config_adapter = EmailSmtpConfigAdapter() + self._last_probe: dict[str, dict] = {} + + async def probe(self, account: dict) -> bool: + account_dict = self._config_adapter.resolve_account(account.get("account_id", "default")) + email_acct = EmailSmtpConfigAdapter.make_email_account(account_dict) + + smtp_ok = await probe_smtp(email_acct.smtp.host, email_acct.smtp.port, email_acct.smtp.use_tls) + imap_ok = await self._probe_imap(email_acct) + + result = { + "status": "ok" if (smtp_ok and imap_ok) else ("degraded" if (smtp_ok or imap_ok) else "down"), + "smtp_connected": smtp_ok, + "imap_connected": imap_ok, + "probed_at": datetime.now(UTC).isoformat(), + "account_id": account.get("account_id", "default"), + } + self._last_probe[account.get("account_id", "default")] = result + return smtp_ok and imap_ok + + def build_summary(self, snapshot: object) -> dict: + account_id = getattr(snapshot, "account_id", "default") + result = self._last_probe.get(account_id, {}) + return { + "channel_type": "email-smtp", + "account_id": account_id, + "status": result.get("status", "unknown"), + "smtp_ok": result.get("smtp_connected", False), + "imap_ok": result.get("imap_connected", False), + "last_probe": result.get("probed_at", ""), + } + + async def _probe_imap(self, account: EmailAccount) -> bool: + from imap_tools import MailBox + + try: + + def _check(): + with MailBox(account.imap.host, account.imap.port).login( + account.imap.username, account.imap.password, initial_folder="INBOX" + ) as mb: + status = mb.folder.status("INBOX") + return status is not None + + return await asyncio.to_thread(_check) + except Exception as e: + logger.debug("IMAP probe failed: %s", e) + return False diff --git a/backend/package/yuxi/channel/extensions/email_smtp/streaming.py b/backend/package/yuxi/channel/extensions/email_smtp/streaming.py new file mode 100644 index 00000000..0d530001 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/streaming.py @@ -0,0 +1,66 @@ +from __future__ import annotations + + +from .outbound import EmailSmtpOutboundAdapter + + +class EmailSmtpStreamingAdapter: + streaming_mode = "block" + preview_stream_throttle_ms = 160 + preview_min_initial_chars = 18 + block_streaming_enabled = True + block_streaming_break = "text_end" + block_streaming_chunk_min_chars = 800 + block_streaming_chunk_max_chars = 1200 + block_streaming_chunk_break_preference = "paragraph" + block_streaming_coalesce_defaults = None + + def __init__(self, outbound: EmailSmtpOutboundAdapter | None = None): + self._outbound = outbound or EmailSmtpOutboundAdapter() + self._idle_ms = 800 + + def create_block_chunker(self) -> BlockChunker: + return BlockChunker( + min_chars=self.block_streaming_chunk_min_chars, + max_chars=self.block_streaming_chunk_max_chars, + break_preference=self.block_streaming_chunk_break_preference, + ) + + def create_draft_stream_session(self, target_id: str) -> dict: + return { + "target_id": target_id, + "accumulated": "", + "chunks_sent": 0, + } + + +class BlockChunker: + def __init__(self, min_chars: int = 800, max_chars: int = 1200, break_preference: str = "paragraph"): + self._min_chars = min_chars + self._max_chars = max_chars + self._break_preference = break_preference + + def chunk(self, text: str) -> list[str]: + if len(text) <= self._max_chars: + return [text] if text else [] + + chunks = [] + remaining = text + while len(remaining) > self._max_chars: + split_at = self._max_chars + if self._break_preference == "paragraph": + para_break = remaining.rfind("\n\n", 0, self._max_chars) + if para_break > self._min_chars: + split_at = para_break + 2 + else: + line_break = remaining.rfind("\n", 0, self._max_chars) + if line_break > self._min_chars: + split_at = line_break + 1 + + chunks.append(remaining[:split_at].strip()) + remaining = remaining[split_at:].strip() + + if remaining: + chunks.append(remaining) + + return chunks diff --git a/backend/package/yuxi/channel/extensions/email_smtp/types.py b/backend/package/yuxi/channel/extensions/email_smtp/types.py new file mode 100644 index 00000000..79d35fa2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/email_smtp/types.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime + + +@dataclass +class SmtpConfig: + host: str + port: int = 465 + use_tls: bool = True + starttls: bool = False + username: str = "" + password: str = "" + sender_display_name: str = "AI \u5ba2\u670d" + + +@dataclass +class ImapConfig: + host: str + port: int = 993 + use_ssl: bool = True + username: str = "" + password: str = "" + idle_timeout_secs: int = 0 + poll_fallback_secs: int = 60 + max_fetch_per_cycle: int = 50 + + +@dataclass +class EmailAccount: + account_id: str + email_address: str + display_name: str + smtp: SmtpConfig = field(default_factory=lambda: SmtpConfig("")) + imap: ImapConfig = field(default_factory=lambda: ImapConfig("")) + dkim_selector: str = "" + dkim_private_key: str = "" + dkim_domain: str = "" + enabled: bool = True + dm_policy: str = "open" + oauth2_enabled: bool = False + oauth2_client_id: str = "" + oauth2_client_secret: str = "" + oauth2_refresh_token: str = "" + oauth2_access_token: str = "" + oauth2_token_expiry: float = 0.0 + + +@dataclass +class ParsedEmail: + message_id: str + uid: str + uid_validity: int + subject: str + from_display_name: str + from_email: str + to_list: list[dict] = field(default_factory=list) + cc_list: list[dict] = field(default_factory=list) + in_reply_to: str | None = None + references: list[str] = field(default_factory=list) + body_text: str = "" + body_html: str = "" + date: datetime | None = None + attachments: list[EmailAttachment] = field(default_factory=list) + raw_bytes: bytes = b"" + gm_thrid: str | None = None + auto_submitted: str = "" + + +@dataclass +class EmailAttachment: + filename: str + content_type: str + data: bytes + size: int + content_id: str = "" + + +@dataclass +class OutboundResult: + success: bool + message_id: str | None = None + queue_id: str | None = None + error: str | None = None + retryable: bool = False + + +@dataclass +class ImapConnectionState: + connected: bool = False + uid_validity: int = 0 + last_uid: str = "" + last_idle_start: float = 0.0 + reconnect_count: int = 0 + last_error: str = "" + + +@dataclass +class ThreadNode: + message_id: str + parent_id: str | None = None + root_id: str | None = None + depth: int = 0 + session_id: str | None = None + gm_thrid: str | None = None