shirobana-bot/bot-utility/for-bot_mail.py

# -*- coding: utf-8 -*-
"""Akaros Mail（shirobana@akaros.si）REST 收发件模块。"""

from __future__ import annotations

import os
import re
from typing import Any

API_BASE = "https://api.mail.akaros.si"
API_KEY = os.environ.get("AKAROS_MAIL_API_KEY", "").strip()
MAIL_ADDRESS = os.environ.get("AKAROS_MAIL_ADDRESS", "shirobana@akaros.si").strip()
FROM_NAME = os.environ.get("AKAROS_MAIL_FROM_NAME", "白花").strip() or "白花"
TIMEOUT = (15, 90)

HELP_TEXT = (
    "Akaros 邮件\n"
    "━━━━━━━━━━━━\n"
    "在 chat 模式下，可以要求白花查看 Akaros 收件箱、读取邮件或发送邮件。\n"
    "发件地址固定为 shirobana@akaros.si；支持纯文本和可选 HTML 正文。"
)

_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+$")
_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,80}$")
_FOLDERS = {"inbox", "sent", "failed", "outbox", "trash", "all"}


class MailInputError(ValueError):
    pass


def _addresses(value, field="to", maximum=50) -> list[str]:
    values = [value] if isinstance(value, str) else value
    if not isinstance(values, list) or not values:
        raise MailInputError(f"{field} 必须是非空地址或地址数组")
    result = []
    for item in values:
        if not isinstance(item, str) or not _EMAIL_RE.fullmatch(item.strip()):
            raise MailInputError(f"{field} 含无效邮件地址")
        address = item.strip()
        if address not in result:
            result.append(address)
    if len(result) > maximum:
        raise MailInputError(f"{field} 最多 {maximum} 个地址")
    return result


def _optional_addresses(value, field):
    if value in (None, "", []):
        return []
    return _addresses(value, field)


def _text(value, field, maximum, required=False):
    if not isinstance(value, str) or (required and not value.strip()):
        raise MailInputError(f"{field} 参数无效")
    if "\x00" in value or len(value.encode("utf-8")) > maximum:
        raise MailInputError(f"{field} 超过大小限制")
    return value


def _message_id(value):
    if not isinstance(value, (str, int)) or not _ID_RE.fullmatch(str(value)):
        raise MailInputError("邮件 ID 无效")
    return str(value)


def _request(method: str, path: str, *, params=None, payload=None):
    import requests

    if not API_KEY:
        raise RuntimeError("AKAROS_MAIL_API_KEY 未配置")
    with requests.Session() as session:
        response = session.request(
            method,
            API_BASE + path,
            headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
            params=params,
            json=payload,
            timeout=TIMEOUT,
        )
    content_type = response.headers.get("Content-Type", "")
    try:
        data = response.json() if "json" in content_type.lower() else None
    except ValueError:
        data = None
    if response.status_code >= 400:
        detail = data.get("error") if isinstance(data, dict) else None
        raise RuntimeError(f"Akaros API HTTP {response.status_code}: {detail or '请求失败'}")
    return data if data is not None else {}


def send_email(to, subject, text, html=None, cc=None, bcc=None):
    payload = {
        "from": MAIL_ADDRESS,
        "from_name": FROM_NAME,
        "to": _addresses(to),
        "cc": _optional_addresses(cc, "cc"),
        "bcc": _optional_addresses(bcc, "bcc"),
        "subject": _text(subject, "subject", 500 * 1024, required=True),
        "text": _text(text, "text", 2 * 1024 * 1024, required=True),
    }
    if html is not None:
        payload["html"] = _text(html, "html", 2 * 1024 * 1024)
    return _request("POST", "/api/send", payload=payload)


def list_messages(folder="inbox", limit=10, query=None, direction=None):
    if folder not in _FOLDERS:
        raise MailInputError("folder 无效")
    if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 100:
        raise MailInputError("limit 必须是 1..100 的整数")
    params = {"folder": folder, "limit": limit}
    if query is not None:
        params["q"] = _text(query, "query", 200, required=True)
    if direction is not None:
        if direction not in ("in", "out"):
            raise MailInputError("direction 无效")
        params["direction"] = direction
    return _request("GET", "/api/messages", params=params)


def get_message(message_id):
    return _request("GET", "/api/messages/" + _message_id(message_id))


def format_messages(data) -> str:
    items = data.get("messages", data) if isinstance(data, dict) else data
    if not isinstance(items, list) or not items:
        return "没有找到邮件"
    lines = []
    for item in items:
        if not isinstance(item, dict):
            continue
        lines.append(
            f"#{item.get('id', '?')}｜{item.get('subject') or '(无主题)'}｜"
            f"{item.get('from') or item.get('sender') or '未知发件人'}｜"
            f"{item.get('created_at') or item.get('date') or '时间未知'}"
        )
    return "\n".join(lines) if lines else "没有找到邮件"


def format_message(data) -> str:
    if not isinstance(data, dict):
        return str(data)
    subject = data.get("subject") or "(无主题)"
    sender = data.get("from") or data.get("sender") or "未知发件人"
    body = data.get("text") or data.get("text_body") or data.get("body") or "(无纯文本正文)"
    return f"主题：{subject}\n发件人：{sender}\n\n{body}"