149 lines
6.7 KiB
Python
149 lines
6.7 KiB
Python
import json
|
||
import logging
|
||
import time
|
||
from typing import Any
|
||
import requests
|
||
from src.config import AIConfig
|
||
from src.retry import retry
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。
|
||
|
||
返回格式必须严格遵循以下 JSON schema：
|
||
{
|
||
"subject": "你概括的简短标题（10字以内）",
|
||
"recipient": "收件邮箱地址",
|
||
"sender": "发件人",
|
||
"category": "normal/promotion/notification",
|
||
"summary": "50字以内的核心摘要",
|
||
"verification_code": "",
|
||
"app_name": "",
|
||
"links": [{"text": "按钮文字", "url": "https://..."}],
|
||
"can_reply": true/false,
|
||
"reply_suggestions": [
|
||
{"text": "回复内容", "is_simple": true/false},
|
||
{"text": "回复内容", "is_simple": true/false},
|
||
{"text": "回复内容", "is_simple": true/false}
|
||
]
|
||
}
|
||
|
||
其中：
|
||
- subject 不能照抄原邮件主题，必须是你总结生成的简短标题
|
||
- category: 邮件类型。推广邮件（营销、促销、广告、 newsletter 推广等）标记为 "promotion"；验证码邮件标记为 "notification"；普通邮件标记为 "normal"
|
||
- verification_code: 仅当 category 为 "notification" 且邮件包含验证码时填写，提取完整的验证码字符串（纯数字、字母数字混合等均可，如「123456」「abc-789」「701383」）。非验证码邮件留空字符串
|
||
- app_name: 仅当 category 为 "notification" 且邮件包含验证码时填写，填写发送验证码的应用/平台名称（如「GitHub」「微信」「Kraken」）。非验证码邮件留空字符串
|
||
- links: 仅提取以下两类链接：1）邮件明确要求用户点击才能完成某操作的链接（如确认邮箱、重置密码、查看账单、审批请求、查看通知详情等）；2）退订/取消订阅链接。不要提取推广性质的链接（如下载App、注册领优惠、查看更多商品等营销链接）、追踪像素、邮件签名中的社交链接。每条包含简短描述文字和完整 URL。最多 3 条。没有符合条件的链接时为空数组
|
||
- summary 要简洁，把需要知道的信息浓缩成一句话，不需要分条列出
|
||
- 当 category 为 "promotion" 时，summary 直接写成一句话：「[发送方名称] 在推广 [推广的产品/服务/活动]」，例如「腾讯云在推广双十一云服务器折扣」。此时 can_reply 为 false，reply_suggestions 为空数组
|
||
- can_reply 为 false 表示无需回复或不适合建议回复（此时忽略 reply_suggestions）
|
||
- is_simple 为 true 表示纯确认性回复（如"好的""收到"），用户选择后可直发
|
||
- is_simple 为 false 表示涉及实质内容，需要用户确认再发送
|
||
- 每条回复控制在 50 字以内
|
||
只返回 JSON，不要包含任何其他文字。"""
|
||
|
||
|
||
@retry()
|
||
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}"
|
||
|
||
payload = {
|
||
"model": ai_cfg.model,
|
||
"messages": [
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "user", "content": content},
|
||
],
|
||
"response_format": {"type": "json_object"},
|
||
}
|
||
|
||
return _call_deepseek(ai_cfg, payload)
|
||
|
||
|
||
PROMO_DETAIL_PROMPT = """你是一个邮件分析助手。用户想了解更多关于这封推广邮件的详情。
|
||
请根据邮件原文，输出 3-5 条关键信息，每条一句话，简洁明了。包括：优惠力度、适用条件、截止时间、核心卖点等用户做决策需要知道的信息。
|
||
只返回一个 JSON: {"details": ["信息1", "信息2", ...]}
|
||
只返回 JSON，不要包含任何其他文字。"""
|
||
|
||
|
||
@retry()
|
||
def expand_promo_detail(ai_cfg: AIConfig, sender: str, subject: str, body: str) -> list[str]:
|
||
logger.info("AI 推广详情请求: sender=%s subject=%s", sender, subject)
|
||
content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body}"
|
||
payload = {
|
||
"model": ai_cfg.model,
|
||
"messages": [
|
||
{"role": "system", "content": PROMO_DETAIL_PROMPT},
|
||
{"role": "user", "content": content},
|
||
],
|
||
"response_format": {"type": "json_object"},
|
||
}
|
||
result = _call_deepseek(ai_cfg, payload)
|
||
return result.get("details", [])
|
||
|
||
|
||
MORE_REPLIES_PROMPT = """你是一个邮件回复助手。根据以下邮件内容生成3个完全不同的建议回复。
|
||
|
||
以下是一些已经生成过的回复，请生成全新的回复，不要与已有回复重复。
|
||
|
||
返回 JSON:
|
||
{
|
||
"suggestions": [
|
||
{"text": "回复内容"},
|
||
{"text": "回复内容"},
|
||
{"text": "回复内容"}
|
||
]
|
||
}
|
||
每条回复控制在 50 字以内。
|
||
只返回 JSON，不要包含任何其他文字。"""
|
||
|
||
|
||
@retry()
|
||
def generate_more_replies(ai_cfg: AIConfig, sender: str, subject: str, body: str,
|
||
previous_suggestions: list[dict], user_hint: str = "") -> list[dict]:
|
||
logger.info("AI 换批请求: sender=%s subject=%s user_hint=%s prev=%d 条",
|
||
sender, subject, user_hint or "(无)", len(previous_suggestions))
|
||
content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body}\n\n已生成过的回复:\n"
|
||
for i, s in enumerate(previous_suggestions, 1):
|
||
content += f"{i}. {s['text']}\n"
|
||
if user_hint:
|
||
content += f"\n用户要求: {user_hint}"
|
||
|
||
payload = {
|
||
"model": ai_cfg.model,
|
||
"messages": [
|
||
{"role": "system", "content": MORE_REPLIES_PROMPT},
|
||
{"role": "user", "content": content},
|
||
],
|
||
"response_format": {"type": "json_object"},
|
||
}
|
||
result = _call_deepseek(ai_cfg, payload)
|
||
return result.get("suggestions", [])
|
||
|
||
|
||
def _call_deepseek(ai_cfg: AIConfig, payload: dict) -> dict:
|
||
model = payload.get("model", ai_cfg.model)
|
||
logger.info("DeepSeek API 请求: model=%s", model)
|
||
t0 = time.time()
|
||
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,
|
||
json=payload,
|
||
timeout=120,
|
||
)
|
||
elapsed = time.time() - t0
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
usage = result.get("usage", {})
|
||
prompt_tokens = usage.get("prompt_tokens", "?")
|
||
completion_tokens = usage.get("completion_tokens", "?")
|
||
raw = result["choices"][0]["message"]["content"]
|
||
logger.info("DeepSeek API 响应: %.2fs, token=%d+%d, 输出长度=%d",
|
||
elapsed, prompt_tokens, completion_tokens, len(raw))
|
||
return json.loads(raw)
|