219 lines
8.1 KiB
Python
219 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
digest.py — Еженедельный дайджест почты.
|
|
|
|
Собирает все письма за 7 дней, группирует по папкам,
|
|
вызывает LLM для генерации краткого дайджеста.
|
|
|
|
Использование:
|
|
python3 digest.py — дайджест за 7 дней
|
|
python3 digest.py --days 14 — за 14 дней
|
|
python3 digest.py --folder INBOX — только INBOX
|
|
python3 digest.py --output — только файл, без вывода в stdout
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import sqlite3
|
|
|
|
# ─── Конфиг ───────────────────────────────────────────────────────────────────
|
|
|
|
DB_PATH = Path("/opt/hermes/email/mail_index.db")
|
|
DIGEST_DIR = Path("/opt/hermes/email/digests")
|
|
OLLAMA_URL = "http://localhost:11434/api/generate"
|
|
OLLAMA_MODEL = "qwen3:8b"
|
|
OLLAMA_TIMEOUT = 60
|
|
|
|
# ─── Промпты ──────────────────────────────────────────────────────────────────
|
|
|
|
SUMMARIZE_PROMPT = """Ты — ассистент, который делает ежедневный дайджест корпоративной почты.
|
|
|
|
Ниже — список писем за последние {days} дней, сгруппированный по папкам.
|
|
Твоя задача — написать КРАТКИЙ дайджест на русском языке для руководителя.
|
|
|
|
Формат:
|
|
1. Заголовок: "📬 Дайджест почты — {date_range}"
|
|
2. Общая статистика: сколько писем, сколько папок затронуто
|
|
3. По каждой папке — 1-3 предложения: основные темы, важные отправители, ключевые решения
|
|
4. Выдели особенно важные письма (от руководства, по тендерам, договорам, финансам)
|
|
|
|
ВАЖНО:
|
|
- Пиши ТОЛЬКО на русском
|
|
- Будь краток — максимум 300 слов
|
|
- Не перечисляй каждое письмо — выделяй тренды и главное
|
|
- Если тема понятна из темы письма — группируй
|
|
- Пропусти технический мусор (уведомления систем, автоответы)
|
|
|
|
Вот данные для анализа:
|
|
|
|
{folder_groups}"""
|
|
|
|
# ─── Вспомогательные ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def get_db():
|
|
conn = sqlite3.connect(str(DB_PATH))
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def call_llm(prompt, max_retries=2):
|
|
"""Вызвать Qwen через Ollama API."""
|
|
payload = json.dumps({
|
|
"model": OLLAMA_MODEL,
|
|
"prompt": prompt,
|
|
"stream": False,
|
|
"options": {
|
|
"temperature": 0.3,
|
|
"num_predict": 2048,
|
|
}
|
|
}).encode("utf-8")
|
|
|
|
for attempt in range(max_retries + 1):
|
|
if attempt > 0:
|
|
time.sleep(2)
|
|
req = urllib.request.Request(
|
|
OLLAMA_URL,
|
|
data=payload,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=OLLAMA_TIMEOUT)
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
return data.get("response", "").strip()
|
|
except (urllib.error.URLError, json.JSONDecodeError, TimeoutError) as e:
|
|
if attempt < max_retries:
|
|
continue
|
|
return f"⚠ Ошибка LLM: {e}"
|
|
|
|
return "⚠ Не удалось сгенерировать дайджест"
|
|
|
|
|
|
def format_date(d):
|
|
"""Привести дату к dd.mm.yyyy."""
|
|
if not d:
|
|
return "—"
|
|
for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S%z",
|
|
"%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
|
try:
|
|
dt = datetime.strptime(d.rstrip("Z"), fmt)
|
|
return dt.strftime("%d.%m.%Y")
|
|
except ValueError:
|
|
continue
|
|
return d[:10]
|
|
|
|
|
|
# ─── Основная логика ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def build_digest(days=7, folder_filter=None):
|
|
"""Собрать дайджест за N дней."""
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
|
|
cutoff = datetime.now() - timedelta(days=days)
|
|
|
|
# Берём письма за период
|
|
query = """
|
|
SELECT folder, date, from_addr, subject, contacts_extracted, path
|
|
FROM emails
|
|
WHERE date >= ?
|
|
ORDER BY date DESC
|
|
"""
|
|
params = [cutoff.strftime("%Y-%m-%d")]
|
|
|
|
if folder_filter:
|
|
query += " AND folder = ?"
|
|
params.append(folder_filter)
|
|
|
|
rows = c.execute(query, params).fetchall()
|
|
conn.close()
|
|
|
|
if not rows:
|
|
msg = f" Нет писем за последние {days} дней"
|
|
print(msg)
|
|
return msg
|
|
|
|
# Группируем по папкам
|
|
groups = {}
|
|
for r in rows:
|
|
f = r["folder"]
|
|
if f not in groups:
|
|
groups[f] = []
|
|
groups[f].append(r)
|
|
|
|
# Формируем текстовое представление для LLM
|
|
folder_parts = []
|
|
total = 0
|
|
for fname, emails in sorted(groups.items()):
|
|
total += len(emails)
|
|
display_name = fname.replace("INBOX/", "") if fname.startswith("INBOX") else fname
|
|
items = []
|
|
for e in emails:
|
|
date_str = format_date(e["date"])
|
|
extr = "✓" if e["contacts_extracted"] else " "
|
|
from_short = e["from_addr"][:40]
|
|
subj = (e["subject"] or "(без темы)")[:60]
|
|
items.append(f" [{extr}] {date_str} | {from_short} | {subj}")
|
|
folder_parts.append(f"─── {display_name} ({len(emails)} писем) ───\n" + "\n".join(items))
|
|
|
|
folder_text = "\n\n".join(folder_parts)
|
|
|
|
# Дата-диапазон
|
|
date_range = f"{format_date((datetime.now() - timedelta(days=days)).isoformat())} — {datetime.now().strftime('%d.%m.%Y')}"
|
|
|
|
prompt = SUMMARIZE_PROMPT.format(
|
|
days=days,
|
|
date_range=date_range,
|
|
folder_groups=folder_text
|
|
)
|
|
|
|
print(" Генерирую дайджест через LLM...", flush=True)
|
|
digest = call_llm(prompt)
|
|
|
|
# Форматируем
|
|
header = f"📬 Дайджест корпоративной почты"
|
|
subheader = f" {total} писем из {len(groups)} папок | {date_range}"
|
|
divider = "─" * 60
|
|
|
|
result = f"{header}\n{subheader}\n{divider}\n\n{digest}\n\n{divider}"
|
|
|
|
# Сохраняем в файл
|
|
DIGEST_DIR.mkdir(parents=True, exist_ok=True)
|
|
filename = datetime.now().strftime("digest-%Y-%m-%d.md")
|
|
filepath = DIGEST_DIR / filename
|
|
filepath.write_text(result, encoding="utf-8")
|
|
|
|
print(f"\n✅ Дайджест сохранён: {filepath}")
|
|
print()
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Еженедельный дайджест почты")
|
|
parser.add_argument("--days", type=int, default=7, help="Глубина в днях")
|
|
parser.add_argument("--folder", type=str, help="Фильтр по папке")
|
|
parser.add_argument("--output", choices=["stdout", "file", "both"], default="both",
|
|
help="Куда вывести результат")
|
|
args = parser.parse_args()
|
|
|
|
result = build_digest(days=args.days, folder_filter=args.folder)
|
|
|
|
if args.output in ("stdout", "both"):
|
|
print()
|
|
print(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|