diff --git a/src/ai_client.py b/src/ai_client.py index 3dbb245..361a703 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -31,6 +31,7 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 - app_name: 仅当 category 为 "notification" 且邮件包含验证码时填写,填写发送验证码的应用/平台名称(如「GitHub」「微信」「Kraken」)。非验证码邮件留空字符串 - links: 提取用户需要点击的链接。其中退订/取消订阅链接按下方「退订链接提取规则」强制执行,其余仅提取邮件明确要求用户点击才能完成某操作的链接(如确认邮箱、重置密码、查看账单、审批请求、查看通知详情等)。不要提取推广性质的链接(如下载App、注册领优惠、查看更多商品等营销链接)、追踪像素、邮件签名中的社交链接。每条包含简短描述文字和完整 URL。最多 3 条。没有符合条件的链接时为空数组 - 退订链接提取规则(最高优先级,仅适用于 category 为 "promotion" 的邮件): + * 若邮件头提供了 List-Unsubscribe 头(用户消息中「List-Unsubscribe:」行),该链接是权威退订入口,优先使用它作为退订链接(除非它明显是 mailto: 形式而正文里能找到更好的 http 链接,此时两者取其一即可) * 推广邮件必须在 links 中输出退订链接,且放在 links 数组的第一位;只有邮件正文确实不存在任何退订入口时,links 才可以为空 * 退订链接通常位于邮件最底部/页脚区域,表现形式多样:文字为 unsubscribe、Unsubscribe、退订、取消订阅、点击这里退订、manage preferences、email preferences、update preferences、subscription settings、opt out 等;也可能是图片链接,或包裹在 HTML 标签中的 ,必须提取 href 属性里的完整 URL * 同一封邮件出现多个退订相关链接时,只提取最主要的那个(优先取正文最后出现的完整 unsubscribe 链接) @@ -44,10 +45,13 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 @retry() -def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str, account: str = "") -> dict[str, Any]: +def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str, + account: str = "", list_unsubscribe: str = "") -> dict[str, Any]: body_preview = body[:80].replace("\n", " ") logger.info("AI 摘要请求: sender=%s subject=%s body_preview=%s", sender, subject, body_preview) content = f"收件人: {recipient}\n发件人: {sender}\n主题: {subject}\n账号: {account}\n正文:\n{body}" + if list_unsubscribe: + content += f"\n\nList-Unsubscribe: {list_unsubscribe}" payload = { "model": ai_cfg.model, diff --git a/src/email_client.py b/src/email_client.py index b011a05..0e7db31 100644 --- a/src/email_client.py +++ b/src/email_client.py @@ -14,7 +14,7 @@ logger = logging.getLogger(__name__) class Email: def __init__(self, uid: bytes, subject: str, sender: str, recipient: str, body: str, date: str, reply_to: str = "", account_email: str = "", - folder: str = "INBOX", cc: str = ""): + folder: str = "INBOX", cc: str = "", list_unsubscribe: str = ""): self.uid = uid self.subject = subject self.sender = sender @@ -25,6 +25,7 @@ class Email: self.account_email = account_email self.folder = folder self.cc = cc + self.list_unsubscribe = list_unsubscribe def _decode_str(s: str) -> str: @@ -72,6 +73,31 @@ def _get_text_from_msg(msg) -> str: return "" +def _parse_list_unsubscribe(header_value: str) -> str: + """解析 List-Unsubscribe 头(RFC 2369),返回退订链接。 + + 优先返回 http(s) 链接(可作按钮直接打开),没有则退回 mailto: 链接。 + 示例头: , + """ + if not header_value: + return "" + # 标准格式是尖括号包裹的 URL 列表,逐个提取 + urls = re.findall(r"<([^>]+)>", header_value) + if not urls: + # 兼容无尖括号的裸格式,按逗号切分后取 http/mailto + urls = [u.strip() for u in header_value.split(",")] + http_url = "" + mailto_url = "" + for u in urls: + u = u.strip() + if u.startswith(("http://", "https://")): + http_url = u + break + if u.startswith("mailto:"): + mailto_url = u + return http_url or mailto_url + + _PROVIDERS_NEED_ID = ("163.com", "126.com") @@ -171,11 +197,12 @@ def fetch_unseen_emails(account: EmailAccount, skip_uids: set[bytes] | None = No date_str = msg.get("Date", "") body = _get_text_from_msg(msg) body_len = len(body) + list_unsubscribe = _parse_list_unsubscribe(msg.get("List-Unsubscribe", "")) logger.info(" 邮件: [%s] from=%s to=%s reply-to=%s (%d 字符)", subject, sender, recipient, reply_to, body_len) emails.append(Email(uid=uid, subject=subject, sender=sender, recipient=recipient, body=body, date=date_str, reply_to=reply_to, account_email=account.username, - folder=folder, cc=cc)) + folder=folder, cc=cc, list_unsubscribe=list_unsubscribe)) # FETCH 可能被服务器自动标已读,立即标回未读 conn.uid("STORE", uid, "-FLAGS", "\\Seen") diff --git a/src/summarizer.py b/src/summarizer.py index a2cda3b..fd6ed9f 100644 --- a/src/summarizer.py +++ b/src/summarizer.py @@ -46,7 +46,8 @@ def ai_process(cfg: Config, user: UserConfig, acct_idx: int, mail: Email, tg_msg # 阶段2:AI 处理,更新为思考中 edit_message(cfg.bot_token, user.chat_id, tg_msg_id, f"🤖 *AI思考中…*\n{label}") - summary = summarize_email(user.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email) + summary = summarize_email(user.ai, mail.recipient, mail.subject, mail.sender, + mail.body, mail.account_email, mail.list_unsubscribe) summary["recipient"] = mail.recipient if "@" not in summary.get("sender", ""): summary["sender"] = mail.sender @@ -65,6 +66,7 @@ def ai_process(cfg: Config, user: UserConfig, acct_idx: int, mail: Email, tg_msg "account_email": mail.account_email, "uid": mail.uid, "folder": mail.folder, + "list_unsubscribe": mail.list_unsubscribe, "thinking_msg_id": tg_msg_id if tg_msg_id else 0, } @@ -74,11 +76,13 @@ def tg_send_and_mark(cfg: Config, user: UserConfig, info: dict): acct = user.email_accounts[info["acct_idx"]] thinking_msg_id = info.get("thinking_msg_id") if thinking_msg_id: - from src.tg_bot import _summary_keyboard + from src.tg_bot import _summary_keyboard, _ensure_unsub_link from src.database import save_email_context can_reply = info["data"].get("can_reply", True) is_promotion = info["data"].get("category") == "promotion" - links = info["data"].get("links", []) + # AI 漏提取退订链接时,用 List-Unsubscribe 头兜底补上 + links = _ensure_unsub_link(info["data"].get("links", []), info.get("list_unsubscribe", "")) + info["data"]["links"] = links edit_message(cfg.bot_token, user.chat_id, thinking_msg_id, info["text"], _summary_keyboard(can_reply, is_promotion, links)) save_email_context(thinking_msg_id, { @@ -105,7 +109,8 @@ def tg_send_and_mark(cfg: Config, user: UserConfig, info: dict): original_sender=info.get("original_sender", ""), original_recipient=info.get("original_recipient", ""), original_reply_to=info.get("original_reply_to", ""), - account_email=info.get("account_email", "")) + account_email=info.get("account_email", ""), + list_unsubscribe=info.get("list_unsubscribe", "")) # 标记已读放到后台线程 def _bg_mark(): try: diff --git a/src/tg_bot.py b/src/tg_bot.py index 2593caa..0c3b2b8 100644 --- a/src/tg_bot.py +++ b/src/tg_bot.py @@ -22,10 +22,12 @@ def send_summary(bot_token: str, chat_id: str, summary_text: str, summary_data: dict, original_body: str, account_idx: int, original_sender: str = "", original_recipient: str = "", original_reply_to: str = "", account_email: str = "", - original_cc: str = "") -> int: + original_cc: str = "", list_unsubscribe: str = "") -> int: can_reply = summary_data.get("can_reply", True) is_promotion = summary_data.get("category") == "promotion" - links = summary_data.get("links", []) + # AI 漏提取退订链接时,用 List-Unsubscribe 头兜底补上 + links = _ensure_unsub_link(summary_data.get("links", []), list_unsubscribe) + summary_data["links"] = links # summary_data 里可能有 cc if original_cc: summary_data["cc"] = original_cc @@ -217,9 +219,14 @@ def _link_buttons(links: list | None) -> list: rows = [] for link in unsub_links + other_links: url = link.get("url", "") - # 校验 URL:合法 http/https,无多重协议头,长度合理。 - # 退订链接常带长 tracking 参数(实测 OpenAI 的 >700 字符),上限放宽到 2048 - if not (url.startswith("http") and url.count("://") == 1 and len(url) < 2048): + # 校验 URL:http/https 需无多重协议头;mailto: 仅用于退订场景(如 mailto:x@y.com?subject=Unsubscribe) + if url.startswith("http"): + valid = url.count("://") == 1 and len(url) < 2048 + elif url.startswith("mailto:"): + valid = len(url) < 2048 + else: + valid = False + if not valid: continue text = (link.get("text") or "").strip()[:30] if not text: @@ -229,6 +236,25 @@ def _link_buttons(links: list | None) -> list: return rows +def _ensure_unsub_link(links: list, list_unsubscribe: str = "") -> list: + """AI 未提取到退订链接时,用邮件头 List-Unsubscribe 的链接兜底补一个退订按钮。 + + 仅当 links 中没有任何退订类链接且头部确实有可用的 http(s)/mailto 链接时生效。 + """ + if any(_is_unsub_link(l) for l in links): + return links + url = (list_unsubscribe or "").strip() + if url.startswith("http"): + valid = url.count("://") == 1 and len(url) < 2048 + elif url.startswith("mailto:"): + valid = len(url) < 2048 + else: + valid = False + if valid: + return [{"text": "退订", "url": url}] + list(links) + return links + + def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False, links: list | None = None) -> dict: if is_promotion: kb = [] diff --git a/tests/test_unsub_keyboard.py b/tests/test_unsub_keyboard.py index 486577b..7c84884 100644 --- a/tests/test_unsub_keyboard.py +++ b/tests/test_unsub_keyboard.py @@ -1,6 +1,11 @@ """测试摘要键盘渲染:推广邮件的退订链接按钮必须显示且排最前。""" -from src.tg_bot import _is_unsub_link, _link_buttons, _summary_keyboard +from src.tg_bot import ( + _ensure_unsub_link, + _is_unsub_link, + _link_buttons, + _summary_keyboard, +) def _rows_of(kb: dict) -> list[list[dict]]: @@ -102,4 +107,43 @@ def test_oversized_url_filtered(): url = "https://example.com/" + "x" * 2100 links = [{"text": "退订", "url": url}] rows = _link_buttons(links) - assert rows == [] \ No newline at end of file + assert rows == [] + + +def test_mailto_unsub_link_rendered(): + """mailto: 退订链接(如 InferX 的 mailto:team@inferx.net?subject=Unsubscribe)必须渲染""" + links = [{"text": "退订", "url": "mailto:team@inferx.net?subject=Unsubscribe"}] + kb = _summary_keyboard(can_reply=False, is_promotion=True, links=links) + rows = kb["inline_keyboard"] + assert rows[0][0]["url"] == "mailto:team@inferx.net?subject=Unsubscribe" + assert "🔕" in rows[0][0]["text"] + assert _is_unsub_link(links[0]) + + +def test_ensure_unsub_link_http_header_fallback(): + """AI 没提取到退订链接时,用 List-Unsubscribe 头的 http 链接兜底""" + header = "https://links.iterable.com/s/uh/tKjpy4SHCKvbzE2votNyZEJDvlMdeodVyWFExNs310ugJ_RvZtoZq4gGLQ0BXMSL2lf3kEZsmr1XJmrODIFHpEwcypo3LR096pZ7jsfBynZBgGWz171Wc1E6ZIIC5Lu0e-Stq6p5-_Ad3CnNlUWrW1tnrLSeiGsOiQG_zMYKVbtfQZgvfm3QYfD9lQ/Cey29KU8yOzIUltaEJA8EMRFf-FKMZQy/24" + links = _ensure_unsub_link([], header) + assert links == [{"text": "退订", "url": header}] + kb = _summary_keyboard(can_reply=False, is_promotion=True, links=links) + assert kb["inline_keyboard"][0][0]["url"] == header + + +def test_ensure_unsub_link_mailto_header_fallback(): + """List-Unsubscribe 头只有 mailto 时也兜底""" + header = "mailto:unsubscribe+19693436+25822730@unsubscribe.iterable.com" + links = _ensure_unsub_link([], header) + assert links == [{"text": "退订", "url": header}] + + +def test_ensure_unsub_link_no_duplicate_when_ai_extracted(): + """AI 已提取退订链接时,不重复添加头部链接""" + ai_links = [{"text": "取消订阅", "url": "https://r.openai.com/asm/unsubscribe/?x=1"}] + links = _ensure_unsub_link(ai_links, "https://links.example.com/unsub") + assert links == ai_links + + +def test_ensure_unsub_link_ignores_invalid_header(): + """头部无有效链接时不兜底""" + assert _ensure_unsub_link([], "") == [] + assert _ensure_unsub_link([], "javascript:alert(1)") == [] \ No newline at end of file