diff --git a/config.yaml.example b/config.yaml.example index 3cc68ed..3e6de2c 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -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" diff --git a/main.py b/main.py index a6d0c4f..3aa4702 100644 --- a/main.py +++ b/main.py @@ -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("机器人已停止") diff --git a/src/ai_client.py b/src/ai_client.py index d75f049..ced66b3 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -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, diff --git a/src/config.py b/src/config.py index 18d901f..3ab4f45 100644 --- a/src/config.py +++ b/src/config.py @@ -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"]) diff --git a/src/smtp_client.py b/src/smtp_client.py new file mode 100644 index 0000000..7ad2b94 --- /dev/null +++ b/src/smtp_client.py @@ -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() diff --git a/src/summarizer.py b/src/summarizer.py index 29d72eb..facf669 100644 --- a/src/summarizer.py +++ b/src/summarizer.py @@ -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: diff --git a/src/tg_bot.py b/src/tg_bot.py index 71a03eb..6eff8dd 100644 --- a/src/tg_bot.py +++ b/src/tg_bot.py @@ -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)