email-assistant/scripts/digest.py

#!/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()