RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
1055 lines
41 KiB
Python
1055 lines
41 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Bot 后端邮件模块 mail.py
|
||
|
||
- 通过 hMailServer COM 管理员接口读写邮件（无需用户邮箱密码，读写发一体）
|
||
- 每个 QQ 一个 JSON 配置文件（mail/<qq>.json），存放绑定的邮件地址与推送配置
|
||
- 内置一个 HTTP 绑定确认端点（Flask + waitress，127.0.0.1:9088），
|
||
由 nginx 的 /bot/mail/ 转发到 api.rrrosmontis.icu
|
||
"""
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import stat
|
||
import html as _html
|
||
import queue
|
||
import time
|
||
import shlex
|
||
import threading
|
||
import secrets
|
||
import email
|
||
from email.utils import parsedate_to_datetime
|
||
from datetime import datetime
|
||
|
||
from flask import Flask, request
|
||
|
||
import identity
|
||
|
||
# ---------------- 配置 ----------------
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
MAIL_DIR = os.path.join(BASE_DIR, "mail")
|
||
|
||
HMAIL_ADMIN_PASSWORD = None # Administrator secret stays on the Windows mail server.
|
||
ALLOWED_DOMAINS = ("aeolianchi.me", "rrrosmontis.icu")
|
||
NOREPLY_ADDRESS = "noreply@rrrosmontis.icu"
|
||
NOREPLY_NAME = "Shirobana Utility Bot"
|
||
|
||
CONFIRM_BASE_URL = "https://api.rrrosmontis.icu/bot/mail" # 确认链接基础域名
|
||
HTTP_HOST = "127.0.0.1"
|
||
HTTP_PORT = 9088
|
||
|
||
TOKEN_EXPIRE = 1800 # 绑定确认 token 有效期（秒）
|
||
LIST_MAX = 10 # 单次最多返回邮件数
|
||
INFO_MAX_CHARS = 2000 # 正文单次最大字符数
|
||
DEFAULT_POLL_MINUTES = 5 # 默认轮询间隔（分钟）
|
||
|
||
|
||
def _positive_env_int(name, default):
|
||
try:
|
||
value = int(os.environ.get(name, default))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
return value if value > 0 else default
|
||
|
||
|
||
# 附件限制（字节）。桥接端会独立重复校验，不能只信任客户端。
|
||
MAX_ATTACHMENT_COUNT = _positive_env_int("MAIL_ATTACHMENT_MAX_COUNT", 10)
|
||
MAX_ATTACHMENT_BYTES = _positive_env_int(
|
||
"MAIL_ATTACHMENT_MAX_FILE_BYTES", 20 * 1024 * 1024)
|
||
MAX_ATTACHMENT_TOTAL_BYTES = _positive_env_int(
|
||
"MAIL_ATTACHMENT_MAX_TOTAL_BYTES", 25 * 1024 * 1024)
|
||
_ATTACHMENT_ADDRESS_RE = re.compile(r"[^\s<>@,;]+@[^\s<>@,;]+")
|
||
|
||
HELP_TEXT = (
|
||
"邮件指令\n"
|
||
"━━━━━━━━━━━━\n"
|
||
".mail help\n"
|
||
"- 显示本帮助\n"
|
||
".mail bind <地址>\n"
|
||
"- 绑定邮箱（仅 @rrrosmontis.icu / @aeolianchi.me）\n"
|
||
".mail unbind <地址>\n"
|
||
"- 解绑邮箱\n"
|
||
".mail account\n"
|
||
"- 查看已绑定邮箱\n"
|
||
".mail default <地址>\n"
|
||
"- 设为默认邮箱\n"
|
||
".mail list <N|N-d|N-h|N-m> [地址]\n"
|
||
"- 最近 N 封 / N 天 / N 小时 / N 分钟内的邮件（标题+id，最多10封）\n"
|
||
".mail info <id>\n"
|
||
"- 查看某封邮件正文\n"
|
||
".mail push <on|off> [分钟]\n"
|
||
"- 开关新邮件推送并设置轮询间隔\n"
|
||
".mail new <收件人> <\"主题\"> <\"正文\"> [地址]\n"
|
||
"- 发信（主题/正文需加引号）\n"
|
||
"━━━━━━━━━━━━\n"
|
||
"邮件支持：https://mail.rrrosmontis.icu"
|
||
)
|
||
|
||
# ---------------- 跨线程状态 ----------------
|
||
# 跨线程 -> asyncio 的通知队列（结构与 live_monitor 一致）
|
||
notification_queue = queue.Queue()
|
||
# 当前 NapCat WebSocket 连接（由 main.py 在连接建立时赋值）
|
||
websocket = None
|
||
|
||
# 绑定确认 token（内存态）：token -> {qq, email, created}
|
||
_pending_tokens = {}
|
||
_pending_lock = threading.Lock()
|
||
|
||
# JSON 文件锁（RLock：_mutate_user 临界区内会嵌套调 _save_user）
|
||
# 所有 读→改→写 必须整体在锁内完成（见 _mutate_user），否则事件循环指令、
|
||
# waitress confirm 线程、push 轮询线程三方并发会互相覆盖丢更新。
|
||
_store_lock = threading.RLock()
|
||
|
||
# 通知消费者启动宽限（秒）：等通道注册/NapCat 反向连接就绪再开始投递，
|
||
# 避免进程启动窗口内投递失败导致新邮件水位线无法推进的反复重投。
|
||
STARTUP_GRACE = 30
|
||
|
||
# 已入队待投递的新邮件 (uid, addr, mid)：水位线在投递成功后才推进，
|
||
# 期间同一封邮件不得重复入队（消费者挂掉时靠这个防队列无限堆积）。
|
||
_inflight = set()
|
||
_inflight_lock = threading.Lock()
|
||
|
||
# push 轮询：每个 qq 的下次检查时间（内存态，单位秒 epoch）
|
||
_poll_next = {}
|
||
|
||
app = Flask(__name__)
|
||
|
||
|
||
# ==================== COM 基础设施 ====================
|
||
class _ComSession:
|
||
"""HTTPS session to the Windows COM bridge; message IDs stay unchanged."""
|
||
def __enter__(self):
|
||
import requests
|
||
from pathlib import Path
|
||
from urllib.parse import urlsplit
|
||
self.base = os.environ.get("MAIL_COM_URL", "https://api.rrrosmontis.icu/v1/mail/api/admin/com").rstrip("/")
|
||
url = urlsplit(self.base)
|
||
if url.scheme != "https" and not (url.scheme == "http" and url.hostname in ("127.0.0.1", "localhost")):
|
||
raise RuntimeError("邮件 COM 接口必须使用 HTTPS")
|
||
token = Path(os.environ.get("MAIL_COM_TOKEN_FILE", os.path.join(BASE_DIR, "mail-com-token"))).read_text().strip()
|
||
if len(token) < 64: raise RuntimeError("邮件 COM Token 未配置")
|
||
self.session = requests.Session()
|
||
self.session.trust_env = False
|
||
self.session.headers["Authorization"] = "Bearer " + token
|
||
return self
|
||
def post(self, endpoint, payload):
|
||
response = self.session.post(self.base + endpoint, json=payload, timeout=(8, 30))
|
||
response.raise_for_status()
|
||
return response.json()
|
||
def __exit__(self, *exc):
|
||
self.session.close()
|
||
return False
|
||
|
||
|
||
|
||
def _run_com(fn):
|
||
"""在独立线程语义下执行一个 COM 操作（fn 接收已认证的 app）。"""
|
||
with _ComSession() as app:
|
||
return fn(app)
|
||
|
||
|
||
# ==================== JSON 存储 ====================
|
||
def _user_file(qq):
|
||
return os.path.join(MAIL_DIR, f"{int(qq)}.json")
|
||
|
||
|
||
def _load_user(qq):
|
||
default = {
|
||
"addresses": [],
|
||
"default": "",
|
||
"push": {"enabled": False, "interval_minutes": DEFAULT_POLL_MINUTES, "last_seen": {}},
|
||
}
|
||
path = _user_file(qq)
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except (OSError, json.JSONDecodeError):
|
||
return default
|
||
if not isinstance(data, dict):
|
||
return default
|
||
data.setdefault("addresses", [])
|
||
data.setdefault("default", "")
|
||
data.setdefault("push", {"enabled": False, "interval_minutes": DEFAULT_POLL_MINUTES, "last_seen": {}})
|
||
data.setdefault("channel", "qq")
|
||
return data
|
||
|
||
|
||
def _save_user(qq, data):
|
||
os.makedirs(MAIL_DIR, exist_ok=True)
|
||
path = _user_file(qq)
|
||
tmp = path + ".tmp"
|
||
with _store_lock:
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
# Windows 上 os.replace 与并发读者（_load_user 不持锁）可抛 PermissionError，
|
||
# 短暂重试即可自愈；彻底失败时 tmp 残留无害（下次写入覆盖）
|
||
for attempt in range(5):
|
||
try:
|
||
os.replace(tmp, path)
|
||
break
|
||
except PermissionError:
|
||
if attempt == 4:
|
||
raise
|
||
time.sleep(0.05 * (attempt + 1))
|
||
|
||
|
||
def _mutate_user(qq, fn):
|
||
"""在 _store_lock 保护下完成 读→改→写 整体（防跨线程丢更新）。
|
||
|
||
fn(data) 在锁内执行，其返回值为 _mutate_user 的返回值；fn 返回后无条件落盘。
|
||
锁内只有纯内存操作 + 一次小文件写，事件循环/HTTP 线程/轮询线程都可安全使用。
|
||
"""
|
||
with _store_lock:
|
||
data = _load_user(qq)
|
||
result = fn(data)
|
||
_save_user(qq, data)
|
||
return result
|
||
|
||
|
||
def _advance_last_seen(qq, addr, mid):
|
||
"""新邮件水位线推进：仅在投递成功后由 notification_loop 调用。
|
||
单调递增，绝不回退（同一封邮件可能因重试被多次确认）。"""
|
||
def _bump(data):
|
||
last_seen = data["push"].setdefault("last_seen", {})
|
||
if int(last_seen.get(addr) or 0) < int(mid):
|
||
last_seen[addr] = int(mid)
|
||
try:
|
||
_mutate_user(qq, _bump)
|
||
except Exception as e:
|
||
print(f"[mail] 水位线推进失败 uid={qq} {addr}#{mid}: {e}")
|
||
|
||
|
||
# ==================== 基础工具 ====================
|
||
def _normalize_address(addr):
|
||
return (addr or "").strip().lower()
|
||
|
||
|
||
def _valid_domain(addr):
|
||
try:
|
||
local, dom = addr.split("@", 1)
|
||
except ValueError:
|
||
return False
|
||
return bool(local) and dom in ALLOWED_DOMAINS
|
||
|
||
|
||
def _mail_changes(app, address, after):
|
||
return app.post("/changes", {"address": address, "after": after, "limit": LIST_MAX})
|
||
|
||
|
||
def _message_ts(m):
|
||
for attr in ("InternalDate", "Date"):
|
||
try:
|
||
v = getattr(m, attr)
|
||
except Exception:
|
||
continue
|
||
if v is None:
|
||
continue
|
||
if isinstance(v, str):
|
||
try:
|
||
return parsedate_to_datetime(v).timestamp()
|
||
except Exception:
|
||
continue
|
||
else: # datetime-like
|
||
try:
|
||
return v.timestamp()
|
||
except Exception:
|
||
continue
|
||
return 0.0
|
||
|
||
|
||
def _parse_window(spec):
|
||
"""'10' -> ('latest', 10)；'1-d'/'2-h'/'30-m' -> ('window', 秒数)。非法返回 None。"""
|
||
m = re.fullmatch(r"(\d+)(?:-([dhm]))?", (spec or "").strip().lower())
|
||
if not m:
|
||
return None
|
||
n = int(m.group(1))
|
||
unit = m.group(2)
|
||
if unit is None:
|
||
return ("latest", n)
|
||
mult = {"d": 86400, "h": 3600, "m": 60}[unit]
|
||
return ("window", n * mult)
|
||
|
||
|
||
def _fmt_time(ts):
|
||
try:
|
||
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _split_sender(from_str, from_addr):
|
||
"""把 From 头拆成 (显示名, 邮箱地址)。优先用 From 头里的地址（转发邮件时 FromAddress 会是转发地址，不可靠）。"""
|
||
raw = (from_str or "").strip()
|
||
name = ""
|
||
addr = ""
|
||
m = re.match(r"^(.*)<\s*([^>]*)\s*>$", raw)
|
||
if m:
|
||
name = m.group(1).strip().strip('"').strip()
|
||
addr = m.group(2).strip()
|
||
else:
|
||
if "@" in raw:
|
||
addr = raw
|
||
else:
|
||
name = raw
|
||
if not addr:
|
||
addr = (from_addr or "").strip()
|
||
return (name or ""), (addr or "")
|
||
|
||
|
||
_TAG_RE = re.compile(r"<[^>]+>")
|
||
|
||
|
||
def _html_to_text(html_text):
|
||
"""安全地把 HTML 转纯文本：剥离 script/style/iframe/图片等，不渲染任何结构。"""
|
||
s = html_text or ""
|
||
# 去掉整块 script/style/iframe/head 等
|
||
s = re.sub(r"(?is)<(script|style|iframe|head|title|noscript)\b[^>]*>.*?</\1>", " ", s)
|
||
# 去掉自闭合/无闭合的 script/style/iframe
|
||
s = re.sub(r"(?is)<(script|style|iframe)\b[^>]*/?>", " ", s)
|
||
# 去掉图片等非文本标签
|
||
s = re.sub(r"(?is)<(img|br|hr)\b[^>]*/?>", "\n", s)
|
||
# 去掉剩余所有标签
|
||
s = _TAG_RE.sub(" ", s)
|
||
s = _html.unescape(s)
|
||
# 压缩空白
|
||
s = re.sub(r"[ \t\r\f\v]+", " ", s)
|
||
s = re.sub(r"\n\s*\n+", "\n", s)
|
||
return s.strip()
|
||
|
||
|
||
def _extract_body(m):
|
||
html_body = m.HTMLBody or ""
|
||
plain = m.Body or ""
|
||
if html_body.strip():
|
||
return _html_to_text(html_body)
|
||
return plain.strip()
|
||
|
||
|
||
def _decode_mime(s):
|
||
if not s:
|
||
return ""
|
||
try:
|
||
from email.header import decode_header
|
||
parts = decode_header(s)
|
||
out = []
|
||
for p, cs in parts:
|
||
if isinstance(p, bytes):
|
||
out.append(p.decode(cs or "utf-8", "replace"))
|
||
else:
|
||
out.append(p)
|
||
return "".join(out)
|
||
except Exception:
|
||
return s
|
||
|
||
|
||
def _analyze_message(m):
|
||
return m.image_count, m.attachment_names
|
||
|
||
|
||
def resolve_context_reference(uid, text):
|
||
"""Resolve 'this mail' from the latest structured mail event in chat history.
|
||
|
||
Explicit numeric IDs always win. Ambiguous or missing references are left to
|
||
the model's clarification path rather than guessed.
|
||
"""
|
||
if not isinstance(text, str) or re.search(r"\d{3,}", text):
|
||
return text
|
||
phrases = ("这篇邮件", "这封邮件", "刚才那封邮件", "刚才的邮件")
|
||
if not any(phrase in text for phrase in phrases):
|
||
return text
|
||
path = os.path.join(BASE_DIR, "chat", "history", f"{int(uid)}.txt")
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
lines = fh.readlines()
|
||
except OSError:
|
||
return text
|
||
mids = []
|
||
for line in reversed(lines):
|
||
match = re.search(r"\[event:mail:(\d+)\]", line)
|
||
if match:
|
||
mids.append(match.group(1))
|
||
break
|
||
if not mids:
|
||
return text
|
||
mid = mids[0]
|
||
for phrase in phrases:
|
||
if phrase in text:
|
||
return text.replace(phrase, f"编号为 {mid} 的邮件", 1)
|
||
return text
|
||
|
||
|
||
# ==================== COM 邮件操作 ====================
|
||
def _list_inbox_messages(app, address, window_seconds=None, limit=LIST_MAX):
|
||
return app.post("/inbox", {"address": address, "window_seconds": window_seconds, "limit": limit})["items"]
|
||
|
||
|
||
def _find_message_by_id(app, addresses, msg_id):
|
||
from types import SimpleNamespace
|
||
data = app.post("/message", {"addresses": addresses, "id": int(msg_id)})["message"]
|
||
return SimpleNamespace(**data) if data is not None else None
|
||
|
||
|
||
def _send_mail(app, from_address, to_address, subject, body_text):
|
||
app.post("/send", dict(from_address=from_address, to_address=to_address, subject=subject, body_text=body_text))
|
||
|
||
|
||
def _save_to_sent(app, from_address, to_address, subject, body_text):
|
||
try:
|
||
app.post("/sent", dict(from_address=from_address, to_address=to_address, subject=subject, body_text=body_text))
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
class _AttachmentInputError(ValueError):
|
||
"""可向工具调用方展示的安全输入错误，不包含本地路径或底层异常。"""
|
||
|
||
def __init__(self, code, message):
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.message = message
|
||
|
||
|
||
def _attachment_filename(value):
|
||
if not value or len(value) > 255 or value != value.strip() or value in (".", ".."):
|
||
raise _AttachmentInputError("invalid_filename", "附件文件名无效")
|
||
if len(value.encode("utf-8")) > 1024:
|
||
raise _AttachmentInputError("invalid_filename", "附件文件名无效")
|
||
if any(ord(ch) < 32 or ord(ch) == 127 or ch in '<>:"/\\|?*' for ch in value):
|
||
raise _AttachmentInputError("invalid_filename", "附件文件名不能包含路径或特殊字符")
|
||
return value
|
||
|
||
|
||
def _load_attachment_payloads(attachment_paths):
|
||
"""在线程中读取调用方文件；只允许普通文件并限制实际读取字节数。"""
|
||
attachments, names, total = [], [], 0
|
||
for raw_path in attachment_paths:
|
||
try:
|
||
path = os.fspath(raw_path)
|
||
except TypeError:
|
||
raise _AttachmentInputError("invalid_path", "附件路径类型无效")
|
||
if not isinstance(path, str) or not path:
|
||
raise _AttachmentInputError("invalid_path", "附件路径无效")
|
||
try:
|
||
# 先 stat 避免直接打开目录/设备，再用 fstat 防止打开过程中的替换。
|
||
if not stat.S_ISREG(os.stat(path).st_mode):
|
||
raise _AttachmentInputError("not_regular_file", "附件必须是普通文件")
|
||
with open(path, "rb") as fh:
|
||
if not stat.S_ISREG(os.fstat(fh.fileno()).st_mode):
|
||
raise _AttachmentInputError("not_regular_file", "附件必须是普通文件")
|
||
content = fh.read(MAX_ATTACHMENT_BYTES + 1)
|
||
except _AttachmentInputError:
|
||
raise
|
||
except (OSError, ValueError):
|
||
raise _AttachmentInputError("unreadable_file", "附件不存在或无法读取")
|
||
if len(content) > MAX_ATTACHMENT_BYTES:
|
||
raise _AttachmentInputError("file_too_large", "单个附件超过大小限制")
|
||
total += len(content)
|
||
if total > MAX_ATTACHMENT_TOTAL_BYTES:
|
||
raise _AttachmentInputError("total_too_large", "附件总大小超过限制")
|
||
filename = _attachment_filename(os.path.basename(path))
|
||
content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
|
||
attachments.append({
|
||
"filename": filename,
|
||
"content_type": content_type,
|
||
"data_base64": base64.b64encode(content).decode("ascii"),
|
||
})
|
||
names.append(filename)
|
||
return attachments, names, total
|
||
|
||
|
||
async def send_with_attachments(user_id, from_address, to_address, subject,
|
||
body_text, attachment_paths):
|
||
"""供工具层调用的附件发信 API；所有失败均返回安全的结构化结果。
|
||
|
||
成功：{ok, from, to, subject, attachment_names, attachment_count, size,
|
||
sent_saved}；失败额外含 error_code/error，不暴露 Token、底层响应
|
||
或本地文件路径。size 为解码后的附件总字节数。
|
||
"""
|
||
from_addr = _normalize_address(from_address) if isinstance(from_address, str) else ""
|
||
to_addr = _normalize_address(to_address) if isinstance(to_address, str) else ""
|
||
result = {"from": from_addr, "to": to_addr,
|
||
"attachment_names": [], "size": 0}
|
||
try:
|
||
data = _load_user(user_id)
|
||
bound = {_normalize_address(a) for a in data.get("addresses", [])
|
||
if isinstance(a, str)}
|
||
if not from_addr or not _valid_domain(from_addr) or from_addr not in bound:
|
||
raise _AttachmentInputError("from_not_bound", "发件地址未绑定到当前用户")
|
||
if not to_addr or not _ATTACHMENT_ADDRESS_RE.fullmatch(to_addr):
|
||
raise _AttachmentInputError("invalid_recipient", "收件地址无效")
|
||
if (not isinstance(subject, str) or not subject.strip() or
|
||
len(subject) > 2000 or "\r" in subject or "\n" in subject):
|
||
raise _AttachmentInputError("invalid_subject", "主题为空、过长或包含换行")
|
||
if not isinstance(body_text, str) or len(body_text) > 200000:
|
||
raise _AttachmentInputError("invalid_body", "正文类型无效或过长")
|
||
if not isinstance(attachment_paths, (list, tuple)):
|
||
raise _AttachmentInputError("invalid_attachments", "附件路径必须是列表")
|
||
if len(attachment_paths) > MAX_ATTACHMENT_COUNT:
|
||
raise _AttachmentInputError("too_many_attachments", "附件数量超过限制")
|
||
|
||
attachments, names, total = await asyncio.to_thread(
|
||
_load_attachment_payloads, attachment_paths)
|
||
result.update(attachment_names=names, size=total)
|
||
payload = {
|
||
"from_address": from_addr,
|
||
"to_address": to_addr,
|
||
"subject": subject,
|
||
"body_text": body_text,
|
||
"attachments": attachments,
|
||
}
|
||
|
||
def _do():
|
||
with _ComSession() as app:
|
||
return app.post("/send_with_attachments", payload)
|
||
|
||
response = await asyncio.to_thread(_do)
|
||
if not isinstance(response, dict) or not (
|
||
response.get("accepted") is True or response.get("ok") is True):
|
||
result.update(ok=False, error_code="send_failed",
|
||
error="附件邮件发送失败，请检查发信配置")
|
||
return result
|
||
result.update(ok=True, subject=subject, attachment_count=len(names),
|
||
sent_saved=response.get("sent_saved"))
|
||
return result
|
||
except _AttachmentInputError as exc:
|
||
result.update(ok=False, error_code=exc.code, error=exc.message)
|
||
return result
|
||
except Exception as exc:
|
||
# 网络/JSON/COM 异常可能含 URL、路径或服务器细节，只记录类型。
|
||
print(f"[mail] 附件邮件发送失败: {type(exc).__name__}")
|
||
result.update(ok=False, error_code="send_unknown",
|
||
error="附件邮件发送结果未知，请勿自动重试；请先检查是否已送达")
|
||
return result
|
||
|
||
|
||
def _send_bind_mail(to_addr, body):
|
||
"""用 noreply 账号向被绑定地址发送确认邮件。"""
|
||
try:
|
||
def _do(app):
|
||
_send_mail(app, NOREPLY_ADDRESS, to_addr, "邮箱绑定确认", body)
|
||
_run_com(_do)
|
||
return True, None
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
# ==================== 命令处理 ====================
|
||
async def _reply(ws, msg_type, target_id, content):
|
||
"""ws 为 Channel 对象（通道抽象）"""
|
||
from channel import send_message
|
||
await send_message(ws, msg_type, target_id, content)
|
||
|
||
|
||
async def _cmd_bind(ws, msg_type, target_id, user_id, arg):
|
||
addr = _normalize_address(arg)
|
||
if not addr:
|
||
await _reply(ws, msg_type, target_id, "用法: .mail bind <邮件地址>")
|
||
return
|
||
if not _valid_domain(addr):
|
||
await _reply(ws, msg_type, target_id, "仅支持绑定 @rrrosmontis.icu / @aeolianchi.me 域名")
|
||
return
|
||
data = _mutate_user(user_id, lambda d: d) # 确保用户文件存在并返回配置（锁内读写）
|
||
if addr in data["addresses"]:
|
||
await _reply(ws, msg_type, target_id, f"{addr} 已绑定，无需重复绑定")
|
||
return
|
||
|
||
token = secrets.token_urlsafe(24)
|
||
with _pending_lock:
|
||
_pending_tokens[token] = {"qq": int(user_id), "email": addr, "created": time.time(),
|
||
"channel": getattr(ws, "name", "qq")}
|
||
confirm_url = f"{CONFIRM_BASE_URL}/confirm?token={token}"
|
||
body = (
|
||
f"当前邮件账户正在被用户 {user_id} 绑定至本站 Qbot (Shirobana Utility Bot) 提供的邮件服务。\n"
|
||
"如果这是您本人的操作，请点击以下链接确认：\n"
|
||
"\n"
|
||
f"{confirm_url}\n"
|
||
"\n"
|
||
"如果这不是您本人的操作，请忽略此消息。\n"
|
||
"如果您的邮箱可能被他人错误地绑定，请联系 support@rrrosmontis.icu 提交申请执行解绑。\n"
|
||
"\n"
|
||
"该链接 30 分钟内有效。\n"
|
||
"绑定邮件涉及您的隐私，请不要将链接泄露给任何其他人。"
|
||
)
|
||
ok, err = await asyncio.to_thread(_send_bind_mail, addr, body)
|
||
if not ok:
|
||
with _pending_lock:
|
||
_pending_tokens.pop(token, None)
|
||
await _reply(ws, msg_type, target_id, f"发送确认邮件失败：{err}")
|
||
return
|
||
await _reply(ws, msg_type, target_id,
|
||
f"已向 {addr} 发送绑定确认邮件，请前往 mail.rrrosmontis.icu 登录并点击邮件中的确认链接完成绑定。")
|
||
|
||
|
||
async def _cmd_unbind(ws, msg_type, target_id, user_id, arg):
|
||
addr = _normalize_address(arg)
|
||
if not addr:
|
||
await _reply(ws, msg_type, target_id, "用法: .mail unbind <邮件地址>")
|
||
return
|
||
|
||
def _do_unbind(data):
|
||
if addr not in data["addresses"]:
|
||
return False
|
||
data["addresses"].remove(addr)
|
||
if data["default"] == addr:
|
||
data["default"] = data["addresses"][0] if data["addresses"] else ""
|
||
data["push"].get("last_seen", {}).pop(addr, None)
|
||
return True
|
||
|
||
if not _mutate_user(user_id, _do_unbind):
|
||
await _reply(ws, msg_type, target_id, f"未绑定 {addr}")
|
||
return
|
||
# 吊销该邮箱所有待确认 token：否则解绑后 30 分钟内点旧确认链接会被静默重新绑定
|
||
with _pending_lock:
|
||
stale = [t for t, it in _pending_tokens.items()
|
||
if it.get("qq") == int(user_id) and it.get("email") == addr]
|
||
for t in stale:
|
||
_pending_tokens.pop(t, None)
|
||
await _reply(ws, msg_type, target_id, f"已解绑 {addr}")
|
||
|
||
|
||
async def _cmd_account(ws, msg_type, target_id, user_id):
|
||
data = _load_user(user_id)
|
||
if not data["addresses"]:
|
||
await _reply(ws, msg_type, target_id, "尚未绑定任何邮箱，使用 .mail bind <地址> 绑定")
|
||
return
|
||
lines = ["已绑定邮箱："]
|
||
for a in data["addresses"]:
|
||
mark = "（默认）" if a == data["default"] else ""
|
||
lines.append(f"• {a}{mark}")
|
||
await _reply(ws, msg_type, target_id, "\n".join(lines))
|
||
|
||
|
||
async def _cmd_default(ws, msg_type, target_id, user_id, arg):
|
||
addr = _normalize_address(arg)
|
||
if not addr:
|
||
await _reply(ws, msg_type, target_id, "用法: .mail default <邮件地址>")
|
||
return
|
||
def _do_default(data):
|
||
if addr not in data["addresses"]:
|
||
return False
|
||
data["default"] = addr
|
||
return True
|
||
|
||
if not _mutate_user(user_id, _do_default):
|
||
await _reply(ws, msg_type, target_id, f"{addr} 未绑定，无法设为默认")
|
||
return
|
||
await _reply(ws, msg_type, target_id, f"默认邮箱已设为 {addr}")
|
||
|
||
|
||
async def _cmd_list(ws, msg_type, target_id, user_id, args):
|
||
if not args:
|
||
await _reply(ws, msg_type, target_id, "用法: .mail list <N|N-d|N-h|N-m> [地址]，如 .mail list 10 或 .mail list 1-d")
|
||
return
|
||
spec = args[0]
|
||
parsed = _parse_window(spec)
|
||
if not parsed:
|
||
await _reply(ws, msg_type, target_id, "时间参数格式错误，如 10 / 1-d / 2-h / 30-m")
|
||
return
|
||
kind, val = parsed
|
||
|
||
data = _load_user(user_id)
|
||
addr = _normalize_address(args[1]) if len(args) > 1 else data["default"]
|
||
if not addr:
|
||
await _reply(ws, msg_type, target_id, "尚未绑定邮箱或未设默认地址")
|
||
return
|
||
if addr not in data["addresses"]:
|
||
await _reply(ws, msg_type, target_id, f"未绑定 {addr}")
|
||
return
|
||
|
||
window_seconds = val if kind == "window" else None
|
||
limit = min(val, LIST_MAX) if kind == "latest" else LIST_MAX
|
||
|
||
def _do():
|
||
with _ComSession() as app:
|
||
return _list_inbox_messages(app, addr, window_seconds=window_seconds, limit=limit)
|
||
|
||
try:
|
||
items = await asyncio.to_thread(_do)
|
||
except Exception as e:
|
||
await _reply(ws, msg_type, target_id, f"读取邮件失败：{e}")
|
||
return
|
||
|
||
if not items:
|
||
await _reply(ws, msg_type, target_id, f"{addr} 没有符合条件的邮件")
|
||
return
|
||
lines = [f"{addr} 最近邮件："]
|
||
for it in items:
|
||
lines.append("==========")
|
||
lines.append(f"#{it['id']} {_fmt_time(it.get('ts', 0))}")
|
||
name, from_addr = _split_sender(it.get("from"), it.get("from_address"))
|
||
if not from_addr:
|
||
sender_part = name or "(未知)"
|
||
elif name:
|
||
sender_part = f"{name} <{from_addr}>"
|
||
else:
|
||
sender_part = f"<{from_addr}>"
|
||
lines.append(sender_part)
|
||
lines.append(f"【{it['subject']}】")
|
||
lines.append("==========")
|
||
lines.append("前往 https://mail.rrrosmontis.icu 查看邮件列表")
|
||
await _reply(ws, msg_type, target_id, "\n".join(lines))
|
||
|
||
|
||
async def _cmd_info(ws, msg_type, target_id, user_id, args):
|
||
if not args:
|
||
await _reply(ws, msg_type, target_id, "用法: .mail info <id>")
|
||
return
|
||
try:
|
||
msg_id = int(args[0])
|
||
except ValueError:
|
||
await _reply(ws, msg_type, target_id, "id 必须为数字")
|
||
return
|
||
data = _load_user(user_id)
|
||
if not data["addresses"]:
|
||
await _reply(ws, msg_type, target_id, "尚未绑定邮箱")
|
||
return
|
||
|
||
def _do():
|
||
with _ComSession() as app:
|
||
m = _find_message_by_id(app, data["addresses"], msg_id)
|
||
if m is None:
|
||
return None
|
||
image_count, attachment_names = _analyze_message(m)
|
||
return {
|
||
"subject": (m.Subject or "").strip(),
|
||
"from": (m.From or "").strip(),
|
||
"date": m.Date or "",
|
||
"body": _extract_body(m),
|
||
"image_count": image_count,
|
||
"attachment_names": attachment_names,
|
||
}
|
||
|
||
try:
|
||
info = await asyncio.to_thread(_do)
|
||
except Exception as e:
|
||
await _reply(ws, msg_type, target_id, f"读取邮件失败：{e}")
|
||
return
|
||
if info is None:
|
||
await _reply(ws, msg_type, target_id, f"未找到 id 为 {msg_id} 的邮件")
|
||
return
|
||
body = info["body"]
|
||
if len(body) > INFO_MAX_CHARS:
|
||
body = body[:INFO_MAX_CHARS] + "\n...(已截断)"
|
||
|
||
lines = [f"主题：{info['subject']}", f"发件人：{info['from']}", f"时间：{info['date']}"]
|
||
if info["image_count"]:
|
||
lines.append(f"[图片 ×{info['image_count']}]" if info["image_count"] > 1 else "[图片]")
|
||
if info["attachment_names"]:
|
||
lines.append("[附件占位: " + ", ".join(info["attachment_names"]) + "]")
|
||
if body:
|
||
lines.append("")
|
||
lines.append(body)
|
||
lines.append("")
|
||
lines.append("前往 https://mail.rrrosmontis.icu 阅读原文")
|
||
await _reply(ws, msg_type, target_id, "\n".join(lines))
|
||
|
||
|
||
async def _cmd_push(ws, msg_type, target_id, user_id, args):
|
||
if not args or args[0].lower() not in ("on", "off"):
|
||
await _reply(ws, msg_type, target_id, "用法: .mail push <on|off> [分钟]，如 .mail push on 5")
|
||
return
|
||
sub = args[0].lower()
|
||
if sub == "on":
|
||
interval = DEFAULT_POLL_MINUTES
|
||
if len(args) > 1:
|
||
try:
|
||
interval = int(args[1])
|
||
except ValueError:
|
||
await _reply(ws, msg_type, target_id, "轮询间隔必须为整数分钟")
|
||
return
|
||
if interval < 1:
|
||
interval = 1
|
||
|
||
def _do_on(data):
|
||
data["push"]["enabled"] = True
|
||
data["push"]["interval_minutes"] = interval
|
||
|
||
_mutate_user(user_id, _do_on)
|
||
_poll_next.pop(int(user_id), None) # 立即生效
|
||
await _reply(ws, msg_type, target_id, f"已开启新邮件推送，每 {interval} 分钟轮询一次")
|
||
else:
|
||
_mutate_user(user_id, lambda data: data["push"].update(enabled=False))
|
||
_poll_next.pop(int(user_id), None)
|
||
await _reply(ws, msg_type, target_id, "已关闭新邮件推送")
|
||
|
||
|
||
async def _cmd_new(ws, msg_type, target_id, user_id, args):
|
||
if len(args) < 3:
|
||
await _reply(ws, msg_type, target_id,
|
||
'用法: .mail new <收件人> <"主题"> <"正文"> [地址]，主题和正文需加引号')
|
||
return
|
||
to_addr, subject, body = args[0], args[1], args[2]
|
||
data = _load_user(user_id)
|
||
addr = _normalize_address(args[3]) if len(args) >= 4 else data["default"]
|
||
if not addr:
|
||
await _reply(ws, msg_type, target_id, "尚未绑定邮箱或未设默认地址")
|
||
return
|
||
if addr not in data["addresses"]:
|
||
await _reply(ws, msg_type, target_id, f"未绑定发件地址 {addr}")
|
||
return
|
||
|
||
def _do():
|
||
with _ComSession() as app:
|
||
_send_mail(app, addr, to_addr, subject, body)
|
||
_save_to_sent(app, addr, to_addr, subject, body)
|
||
|
||
try:
|
||
await asyncio.to_thread(_do)
|
||
except Exception as e:
|
||
await _reply(ws, msg_type, target_id, f"发送失败：{e}")
|
||
return
|
||
await _reply(ws, msg_type, target_id, f"已从 {addr} 发送给 {to_addr}")
|
||
|
||
|
||
# ==================== 入口 ====================
|
||
async def handle_mail(ws, msg_type, target_id, user_id, raw_text):
|
||
global websocket
|
||
websocket = ws
|
||
|
||
t = (raw_text or "").strip()
|
||
if t.lower() == ".mail":
|
||
await _reply(ws, msg_type, target_id, HELP_TEXT)
|
||
return
|
||
if not t.lower().startswith(".mail "):
|
||
return
|
||
rest = t[6:].strip()
|
||
if not rest:
|
||
await _reply(ws, msg_type, target_id, HELP_TEXT)
|
||
return
|
||
|
||
try:
|
||
tokens = shlex.split(rest)
|
||
except ValueError:
|
||
await _reply(ws, msg_type, target_id, "参数引号不匹配，请检查（主题/正文需用英文引号包裹）")
|
||
return
|
||
if not tokens:
|
||
await _reply(ws, msg_type, target_id, HELP_TEXT)
|
||
return
|
||
|
||
sub = tokens[0].lower()
|
||
args = tokens[1:]
|
||
|
||
if sub == "help":
|
||
await _reply(ws, msg_type, target_id, HELP_TEXT)
|
||
elif sub == "bind":
|
||
await _cmd_bind(ws, msg_type, target_id, user_id, args[0] if args else "")
|
||
elif sub == "unbind":
|
||
await _cmd_unbind(ws, msg_type, target_id, user_id, args[0] if args else "")
|
||
elif sub == "account":
|
||
await _cmd_account(ws, msg_type, target_id, user_id)
|
||
elif sub == "default":
|
||
await _cmd_default(ws, msg_type, target_id, user_id, args[0] if args else "")
|
||
elif sub == "list":
|
||
await _cmd_list(ws, msg_type, target_id, user_id, args)
|
||
elif sub == "info":
|
||
await _cmd_info(ws, msg_type, target_id, user_id, args)
|
||
elif sub == "push":
|
||
await _cmd_push(ws, msg_type, target_id, user_id, args)
|
||
elif sub == "new":
|
||
await _cmd_new(ws, msg_type, target_id, user_id, args)
|
||
else:
|
||
await _reply(ws, msg_type, target_id, f"未知子命令：{sub}，发送 .mail help 查看帮助")
|
||
|
||
|
||
# ==================== 绑定确认端点（HTTP） ====================
|
||
@app.route("/bot/mail/confirm")
|
||
def confirm():
|
||
token = request.args.get("token", "")
|
||
with _pending_lock:
|
||
info = _pending_tokens.pop(token, None)
|
||
if not info:
|
||
return "<html><body style='font-family:sans-serif;padding:40px'><h3>链接无效</h3><p>该确认链接不存在或已被使用。</p></body></html>", 400
|
||
if time.time() - info["created"] > TOKEN_EXPIRE:
|
||
return "<html><body style='font-family:sans-serif;padding:40px'><h3>链接已过期</h3><p>请回到聊天窗口重新发送 .mail bind 命令。</p></body></html>", 400
|
||
|
||
qq = info["qq"]
|
||
email = info["email"]
|
||
|
||
def _do_confirm(data):
|
||
if email not in data["addresses"]:
|
||
data["addresses"].append(email)
|
||
if not data["default"]:
|
||
data["default"] = email
|
||
|
||
_mutate_user(qq, _do_confirm)
|
||
|
||
# qq 是统一身份 uid（历史变量名遗留）：按 TG + 已绑定 QQ 扇出镜像推送
|
||
notification_queue.put({
|
||
"uid": int(qq),
|
||
"content": f"邮箱 {email} 绑定成功",
|
||
"tag": "邮件绑定",
|
||
})
|
||
return "<html><body style='font-family:sans-serif;padding:40px'><h3>绑定成功</h3><p>可以返回聊天窗口了。</p></body></html>"
|
||
|
||
|
||
# ==================== 后台线程 ====================
|
||
def _poll_once():
|
||
try:
|
||
os.makedirs(MAIL_DIR, exist_ok=True)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
files = [f for f in os.listdir(MAIL_DIR) if f.endswith(".json")]
|
||
except OSError:
|
||
return
|
||
if not files:
|
||
return
|
||
|
||
now = time.time()
|
||
with _ComSession() as app:
|
||
for fn in files:
|
||
qq_str = fn[:-5]
|
||
if not qq_str.isdigit():
|
||
continue
|
||
qq = int(qq_str)
|
||
data = _load_user(qq)
|
||
push = data.get("push", {})
|
||
if not push.get("enabled"):
|
||
continue
|
||
if now < _poll_next.get(qq, 0):
|
||
continue
|
||
interval = int(push.get("interval_minutes", DEFAULT_POLL_MINUTES))
|
||
_poll_next[qq] = now + interval * 60
|
||
|
||
last_seen = dict(push.get("last_seen", {}))
|
||
addrs = data.get("addresses", [])
|
||
init_seen = {} # 首次轮询建立的水位（不推历史邮件），循环末统一落盘
|
||
for addr in addrs:
|
||
try:
|
||
prev_max = last_seen.get(addr)
|
||
changes = _mail_changes(app, addr, prev_max)
|
||
cur_max = changes["max_id"]
|
||
if prev_max is None:
|
||
init_seen[addr] = cur_max
|
||
continue
|
||
if cur_max <= prev_max:
|
||
continue
|
||
new_items = [(m["id"], m["subject"], m["from"], m["from_address"], m["ts"])
|
||
for m in changes["items"]]
|
||
if not new_items:
|
||
continue
|
||
# 水位线不在此推进：投递成功后由 notification_loop 调
|
||
# _advance_last_seen 推进。投递全失败/无推送目标时保留水位，
|
||
# 下一轮轮询重新发现（通知不丢；_inflight 防重复入队）。
|
||
if not identity.push_targets(qq):
|
||
continue
|
||
for mid, subj, from_str, from_addr, ts in new_items[:LIST_MAX]:
|
||
sender_name, sender_addr = _split_sender(from_str, from_addr)
|
||
if not sender_addr:
|
||
sender_part = sender_name or "(未知)"
|
||
elif sender_name:
|
||
sender_part = f"{sender_name} <{sender_addr}>"
|
||
else:
|
||
sender_part = f"<{sender_addr}>"
|
||
key = (int(qq), addr, int(mid))
|
||
with _inflight_lock:
|
||
if key in _inflight:
|
||
continue
|
||
_inflight.add(key)
|
||
# qq 是统一身份 uid（历史变量名遗留）：按 TG + 已绑定 QQ 扇出
|
||
notification_queue.put({
|
||
"uid": int(qq),
|
||
"tag": "新邮件",
|
||
"addr": addr,
|
||
"mid": int(mid),
|
||
"content": (
|
||
f"新邮件 [{addr}]\n\n"
|
||
f"#{mid} {_fmt_time(ts)}\n"
|
||
f"{sender_part}\n"
|
||
f"【{subj}】\n\n"
|
||
f"发送 .mail info {mid} 或前往 https://mail.rrrosmontis.icu 以查看正文"
|
||
),
|
||
"subject": subj,
|
||
"sender": sender_part,
|
||
"ts": ts,
|
||
"time": _fmt_time(ts),
|
||
"url": "https://mail.rrrosmontis.icu",
|
||
})
|
||
except Exception:
|
||
continue
|
||
if init_seen:
|
||
# 首轮水位与并发指令写合并落盘（不覆盖 addresses/push.enabled 等字段）
|
||
def _merge_seen(data, _init=init_seen):
|
||
ls = data["push"].setdefault("last_seen", {})
|
||
for a, v in _init.items():
|
||
if a not in ls:
|
||
ls[a] = v
|
||
try:
|
||
_mutate_user(qq, _merge_seen)
|
||
except Exception as e:
|
||
print(f"[mail] 首轮水位落盘失败 uid={qq}: {e}")
|
||
|
||
|
||
def _poll_loop():
|
||
while True:
|
||
try:
|
||
_poll_once()
|
||
except Exception as e:
|
||
print(f"[mail] 轮询异常: {e}")
|
||
time.sleep(30)
|
||
|
||
|
||
def _run_http():
|
||
from waitress import serve
|
||
serve(app, host=HTTP_HOST, port=HTTP_PORT)
|
||
|
||
|
||
def start_services():
|
||
"""由 main.py 在启动时调用：创建 mail 目录、起 HTTP 确认服务与 push 轮询线程。"""
|
||
os.makedirs(MAIL_DIR, exist_ok=True)
|
||
threading.Thread(target=_run_http, daemon=True).start()
|
||
threading.Thread(target=_poll_loop, daemon=True).start()
|
||
|
||
|
||
# ==================== asyncio 通知消费 ====================
|
||
async def notification_loop():
|
||
"""Publish mail events to the durable per-target notification center.
|
||
|
||
Watermarks advance only through the center's mail acknowledger after a real
|
||
delivery. LLM persona rendering is a per-target concern and cannot block
|
||
command-mode recipients or lose an event during a model outage.
|
||
"""
|
||
import notifications
|
||
await asyncio.sleep(STARTUP_GRACE)
|
||
while True:
|
||
try:
|
||
while True:
|
||
item = notification_queue.get_nowait()
|
||
try:
|
||
mid = item.get("mid")
|
||
uid = item.get("uid")
|
||
kind = "mail" if mid is not None else "mail_bind"
|
||
dedupe = (
|
||
f"mail:{uid}:{item.get('addr')}:{mid}" if mid is not None
|
||
else f"mail-bind:{uid}:{item.get('content')}:{time.time()}"
|
||
)
|
||
event = notifications.Event(
|
||
kind=kind,
|
||
dedupe_key=dedupe,
|
||
legacy_text=item.get("content", ""),
|
||
uid=int(uid) if uid is not None else None,
|
||
channel=item.get("channel"),
|
||
msg_type=item.get("message_type"),
|
||
target_id=item.get("target_id"),
|
||
payload={
|
||
"addr": item.get("addr"), "mid": mid,
|
||
"subject": item.get("subject", ""),
|
||
"sender": item.get("sender", ""),
|
||
"time": item.get("time", ""),
|
||
"url": item.get("url", "https://mail.rrrosmontis.icu"),
|
||
},
|
||
)
|
||
notifications.center.publish(event)
|
||
except Exception as e:
|
||
print(f"[mail] 通知入队异常: {type(e).__name__}: {e}")
|
||
finally:
|
||
if item.get("addr") is not None and item.get("mid") is not None:
|
||
with _inflight_lock:
|
||
_inflight.discard((int(item["uid"]), item["addr"], int(item["mid"])))
|
||
except queue.Empty:
|
||
pass
|
||
await asyncio.sleep(2)
|