Zichao Lin
e2826a3e3b
- Multi-account IMAP email polling with UID tracking - DeepSeek API integration with JSON Mode structured output - Telegram notification with formatted MarkdownV2 message - YAML config with dataclass-based type validation - Graceful shutdown on SIGINT/SIGTERM - 60s default polling interval
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import requests
|
|
from src.config import TelegramConfig
|
|
|
|
|
|
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()
|
|
|
|
|
|
_priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}
|
|
|
|
|
|
def format_summary(data: dict) -> str:
|
|
priority = data.get("priority", "medium")
|
|
icon = _priority_icon.get(priority, "⚪")
|
|
|
|
lines = [
|
|
f"*📧 新邮件摘要*",
|
|
f"━━━━━━━━━━━━━━━━━━",
|
|
f"*发件人:* {_escape(data.get('sender', '未知'))}",
|
|
f"*主题:* {_escape(data.get('subject', '无主题'))}",
|
|
f"*优先级:* {icon} {priority.upper()}",
|
|
"",
|
|
_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("")
|
|
|
|
points = data.get("key_points", [])
|
|
if points:
|
|
lines.append(f"*关键要点:*")
|
|
for p in points:
|
|
lines.append(f" • {_escape(p)}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _escape(text: str) -> str:
|
|
special = "_*[]()~`>#+-=|{}.!"
|
|
for ch in special:
|
|
text = text.replace(ch, f"\\{ch}")
|
|
return text
|