earthjasonlin/ai-mail-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: interactive inline buttons, SMTP reply, AI reply suggestions

Browse Source 
- Inline keyboard: view original / back to summary toggle per message
- SMTP config per account for sending replies
- Reply button: click, type message, auto-send via SMTP
- AI Reply button: generates 3 suggestions via DeepSeek
  - Simple replies (OK, Got it) send immediately on tap
  - Substantive replies require confirm before send
  - Hides AI Reply button when AI determines reply unnecessary
- Telegram long polling integrated into main loop for real-time interaction
This commit is contained in:
Zichao Lin
2026-07-02 20:21:06 +08:00
parent 279f8c50e7
commit 0dbc7ee661
7 changed files with 410 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
config.yaml.example
View File

@@ -3,11 +3,19 @@ email_accounts:
imap_port: 993
username: "your_email@gmail.com"
password: "your_app_password"
smtp:
server: "smtp.gmail.com"
port: 465
use_ssl: true
- imap_server: "imap.qq.com"
- imap_server: "imap.126.com"
imap_port: 993
username: "your_email@qq.com"
password: "your_authorization_code"
username: "your_email@126.com"
password: "your_auth_code"
smtp:
server: "smtp.126.com"
port: 465
use_ssl: true
ai:
api_key: "sk-your_deepseek_api_key"

18
main.py
View File

@@ -4,6 +4,7 @@ import sys
import time
from src.config import load_config
from src.summarizer import process_all
from src.tg_bot import poll_telegram
logging.basicConfig(
level=logging.INFO,
@@ -30,17 +31,22 @@ def main():
cfg = load_config(cfg_path)
logger.info(f"AI邮件摘要机器人已启动，轮询间隔: {cfg.polling.interval_seconds}s")
last_check = 0.0
last_update_id = 0
while _running:
try:
process_all(cfg)
now = time.time()
if now - last_check >= cfg.polling.interval_seconds:
process_all(cfg)
last_check = now
last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id)
except Exception as e:
logger.error(f"轮询出错: {e}", exc_info=True)
logger.error(f"主循环出错: {e}", exc_info=True)
if _running:
for _ in range(cfg.polling.interval_seconds):
if not _running:
break
time.sleep(1)
time.sleep(1)
logger.info("机器人已停止")

43
src/ai_client.py
View File

@@ -40,6 +40,49 @@ def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str,
"Content-Type": "application/json",
}
return _call_deepseek(ai_cfg, payload)
REPLY_PROMPT = """你是一个邮件回复助手。根据以下邮件内容生成3个建议回复。
返回 JSON:
{
"can_reply": true/false,
"suggestions": [
{"text": "回复内容", "is_simple": true/false},
{"text": "回复内容", "is_simple": true/false},
{"text": "回复内容", "is_simple": true/false}
]
}
- can_reply 为 false 时表示无需回复或不适合自动建议（此时忽略 suggestions）
- is_simple 为 true 表示纯确认性回复（如"好的""收到""明白"），用户选择后可直接发送
- is_simple 为 false 表示涉及实质内容，需要用户确认后再发送
- 每条回复控制在 50 字以内
只返回 JSON，不要包含任何其他文字。"""
@retry()
def generate_reply_suggestions(ai_cfg: AIConfig, sender: str, subject: str, body: str) -> dict:
content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body[:4000]}"
payload = {
"model": ai_cfg.model,
"messages": [
{"role": "system", "content": REPLY_PROMPT},
{"role": "user", "content": content},
],
"response_format": {"type": "json_object"},
}
return _call_deepseek(ai_cfg, payload)
def _call_deepseek(ai_cfg: AIConfig, payload: dict) -> dict:
headers = {
"Authorization": f"Bearer {ai_cfg.api_key}",
"Content-Type": "application/json",
}
resp = requests.post(
f"{ai_cfg.base_url.rstrip('/')}/chat/completions",
headers=headers,

19
src/config.py
View File

@@ -1,14 +1,23 @@
from dataclasses import dataclass, field
from typing import List
from dataclasses import dataclass
from typing import List, Optional
import yaml
@dataclass
class SmtpConfig:
server: str
port: int
use_ssl: bool = True
use_starttls: bool = False
@dataclass
class EmailAccount:
imap_server: str
imap_port: int
username: str
password: str
smtp: Optional[SmtpConfig] = None
@dataclass
@@ -41,7 +50,11 @@ def load_config(path: str) -> Config:
with open(path, "r", encoding="utf-8") as f:
raw = yaml.safe_load(f)
accounts = [EmailAccount(**a) for a in raw["email_accounts"]]
accounts = []
for a in raw["email_accounts"]:
if "smtp" in a and a["smtp"] is not None:
a["smtp"] = SmtpConfig(**a["smtp"])
accounts.append(EmailAccount(**a))
ai = AIConfig(**raw["ai"])
tg = TelegramConfig(**raw["telegram"])
polling = PollingConfig(**raw["polling"])

32
src/smtp_client.py Normal file
View File

@@ -0,0 +1,32 @@
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from src.config import EmailAccount
from src.retry import retry
@retry()
def send_reply(account: EmailAccount, to_addr: str, subject: str, body: str):
if not account.smtp:
raise RuntimeError(f"账号 {account.username} 未配置 SMTP")
s = account.smtp
if s.use_ssl:
conn = smtplib.SMTP_SSL(s.server, s.port, timeout=15)
else:
conn = smtplib.SMTP(s.server, s.port, timeout=15)
if s.use_starttls:
conn.starttls()
conn.login(account.username, account.password)
reply_subject = subject if subject.lower().startswith("re:") else f"Re: {subject}"
msg = MIMEMultipart("alternative")
msg["From"] = account.username
msg["To"] = to_addr
msg["Subject"] = reply_subject
msg.attach(MIMEText(body, "plain", "utf-8"))
conn.sendmail(account.username, [to_addr], msg.as_string())
conn.quit()

10
src/summarizer.py
View File

@@ -2,20 +2,20 @@ import logging
from src.config import Config
from src.email_client import fetch_unseen_emails, mark_as_seen
from src.ai_client import summarize_email
from src.tg_bot import send_message, format_summary
from src.tg_bot import send_summary, format_summary
logger = logging.getLogger(__name__)
def process_all(cfg: Config):
for acct in cfg.email_accounts:
for idx, acct in enumerate(cfg.email_accounts):
try:
_process_account(cfg, acct)
_process_account(cfg, acct, idx)
except Exception as e:
logger.error(f"处理邮箱 {acct.username} 时出错: {e}", exc_info=True)
def _process_account(cfg: Config, acct):
def _process_account(cfg: Config, acct, acct_idx: int):
logger.info(f"检查邮箱: {acct.username}")
emails = fetch_unseen_emails(acct)
@@ -32,7 +32,7 @@ def _process_account(cfg: Config, acct):
summary = summarize_email(cfg.ai, acct.username, mail.subject, mail.sender, mail.body)
summary["recipient"] = acct.username
text = format_summary(summary)
send_message(cfg.telegram, text)
send_summary(cfg.telegram, cfg.telegram.chat_id, text, summary, mail.body, acct_idx)
seen_uids.append(mail.uid)
logger.info(f" 已发送到 Telegram")
except Exception as e:

322
src/tg_bot.py
View File

@@ -1,30 +1,79 @@
import json
import logging
import requests
from src.config import TelegramConfig
from src.config import TelegramConfig, Config
from src.ai_client import generate_reply_suggestions
from src.smtp_client import send_reply
from src.retry import retry
@retry()
def send_message(tg_cfg: TelegramConfig, text: str):
url = f"https://api.telegram.org/bot{tg_cfg.bot_token}/sendMessage"
payload = {
"chat_id": tg_cfg.chat_id,
"text": text,
"parse_mode": "MarkdownV2",
}
resp = requests.post(url, json=payload, timeout=30)
resp.raise_for_status()
logger = logging.getLogger(__name__)
_priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}
# msg_id -> email context
_email_contexts: dict[int, dict] = {}
# chat_id -> conversation state
_conversations: dict[int, dict] = {}
# ── Public API ──────────────────────────────────────────
@retry()
def send_summary(tg_cfg: TelegramConfig, chat_id: str, summary_text: str,
summary_data: dict, original_body: str, account_idx: int) -> int:
result = _tg_req(tg_cfg, "sendMessage", {
"chat_id": chat_id,
"text": summary_text,
"parse_mode": "MarkdownV2",
"reply_markup": _summary_keyboard(),
})
msg_id = result["result"]["message_id"]
_email_contexts[msg_id] = {
"summary_text": summary_text,
"summary_data": summary_data,
"original_body": original_body[:10000],
"account_idx": account_idx,
"sender": summary_data.get("sender", ""),
"subject": summary_data.get("subject", ""),
}
return msg_id
@retry()
def send_text(tg_cfg: TelegramConfig, chat_id: str, text: str):
_tg_req(tg_cfg, "sendMessage", {
"chat_id": chat_id,
"text": text,
})
def poll_telegram(tg_cfg: TelegramConfig, cfg: Config, last_update_id: int) -> int:
try:
resp = requests.get(
f"https://api.telegram.org/bot{tg_cfg.bot_token}/getUpdates",
params={"offset": last_update_id, "timeout": 10},
timeout=15,
)
resp.raise_for_status()
data = resp.json()
if not data.get("ok"):
return last_update_id
for update in data.get("result", []):
_handle_update(tg_cfg, cfg, update)
last_update_id = update["update_id"] + 1
except Exception as e:
logger.warning("Telegram polling error: %s", e)
return last_update_id
# ── Formatting ──────────────────────────────────────────
def format_summary(data: dict) -> str:
priority = data.get("priority", "medium")
icon = _priority_icon.get(priority, "⚪")
lines = [
f"*📧 {_escape(data.get('subject', '邮件摘要'))}*",
f"━━━━━━━━━━━━━━━━━━",
"━━━━━━━━━━━━━━━━━━",
f"*收件人:* {_escape(data.get('recipient', '未知'))}",
f"*发件人:* {_escape(data.get('sender', '未知'))}",
f"*优先级:* {icon} {priority.upper()}",
@@ -32,27 +81,238 @@ def format_summary(data: dict) -> str:
_escape(data.get("summary", "")),
"",
]
if data.get("action_required"):
lines.append(f"*📌 需要处理:* 是")
items = data.get("action_items", [])
if items:
lines.append(f"*待办事项:*")
for item in items:
lines.append(f" • {_escape(item)}")
lines.append("*📌 需要处理:* 是")
for item in data.get("action_items", []):
lines.append(f" \\- {_escape(item)}")
lines.append("")
points = data.get("key_points", [])
if points:
lines.append(f"*关键要点:*")
for p in points:
lines.append(f" • {_escape(p)}")
for p in data.get("key_points", []):
lines.append(f" \\- {_escape(p)}")
return "\n".join(lines)
def _escape(text: str) -> str:
special = "_*[]()~`>#+-=|{}.!"
for ch in special:
for ch in "_*[]()~`>#+-=|{}.!":
text = text.replace(ch, f"\\{ch}")
return text
# ── Inline keyboards ────────────────────────────────────
CALLBACK_VIEW_ORIG = "v"
CALLBACK_VIEW_SUMM = "s"
CALLBACK_REPLY = "r"
CALLBACK_AI_REPLY = "a"
CALLBACK_SELECT_REPLY = "sr"
CALLBACK_CONFIRM_REPLY = "cr"
CALLBACK_CANCEL = "c"
def _summary_keyboard() -> dict:
return {
"inline_keyboard": [
[
{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG},
{"text": "回复", "callback_data": CALLBACK_REPLY},
],
[
{"text": "AI回复", "callback_data": CALLBACK_AI_REPLY},
],
]
}
def _orig_keyboard() -> dict:
return {
"inline_keyboard": [
[
{"text": "返回摘要", "callback_data": CALLBACK_VIEW_SUMM},
{"text": "回复", "callback_data": CALLBACK_REPLY},
],
[
{"text": "AI回复", "callback_data": CALLBACK_AI_REPLY},
],
]
}
def _ai_reply_keyboard(msg_id: int, suggestions: list) -> dict:
kb = []
for i, s in enumerate(suggestions):
data = f"{CALLBACK_SELECT_REPLY}|{msg_id}|{i}"
kb.append([{"text": s["text"][:30], "callback_data": data}])
kb.append([{"text": "取消", "callback_data": CALLBACK_CANCEL}])
return {"inline_keyboard": kb}
def _confirm_keyboard(msg_id: int, idx: int) -> dict:
return {
"inline_keyboard": [
[
{"text": "确认发送", "callback_data": f"{CALLBACK_CONFIRM_REPLY}|{msg_id}|{idx}"},
{"text": "取消", "callback_data": CALLBACK_CANCEL},
],
]
}
# ── Telegram API ────────────────────────────────────────
def _tg_req(tg_cfg: TelegramConfig, method: str, payload: dict) -> dict:
url = f"https://api.telegram.org/bot{tg_cfg.bot_token}/{method}"
r = requests.post(url, json=payload, timeout=30)
r.raise_for_status()
result = r.json()
if not result.get("ok"):
raise RuntimeError(f"Telegram API error: {result}")
return result
# ── Update handling ─────────────────────────────────────
def _handle_update(tg_cfg: TelegramConfig, cfg: Config, update: dict):
if "callback_query" in update:
_handle_callback(tg_cfg, cfg, update["callback_query"])
elif "message" in update and "text" in update["message"]:
_handle_message(tg_cfg, cfg, update["message"])
def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
chat_id = cb["message"]["chat"]["id"]
msg_id = cb["message"]["message_id"]
data = cb["data"]
cb_id = cb["id"]
_tg_req(tg_cfg, "answerCallbackQuery", {"callback_query_id": cb_id})
parts = data.split("|")
action = parts[0]
ctx = _email_contexts.get(msg_id)
if action == CALLBACK_VIEW_ORIG and ctx:
text = f"*📄 原文 \\- {_escape(ctx['subject'])}*\n\n{_escape(ctx['original_body'][:3500])}"
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": text, "parse_mode": "MarkdownV2",
"reply_markup": _orig_keyboard(),
})
elif action == CALLBACK_VIEW_SUMM and ctx:
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": ctx["summary_text"], "parse_mode": "MarkdownV2",
"reply_markup": _summary_keyboard(),
})
elif action == CALLBACK_REPLY and ctx:
_conversations[chat_id] = {"state": "awaiting_reply", "msg_id": msg_id}
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": ctx["summary_text"], "parse_mode": "MarkdownV2",
"reply_markup": _summary_keyboard(),
})
send_text(tg_cfg, str(chat_id), "请输入你的回复内容：")
elif action == CALLBACK_AI_REPLY and ctx:
try:
acct = cfg.email_accounts[ctx["account_idx"]]
suggestions = generate_reply_suggestions(
cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
)
except Exception as e:
send_text(tg_cfg, str(chat_id), f"AI 生成回复失败: {e}")
return
if not suggestions.get("can_reply") or not suggestions.get("suggestions"):
send_text(tg_cfg, str(chat_id), "此邮件无需回复，已隐藏 AI 回复功能。")
return
_conversations[chat_id] = {
"state": "ai_reply_selection",
"msg_id": msg_id,
"ai_suggestions": suggestions["suggestions"],
}
text = "*AI 建议回复，请选择：*\n\n"
for i, s in enumerate(suggestions["suggestions"], 1):
tag = "✅ 可直接发送" if s.get("is_simple") else "📝 需确认"
text += f"{i}\\. {_escape(s['text'])} _{tag}_\n"
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": text, "parse_mode": "MarkdownV2",
"reply_markup": _ai_reply_keyboard(msg_id, suggestions["suggestions"]),
})
elif action == CALLBACK_SELECT_REPLY and ctx:
idx = int(parts[2])
conv = _conversations.get(chat_id)
if not conv or conv.get("msg_id") != msg_id:
return
suggestions = conv.get("ai_suggestions", [])
if idx >= len(suggestions):
return
sel = suggestions[idx]
if sel.get("is_simple"):
_do_send_reply(tg_cfg, cfg, chat_id, msg_id, ctx, sel["text"])
send_text(tg_cfg, str(chat_id), f"✅ 已发送: {sel['text']}")
_conversations.pop(chat_id, None)
else:
_conversations[chat_id]["selected_idx"] = idx
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": f"确认发送此回复？\n\n{_escape(sel['text'])}",
"parse_mode": "MarkdownV2",
"reply_markup": _confirm_keyboard(msg_id, idx),
})
elif action == CALLBACK_CONFIRM_REPLY and ctx:
idx = int(parts[2])
conv = _conversations.get(chat_id)
if not conv or conv.get("msg_id") != msg_id:
return
suggestions = conv.get("ai_suggestions", [])
if idx >= len(suggestions):
return
_do_send_reply(tg_cfg, cfg, chat_id, msg_id, ctx, suggestions[idx]["text"])
send_text(tg_cfg, str(chat_id), "✅ 回复已发送！")
_conversations.pop(chat_id, None)
elif action == CALLBACK_CANCEL:
_conversations.pop(chat_id, None)
if ctx:
_tg_req(tg_cfg, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": ctx["summary_text"], "parse_mode": "MarkdownV2",
"reply_markup": _summary_keyboard(),
})
def _handle_message(tg_cfg: TelegramConfig, cfg: Config, msg: dict):
chat_id = msg["chat"]["id"]
text = msg["text"].strip()
conv = _conversations.get(chat_id)
if not conv or conv.get("state") != "awaiting_reply":
return
ctx = _email_contexts.get(conv["msg_id"])
if not ctx:
_conversations.pop(chat_id, None)
return
_do_send_reply(tg_cfg, cfg, chat_id, conv["msg_id"], ctx, text)
send_text(tg_cfg, str(chat_id), "✅ 回复已发送！")
_conversations.pop(chat_id, None)
def _do_send_reply(tg_cfg: TelegramConfig, cfg: Config,
chat_id: int, msg_id: int, ctx: dict, reply_text: str):
acct = cfg.email_accounts[ctx["account_idx"]]
if not acct.smtp:
send_text(tg_cfg, str(chat_id), "❌ 该邮箱未配置 SMTP，无法发送回复。")
return
to_addr = ctx["sender"]
subject = ctx["subject"]
send_reply(acct, to_addr, subject, reply_text)