Compare commits

..
0 Commits
6 changed files with 36 additions and 302 deletions

No files matched your search

-1
View File
@@ -3,4 +3,3 @@ __pycache__/
*.pyc
config.json
data/
*_TODO.md
+3 -14
View File
@@ -29,14 +29,8 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以
- category: 邮件类型。推广邮件(营销、促销、广告、 newsletter 推广等)标记为 "promotion";验证码邮件标记为 "notification";普通邮件标记为 "normal"
- verification_code: 仅当 category 为 "notification" 且邮件包含验证码时填写,提取完整的验证码字符串(纯数字、字母数字混合等均可,如「123456」「abc-789」「701383」)。非验证码邮件留空字符串
- 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 标签中的 <a href="...">,必须提取 href 属性里的完整 URL
* 同一封邮件出现多个退订相关链接时,只提取最主要的那个(优先取正文最后出现的完整 unsubscribe 链接)
* 退订链接固定占用 1 个位置,其余链接最多 2 条,不得挤掉退订链接
* 严禁编造退订 URL,确实找不到时 links 输出空数组
- links: 仅提取以下两类链接:1)邮件明确要求用户点击才能完成某操作的链接(如确认邮箱、重置密码、查看账单、审批请求、查看通知详情等);2)退订/取消订阅链接。不要提取推广性质的链接(如下载App、注册领优惠、查看更多商品等营销链接)、追踪像素、邮件签名中的社交链接。每条包含简短描述文字和完整 URL。最多 3 条。没有符合条件的链接时为空数组
- 退订链接提取规则:推广邮件必须提取退订链接。退订链接通常在邮件最底部,包含以下关键词之一:unsubscribe、退订、取消订阅、opt out、manage preferences、email preferences。即使链接被包裹在 HTML 标签中,也要提取 href 属性里的完整 URL。如果邮件底部同时有多个退订相关链接,只提取最主要的那个
- plain_text: 将邮件全文转换为可读的纯文本版本。去除所有HTML标签、CSS样式、脚本代码,保留完整的文字内容和原始段落结构。不删减、不修改、不总结任何内容,完整保留原文所有信息
- summary 要简洁,把需要知道的信息浓缩成一句话,不需要分条列出
- 当 category 为 "promotion" 时,summary 直接写成一句话:「[发送方名称] 在推广 [推广的产品/服务/活动]」,例如「腾讯云在推广双十一云服务器折扣」。此时 can_reply 为 false
@@ -45,17 +39,13 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以
@retry()
def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str,
account: str = "", list_unsubscribe: str = "") -> dict[str, Any]:
def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str, account: 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,
"thinking": {"type": "disabled"},
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": content},
@@ -78,7 +68,6 @@ def expand_promo_detail(ai_cfg: AIConfig, sender: str, subject: str, body: str)
content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body}"
payload = {
"model": ai_cfg.model,
"thinking": {"type": "disabled"},
"messages": [
{"role": "system", "content": PROMO_DETAIL_PROMPT},
{"role": "user", "content": content},
+2 -29
View File
@@ -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 = "", list_unsubscribe: str = ""):
folder: str = "INBOX", cc: str = ""):
self.uid = uid
self.subject = subject
self.sender = sender
@@ -25,7 +25,6 @@ class Email:
self.account_email = account_email
self.folder = folder
self.cc = cc
self.list_unsubscribe = list_unsubscribe
def _decode_str(s: str) -> str:
@@ -73,31 +72,6 @@ def _get_text_from_msg(msg) -> str:
return ""
def _parse_list_unsubscribe(header_value: str) -> str:
"""解析 List-Unsubscribe 头(RFC 2369),返回退订链接。
优先返回 http(s) 链接(可作按钮直接打开),没有则退回 mailto: 链接。
示例头: <https://x.com/unsub>, <mailto:unsub@x.com>
"""
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")
@@ -197,12 +171,11 @@ 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, list_unsubscribe=list_unsubscribe))
folder=folder, cc=cc))
# FETCH 可能被服务器自动标已读,立即标回未读
conn.uid("STORE", uid, "-FLAGS", "\\Seen")
+4 -11
View File
@@ -46,8 +46,7 @@ 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, mail.list_unsubscribe)
summary = summarize_email(user.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email)
summary["recipient"] = mail.recipient
if "@" not in summary.get("sender", ""):
summary["sender"] = mail.sender
@@ -61,12 +60,10 @@ def ai_process(cfg: Config, user: UserConfig, acct_idx: int, mail: Email, tg_msg
"original_recipient": mail.recipient,
"original_reply_to": mail.reply_to,
"original_cc": mail.cc,
"original_subject": mail.subject,
"acct_idx": acct_idx,
"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,
}
@@ -76,13 +73,11 @@ 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, _ensure_unsub_link
from src.tg_bot import _summary_keyboard
from src.database import save_email_context
can_reply = info["data"].get("can_reply", True)
is_promotion = info["data"].get("category") == "promotion"
# AI 漏提取退订链接时,用 List-Unsubscribe 头兜底补上
links = _ensure_unsub_link(info["data"].get("links", []), info.get("list_unsubscribe", ""))
info["data"]["links"] = links
links = info["data"].get("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, {
@@ -94,7 +89,6 @@ def tg_send_and_mark(cfg: Config, user: UserConfig, info: dict):
"original_recipient": info.get("original_recipient", ""),
"original_cc": info.get("original_cc", ""),
"original_reply_to": info.get("original_reply_to", ""),
"original_subject": info.get("original_subject", ""),
"account_email": info.get("account_email", ""),
"account_idx": info["acct_idx"],
"sender": info["data"].get("sender", ""),
@@ -109,8 +103,7 @@ 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", ""),
list_unsubscribe=info.get("list_unsubscribe", ""))
account_email=info.get("account_email", ""))
# 标记已读放到后台线程
def _bg_mark():
try:
+27 -98
View File
@@ -22,12 +22,10 @@ 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 = "", list_unsubscribe: str = "") -> int:
original_cc: str = "") -> int:
can_reply = summary_data.get("can_reply", True)
is_promotion = summary_data.get("category") == "promotion"
# AI 漏提取退订链接时,用 List-Unsubscribe 头兜底补上
links = _ensure_unsub_link(summary_data.get("links", []), list_unsubscribe)
summary_data["links"] = links
links = summary_data.get("links", [])
# summary_data 里可能有 cc
if original_cc:
summary_data["cc"] = original_cc
@@ -193,75 +191,12 @@ CALLBACK_CANCEL = "c"
CALLBACK_PROMO_DETAIL = "pd"
# ── Link buttons ────────────────────────────────────────
_UNSUB_KEYWORDS = (
"unsubscribe", "opt out", "unsub", "退订", "取消订阅",
"manage preferences", "email preferences", "subscription settings",
"subscription preferences", "preferences", "email-pref", "manage-pref",
"偏好设置", "管理偏好", "订阅设置",
)
def _is_unsub_link(link: dict) -> bool:
"""判断链接是否为退订/取消订阅类链接(按文字或 URL 关键词)"""
text = (link.get("text") or "").lower()
url = (link.get("url") or "").lower()
return any(k in text or k in url for k in _UNSUB_KEYWORDS)
def _link_buttons(links: list | None) -> list:
"""把 AI 提取的链接转成键盘按钮行;退订链接排最前并用 🔕 标识,其余用 🔗"""
if not links:
return []
unsub_links = [l for l in links if _is_unsub_link(l)]
other_links = [l for l in links if not _is_unsub_link(l)]
rows = []
for link in unsub_links + other_links:
url = link.get("url", "")
# 校验 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:
text = "退订" if _is_unsub_link(link) else "打开链接"
icon = "🔕" if _is_unsub_link(link) else "🔗"
rows.append([{"text": f"{icon} {text}", "url": url}])
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:
def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False, links: list = None) -> dict:
if is_promotion:
kb = []
kb.extend(_link_buttons(links))
kb.append([{"text": "告诉我更多", "callback_data": CALLBACK_PROMO_DETAIL}])
kb.append([{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}])
return {"inline_keyboard": kb}
return {"inline_keyboard": [
[{"text": "告诉我更多", "callback_data": CALLBACK_PROMO_DETAIL}],
[{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}],
]}
kb = [
[
{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG},
@@ -273,7 +208,12 @@ def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False, links:
{"text": "AI完整回复", "callback_data": CALLBACK_AI_REPLY_FULL},
{"text": "AI简洁回复", "callback_data": CALLBACK_AI_REPLY_SHORT},
])
kb.extend(_link_buttons(links))
for link in (links or []):
url = link.get("url", "")
text = link.get("text", "打开链接")[:30]
# 校验 URL:合法 http/https,无多重协议头,长度合理
if url.startswith("http") and url.count("://") == 1 and len(url) < 200:
kb.append([{"text": f"🔗 {text}", "url": url}])
return {"inline_keyboard": kb}
@@ -370,18 +310,16 @@ def _do_regen(bot_token: str, cfg: Config, chat_id: int, msg_id: int, ctx: dict,
_show_ai_suggestions(bot_token, chat_id, msg_id, new_suggestions, comparison)
def _do_ai_hint(bot_token: str, cfg: Config, chat_id: int, conv: dict, ctx: dict, hint_history: list[str]):
"""在 AI 线程池中根据用户提示生成回复(支持多轮追问)"""
def _do_ai_hint(bot_token: str, cfg: Config, chat_id: int, conv: dict, ctx: dict, hint: str):
"""在 AI 线程池中根据用户提示生成回复"""
user = find_user_by_chat_id(cfg, chat_id)
if not user:
return
style = conv.get("reply_style", "short")
# 拼接历史提示作为上下文
combined_hint = " → ".join(hint_history)
try:
result = generate_more_replies(
user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
conv.get("all_suggestions", []), user_hint=combined_hint, style=style,
conv.get("all_suggestions", []), user_hint=hint, style=style,
)
new_suggestions = result["suggestions"]
comparison = result.get("comparison", "")
@@ -394,10 +332,9 @@ def _do_ai_hint(bot_token: str, cfg: Config, chat_id: int, conv: dict, ctx: dict
conv["all_suggestions"].extend(new_suggestions)
conv["comparison"] = comparison
conv["state"] = "ai_reply_selection"
conv["hint_history"] = hint_history
conv.pop("prompt_msg_id", None)
save_conversation(chat_id, conv)
logger.info("[ai_hint] 根据提示生成 %d 条新回复 (历史 %d 条)", len(new_suggestions), len(hint_history))
logger.info("[ai_hint] 根据提示生成 %d 条新回复", len(new_suggestions))
_show_ai_suggestions(bot_token, chat_id, conv["summary_msg_id"], new_suggestions, comparison)
@@ -423,7 +360,7 @@ def _handle_callback(bot_token: str, cfg: Config, cb: dict):
action = parts[0]
ctx = load_email_context(msg_id)
subject = ctx.get("original_subject") or ctx["subject"] if ctx else "?"
subject = ctx["subject"] if ctx else "?"
logger.info("TG 回调: action=%s msg_id=%d subject=%s", action, msg_id, subject)
if action == CALLBACK_VIEW_ORIG and ctx:
@@ -562,10 +499,8 @@ def _handle_callback(bot_token: str, cfg: Config, cb: dict):
if conv and conv.get("summary_msg_id") == msg_id:
conv["state"] = "awaiting_ai_hint"
conv["prompt_msg_id"] = prompt_msg_id
if "hint_history" not in conv:
conv["hint_history"] = []
save_conversation(chat_id, conv)
logger.info(" 等待用户输入 AI 提示 (历史 %d 条)", len(conv["hint_history"]))
logger.info(" 等待用户输入 AI 提示")
elif action == CALLBACK_CANCEL:
conv = load_conversation(chat_id)
@@ -595,9 +530,8 @@ def _handle_callback(bot_token: str, cfg: Config, cb: dict):
"text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
})
try:
user = find_user_by_chat_id(cfg, chat_id)
details = expand_promo_detail(
user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
)
except Exception as e:
send_text(bot_token, str(chat_id), f"获取详情失败: {e}")
@@ -611,14 +545,13 @@ def _handle_callback(bot_token: str, cfg: Config, cb: dict):
f"\n{_escape(summary)}\n"
f"\n*📋 详情:*\n{detail_lines}"
)
promo_kb = []
promo_kb.extend(_link_buttons(ctx.get("summary_data", {}).get("links", [])))
promo_kb.append([{"text": "收起", "callback_data": CALLBACK_VIEW_SUMM}])
promo_kb.append([{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}])
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": expanded, "parse_mode": "MarkdownV2",
"reply_markup": {"inline_keyboard": promo_kb},
"reply_markup": {"inline_keyboard": [
[{"text": "收起", "callback_data": CALLBACK_VIEW_SUMM}],
[{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}],
]},
})
logger.info(" 显示推广详情")
@@ -683,17 +616,13 @@ def _handle_message(bot_token: str, cfg: Config, msg: dict):
if prompt_msg_id:
delete_message(bot_token, chat_id_str, prompt_msg_id)
# 追加到 hint 历史
hint_history = conv.get("hint_history", [])
hint_history.append(text)
# 更新消息为 AI 思考中
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": conv["summary_msg_id"],
"text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
})
# 丢到 AI 线程池处理,传入完整 hint 历史
_ai_pool.submit(_do_ai_hint, bot_token, cfg, chat_id, conv, ctx, hint_history)
# 丢到 AI 线程池处理
_ai_pool.submit(_do_ai_hint, bot_token, cfg, chat_id, conv, ctx, text)
def _show_ai_suggestions(bot_token: str, chat_id: int, msg_id: int,
@@ -796,7 +725,7 @@ def _do_send_reply(bot_token: str, cfg: Config,
to_addr = parseaddr(ctx["sender"])[1]
if not to_addr:
to_addr = ctx["sender"]
subject = ctx.get("original_subject") or ctx["subject"]
subject = ctx["subject"]
# 下载 Telegram 附件到本地
local_files = []
-149
View File
@@ -1,149 +0,0 @@
"""测试摘要键盘渲染:推广邮件的退订链接按钮必须显示且排最前。"""
from src.tg_bot import (
_ensure_unsub_link,
_is_unsub_link,
_link_buttons,
_summary_keyboard,
)
def _rows_of(kb: dict) -> list[list[dict]]:
return kb["inline_keyboard"]
def test_promotion_with_unsub_link_shows_button_first():
"""推广邮件:AI 提取到退订链接时必须显示,且排在第一行"""
links = [
{"text": "Manage Preferences", "url": "https://e.example.com/pref"},
{"text": "推广活动详情", "url": "https://e.example.com/deal"},
]
kb = _summary_keyboard(can_reply=False, is_promotion=True, links=links)
rows = _rows_of(kb)
first = rows[0][0]
assert first["url"] == "https://e.example.com/pref"
assert "🔕" in first["text"]
# 其余按钮仍然存在
texts = [b["text"] for r in rows for b in r]
assert "告诉我更多" in texts
assert "查看原文" in texts
def test_promotion_without_links_keeps_default_buttons():
"""推广邮件:没有链接时保持原有按钮"""
kb = _summary_keyboard(can_reply=False, is_promotion=True, links=None)
rows = _rows_of(kb)
assert len(rows) == 2
assert rows[0][0]["text"] == "告诉我更多"
assert rows[1][0]["text"] == "查看原文"
def test_promotion_with_only_non_unsub_link():
"""推广邮件:只有非退订链接时也正常渲染(🔗),不冒充退订按钮"""
links = [{"text": "查看活动", "url": "https://e.example.com/event"}]
kb = _summary_keyboard(can_reply=False, is_promotion=True, links=links)
rows = _rows_of(kb)
assert rows[0][0]["text"] == "🔗 查看活动"
assert rows[1][0]["text"] == "告诉我更多"
assert rows[2][0]["text"] == "查看原文"
def test_normal_email_links_still_rendered():
"""普通邮件:链接按钮行为不变"""
links = [{"text": "查看账单", "url": "https://e.example.com/bill"}]
kb = _summary_keyboard(can_reply=True, is_promotion=False, links=links)
rows = _rows_of(kb)
texts = [b["text"] for r in rows for b in r]
assert "🔗 查看账单" in texts
def test_unsub_detection_variants():
"""退订链接识别:文字与 URL 关键词变体"""
assert _is_unsub_link({"text": "Unsubscribe", "url": "https://x.com"})
assert _is_unsub_link({"text": "退订", "url": "https://x.com"})
assert _is_unsub_link({"text": "点击这里取消订阅", "url": "https://x.com"})
assert _is_unsub_link({"text": "管理偏好设置", "url": "https://x.com"})
assert _is_unsub_link({"text": "更多", "url": "https://x.com/unsubscribe"})
assert _is_unsub_link({"text": "Preferences", "url": "https://x.com/email-pref"})
assert not _is_unsub_link({"text": "查看详情", "url": "https://x.com/deal"})
def test_unsub_link_sorted_before_other_links():
"""退订链接即使不在 links 第一位也会被排到最前"""
links = [
{"text": "查看详情", "url": "https://x.com/a"},
{"text": "unsubscribe", "url": "https://x.com/b"},
]
rows = _link_buttons(links)
assert rows[0][0]["url"] == "https://x.com/b"
assert rows[1][0]["url"] == "https://x.com/a"
def test_invalid_url_filtered():
"""非法 URL(非 http、多重协议头)被过滤"""
links = [
{"text": "退订", "url": "javascript:alert(1)"},
{"text": "退订", "url": "https://x.com/https://evil.com"},
{"text": "退订", "url": "https://ok.example.com/unsub"},
]
rows = _link_buttons(links)
assert len(rows) == 1
assert rows[0][0]["url"] == "https://ok.example.com/unsub"
def test_long_unsub_url_rendered():
"""长 tracking 退订链接(如 OpenAI 的 >700 字符)必须渲染,不能被长度过滤误杀"""
url = "https://r.openai.com/asm/unsubscribe/?user_id=108370056&data=" + "x" * 700
assert len(url) > 200
links = [{"text": "取消订阅", "url": url}]
rows = _link_buttons(links)
assert len(rows) == 1
assert rows[0][0]["url"] == url
assert "🔕" in rows[0][0]["text"]
def test_oversized_url_filtered():
"""超过 2048 的 URL 仍被过滤"""
url = "https://example.com/" + "x" * 2100
links = [{"text": "退订", "url": url}]
rows = _link_buttons(links)
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)") == []
Reference in new issue
Block a user
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.