zcbot/tools/send_email.py

"""发邮件(host-side,DESIGN §8.5 投递第 2/3 层)。

- `send_email_smtp(...)`:纯函数,读 SMTP_* env 发信。SendEmailTool 与定时任务的
  确定性兜底投递(core/scheduler.py notify)共用它。
- `smtp_configured()`:agent_builder 据此决定挂不挂 tool(沿用"有 key 才注册"范式,
  §3.4);没配 SMTP 的部署里 agent 看不到一个永远报错的工具。
- `SendEmailTool`:agent 在(定时或交互)run 里调,附件路径强制落 user_root 内防越界。

密钥只在 host 进程读,绝不进沙箱 / run_python。env:
  SMTP_HOST  SMTP_PORT(默 465)  SMTP_USER  SMTP_PASSWORD
  SMTP_FROM(默 SMTP_USER)  SMTP_FROM_NAME(发件人显示名,默"总院科研辅助智能体")
  SMTP_TLS(ssl|starttls|none;默按端口:465→ssl 否则 starttls)
"""
from __future__ import annotations

import os
import smtplib
from email.message import EmailMessage
from email.utils import formataddr
from pathlib import Path
from typing import Iterable, Optional

from .base import Tool

_MAX_ATTACH_BYTES = 20 * 1024 * 1024   # 单封附件总上限,防把大产物塞爆 SMTP
_MAX_RECIPIENTS = 10


def smtp_configured() -> bool:
    """最小可发信集合是否齐全。"""
    return bool(
        os.getenv("SMTP_HOST", "").strip()
        and os.getenv("SMTP_USER", "").strip()
        and os.getenv("SMTP_PASSWORD", "").strip()
    )


def _tls_mode(port: int) -> str:
    mode = os.getenv("SMTP_TLS", "").strip().lower()
    if mode in ("ssl", "starttls", "none"):
        return mode
    return "ssl" if port == 465 else "starttls"


def send_email_smtp(
    to: Iterable[str] | str,
    subject: str,
    body: str,
    attachments: Optional[Iterable[Path]] = None,
    *,
    timeout: float = 30.0,
) -> None:
    """同步发一封纯文本邮件(可带附件)。失败抛异常,由调用方决定如何处理。

    host-side 调用(SendEmailTool 在 to_thread 的 run 线程里;scheduler 在
    run_in_executor 里)—— smtplib 是阻塞 IO,不要在 asyncio loop 直接 await。
    """
    if not smtp_configured():
        raise RuntimeError("SMTP 未配置(需 SMTP_HOST/SMTP_USER/SMTP_PASSWORD)")

    host = os.getenv("SMTP_HOST", "").strip()
    port = int(os.getenv("SMTP_PORT", "465").strip() or "465")
    user = os.getenv("SMTP_USER", "").strip()
    password = os.getenv("SMTP_PASSWORD", "").strip()
    sender = os.getenv("SMTP_FROM", "").strip() or user
    from_name = os.getenv("SMTP_FROM_NAME", "").strip() or "总院科研辅助智能体"

    if isinstance(to, str):
        to_list = [to]
    else:
        to_list = list(to)
    to_list = [a.strip() for a in to_list if a and a.strip()]
    if not to_list:
        raise ValueError("收件人为空")
    if len(to_list) > _MAX_RECIPIENTS:
        raise ValueError(f"收件人过多(上限 {_MAX_RECIPIENTS})")

    msg = EmailMessage()
    msg["From"] = formataddr((from_name, sender))
    msg["To"] = ", ".join(to_list)
    msg["Subject"] = subject or "(无主题)"
    msg.set_content(body or "")

    total = 0
    for p in attachments or []:
        p = Path(p)
        if not p.is_file():
            continue
        data = p.read_bytes()
        total += len(data)
        if total > _MAX_ATTACH_BYTES:
            raise ValueError(f"附件总大小超过 {_MAX_ATTACH_BYTES // (1024*1024)}MB")
        msg.add_attachment(
            data, maintype="application", subtype="octet-stream", filename=p.name
        )

    tls = _tls_mode(port)
    if tls == "ssl":
        with smtplib.SMTP_SSL(host, port, timeout=timeout) as smtp:
            smtp.login(user, password)
            smtp.send_message(msg)
    else:
        with smtplib.SMTP(host, port, timeout=timeout) as smtp:
            if tls == "starttls":
                smtp.starttls()
            smtp.login(user, password)
            smtp.send_message(msg)


class SendEmailTool(Tool):
    name = "send_email"
    description = (
        "Send an email (plain text, optional file attachments) via the server's configured "
        "SMTP account. Use this when the user asks to email a result/report to someone, or when "
        "a scheduled task's instruction says to email its output. Attachments are paths inside "
        "the working directory (e.g. 'report.docx' or '<wd>/figures/x.png'). Returns a short "
        "confirmation or an [Error] line."
    )
    parameters = {
        "type": "object",
        "properties": {
            "to": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Recipient email address(es).",
            },
            "subject": {"type": "string", "description": "Email subject line."},
            "body": {"type": "string", "description": "Plain-text email body."},
            "attachments": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Optional file paths (relative to working dir) to attach.",
            },
        },
        "required": ["to", "subject", "body"],
    }

    def execute(
        self,
        to: list[str] | str,
        subject: str,
        body: str,
        attachments: Optional[list[str]] = None,
    ) -> str:
        if isinstance(to, str):
            to = [to]
        recipients = [a.strip() for a in (to or []) if isinstance(a, str) and a.strip()]
        if not recipients:
            return "[Error] to 不能为空"

        resolved: list[Path] = []
        for raw in attachments or []:
            if not isinstance(raw, str) or not raw.strip():
                continue
            p = self._resolve(raw.strip()).resolve()
            # 附件强制落 user_root 内,防 ../ 读到别人/系统文件
            if self.user_root is not None:
                try:
                    p.relative_to(self.user_root.resolve())
                except ValueError:
                    return f"[Error] 附件路径越界(必须在工作目录内): {raw}"
            if not p.is_file():
                return f"[Error] 附件不存在: {raw}"
            resolved.append(p)

        try:
            send_email_smtp(recipients, subject, body, resolved)
        except Exception as e:
            return f"[Error] 发送失败: {type(e).__name__}: {e}"
        n = f"(含 {len(resolved)} 个附件)" if resolved else ""
        return f"[ok] 已发送给 {', '.join(recipients)} {n}".strip()