email-assistant/scripts/sqlite_search.py

#!/usr/bin/env python3
"""
sqlite_search.py — FTS5-поиск по архиву писем.

Использование:
  python3 sqlite_search.py "запрос"
  python3 sqlite_search.py "запрос" --limit 20
  python3 sqlite_search.py "тема:отчёт" --folder "INBOX/!Отчеты"
  python3 sqlite_search.py "from:example@mail.ru" --body

Операторы FTS5 (по умолчанию AND):
  "точная фраза"   — точное совпадение
  OR               — любой из терминов
  -исключить       — исключить термин
  prefix*          — префикс (wildcard)
"""

import argparse
import sqlite3
import sys
from datetime import datetime
from pathlib import Path

DB_PATH = Path("/opt/hermes/email/mail_index.db")


def get_db():
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn


def format_date(d):
    """Привести дату к читаемому виду."""
    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 %H:%M")
        except ValueError:
            continue
    return d[:10]


def search(query, limit=15, folder=None, search_body=False):
    conn = get_db()
    c = conn.cursor()

    # Обрабатываем префиксы from:/subject:/body:
    from_filter = None
    subject_filter = None
    body_only = search_body

    # Если запрос содержит from: или subject: — выносим в WHERE
    import re
    mf = re.search(r'\bfrom:(\S+)', query)
    if mf:
        from_filter = mf.group(1).replace('"', '').replace("'", "")
        query = query.replace(mf.group(0), '').strip()

    ms = re.search(r'\bsubject:(\S+)', query)
    if ms:
        subject_filter = ms.group(1).replace('"', '').replace("'", "")
        query = query.replace(ms.group(0), '').strip()

    if not query:
        print("❌ Укажите поисковый запрос")
        sys.exit(1)

    # Строим SQL
    if body_only:
        # Ищем в FTS5 с телом письма
        sql = """
            SELECT e.path, e.folder, e.date, e.from_addr, e.subject,
                   e.body_preview, e.contacts_extracted, e.contacts_skipped
            FROM email_fts f
            JOIN emails e ON e.rowid = f.rowid
            WHERE email_fts MATCH ?
        """
        params = [query]
    else:
        # Стандартный поиск — FTS5 по subject + from + to
        sql = """
            SELECT e.path, e.folder, e.date, e.from_addr, e.subject,
                   e.body_preview, e.contacts_extracted, e.contacts_skipped
            FROM email_fts f
            JOIN emails e ON e.rowid = f.rowid
            WHERE email_fts MATCH ?
        """
        params = [query]

    if folder:
        sql += " AND e.folder = ?"
        params.append(folder)

    # Дополнительные фильтры для точного поиска
    conditions = []
    if from_filter:
        conditions.append("e.from_addr LIKE ?")
        params.append(f"%{from_filter}%")
    if subject_filter:
        conditions.append("e.subject LIKE ?")
        params.append(f"%{subject_filter}%")

    if conditions:
        sql += " AND " + " AND ".join(conditions)

    sql += " ORDER BY e.date DESC LIMIT ?"
    params.append(limit)

    try:
        rows = c.execute(sql, params).fetchall()
    except sqlite3.OperationalError as e:
        print(f"❌ Ошибка FTS5: {e}")
        print()
        print("Подсказки:")
        print("  Используйте кавычки для точной фразы: \"Иван Иванов\"")
        print("  Избегайте спецсимволов: * ? - (их не должно быть в простых словах)")
        print("  Пример: python3 sqlite_search.py 'битрикс OR контрагент'")
        conn.close()
        sys.exit(1)

    if not rows:
        print("  Ничего не найдено")
        conn.close()
        return

    print(f"  Найдено: {len(rows)} писем\n")

    for r in rows:
        extr = "+" if r["contacts_extracted"] else "·"
        if r["contacts_skipped"]:
            extr = "-"
        date_fmt = format_date(r["date"])
        folder_short = r["folder"].replace("INBOX/", "") if r["folder"].startswith("INBOX") else r["folder"]

        print(f"  [{extr}] {date_fmt} | {folder_short:>20}")
        print(f"       От: {r['from_addr'][:60]}")
        print(f"       Тема: {r['subject'][:80]}")
        print(f"       {EMAIL_ROOT / r['path']}")
        print()

    conn.close()


def main():
    global EMAIL_ROOT
    from pathlib import Path as P
    EMAIL_ROOT = P("/opt/hermes/email")

    parser = argparse.ArgumentParser(description="FTS5-поиск по архиву писем")
    parser.add_argument("query", type=str, nargs="?", help="Поисковый запрос")
    parser.add_argument("--limit", "-l", type=int, default=15, help="Макс. результатов")
    parser.add_argument("--folder", "-f", type=str, help="Фильтр по папке")
    parser.add_argument("--body", "-b", action="store_true", help="Поиск с учётом тела письма")
    args = parser.parse_args()

    if not args.query:
        parser.print_help()
        print()
        print("Примеры:")
        print(f"  {sys.argv[0]} 'Стороженко'")
        print(f"  {sys.argv[0]} 'from:example@mail'")
        print(f"  {sys.argv[0]} 'subject:отчёт' --folder 'INBOX/!Отчеты'")
        print(f"  {sys.argv[0]} 'битрикс OR контрагент'")
        sys.exit(0)

    search(args.query, args.limit, args.folder, args.body)


if __name__ == "__main__":
    main()