FEDEO/scripts/download-gmail-attachments.py

#!/usr/bin/env python3
"""Lädt Anhänge aus einem Gmail-Postfach per IMAP herunter."""

from __future__ import annotations

import argparse
import email
import getpass
import imaplib
import os
import re
import sys
from dataclasses import dataclass
from datetime import datetime
from email.header import decode_header, make_header
from email.message import Message
from email.utils import parsedate_to_datetime
from pathlib import Path


GMAIL_IMAP_HOST = "imap.gmail.com"
GMAIL_IMAP_PORT = 993


@dataclass(frozen=True)
class DownloadStats:
    messages_seen: int = 0
    attachments_saved: int = 0
    attachments_skipped: int = 0


def decode_mime_header(value: str | None, fallback: str = "") -> str:
    if not value:
        return fallback
    try:
        return str(make_header(decode_header(value))).strip()
    except Exception:
        return value.strip()


def safe_filename(name: str, fallback: str) -> str:
    cleaned = decode_mime_header(name, fallback)
    cleaned = cleaned.replace("\x00", "")
    cleaned = re.sub(r"[\\/:\*\?\"<>\|]+", "_", cleaned)
    cleaned = re.sub(r"\s+", " ", cleaned).strip(" .")
    return cleaned or fallback


def unique_path(path: Path, overwrite: bool) -> Path:
    if overwrite or not path.exists():
        return path

    stem = path.stem
    suffix = path.suffix
    parent = path.parent
    counter = 2

    while True:
        candidate = parent / f"{stem}-{counter}{suffix}"
        if not candidate.exists():
            return candidate
        counter += 1


def quote_imap_string(value: str) -> str:
    """Maskiert einen Text als quotierten IMAP-String."""
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def debug(args: argparse.Namespace, message: str) -> None:
    if args.debug:
        print(f"KI-AGENT: DEBUG: {message}")


def mailbox_name_from_list_entry(entry: bytes) -> str | None:
    """Liest den Postfachnamen aus einer IMAP-LIST-Antwort."""
    decoded = entry.decode("utf-8", errors="replace")
    match = re.search(r'(?:"((?:[^"\\]|\\.)*)"|([^ ]+))$', decoded)
    if not match:
        return None
    if match.group(1) is not None:
        return re.sub(r"\\([\\\"])", r"\1", match.group(1))
    return match.group(2)


def find_all_mailbox(mailbox_data: list[bytes]) -> str | None:
    for entry in mailbox_data:
        flags = entry.split(b")", 1)[0].lower()
        if b"\\all" in flags:
            return mailbox_name_from_list_entry(entry)
    return None


def message_folder(message: Message, message_id: bytes) -> str:
    date_header = decode_mime_header(message.get("Date"))
    subject = safe_filename(decode_mime_header(message.get("Subject"), "ohne-betreff"), "ohne-betreff")

    date_prefix = "ohne-datum"
    if date_header:
        try:
            date_prefix = parsedate_to_datetime(date_header).strftime("%Y-%m-%d_%H-%M-%S")
        except Exception:
            pass

    short_id = message_id.decode("ascii", errors="ignore") or "mail"
    return safe_filename(f"{date_prefix}_{short_id}_{subject[:80]}", f"mail-{short_id}")


def build_search_query(args: argparse.Namespace) -> str:
    parts = [args.search]
    if args.since:
        parts.append(f'SINCE "{args.since}"')
    if args.before:
        parts.append(f'BEFORE "{args.before}"')
    return f"({' '.join(parts)})"


def parse_imap_date(value: str | None, option_name: str) -> str | None:
    if not value:
        return None
    try:
        return datetime.strptime(value, "%Y-%m-%d").strftime("%d-%b-%Y")
    except ValueError:
        raise argparse.ArgumentTypeError(f"{option_name} muss im Format YYYY-MM-DD angegeben werden.")


def iter_attachment_parts(message: Message):
    attachment_index = 1
    for part in message.walk():
        if part.is_multipart():
            continue

        disposition = part.get_content_disposition()
        filename = part.get_filename()
        if disposition != "attachment" and not filename:
            continue

        fallback = f"anhang-{attachment_index}"
        attachment_index += 1
        yield part, safe_filename(filename or fallback, fallback)


def download_attachments(args: argparse.Namespace) -> DownloadStats:
    password = args.password or os.environ.get("GMAIL_APP_PASSWORD")
    if not password:
        debug(args, "Kein Passwortparameter gefunden; frage das App-Passwort interaktiv ab.")
        password = getpass.getpass("Gmail App-Passwort: ")
    else:
        source = "--password" if args.password else "GMAIL_APP_PASSWORD"
        debug(args, f"App-Passwort aus {source} geladen (Inhalt wird nicht ausgegeben).")

    stats = DownloadStats()

    debug(args, f"Verbinde verschlüsselt mit {GMAIL_IMAP_HOST}:{GMAIL_IMAP_PORT}.")
    with imaplib.IMAP4_SSL(GMAIL_IMAP_HOST, GMAIL_IMAP_PORT) as client:
        debug(args, "TLS-Verbindung zum Gmail-IMAP-Server steht.")
        debug(args, f"Melde Benutzer {args.email} an.")
        client.login(args.email, password)
        debug(args, "Login erfolgreich.")

        mailbox_data: list[bytes] = []
        if args.list_mailboxes or args.mailbox is None:
            debug(args, "Rufe die verfügbaren IMAP-Postfächer ab.")
            status, mailbox_data = client.list()
            if status != "OK":
                raise RuntimeError("IMAP-Postfächer konnten nicht gelesen werden.")
            mailbox_data = mailbox_data or []
            debug(args, f"{len(mailbox_data)} Postfächer gefunden.")

        if args.list_mailboxes:
            for mailbox in mailbox_data:
                print(f"KI-AGENT: {mailbox.decode('utf-8', errors='replace')}")
            return stats

        mailbox = args.mailbox or find_all_mailbox(mailbox_data)
        if not mailbox:
            raise RuntimeError(
                "Das Gmail-Postfach für alle Nachrichten konnte nicht automatisch gefunden werden. "
                "Nutze --list-mailboxes und gib den Namen anschließend mit --mailbox an."
            )
        if args.mailbox:
            debug(args, f'Verwende das angegebene Postfach "{mailbox}".')
        else:
            debug(args, f'Postfach für alle Nachrichten automatisch erkannt: "{mailbox}".')

        output_dir = args.output.expanduser().resolve()
        debug(args, f"Lege den Zielordner an, falls nötig: {output_dir}")
        output_dir.mkdir(parents=True, exist_ok=True)

        debug(args, f'Öffne das Postfach "{mailbox}" schreibgeschützt.')
        status, select_data = client.select(quote_imap_string(mailbox), readonly=True)
        if status != "OK":
            details = b" ".join(select_data or []).decode("utf-8", errors="replace")
            raise RuntimeError(
                f'IMAP-Postfach "{mailbox}" konnte nicht geöffnet werden. '
                f"Nutze --list-mailboxes, um verfügbare Postfächer anzuzeigen. {details}".strip()
            )
        debug(args, "Postfach erfolgreich geöffnet.")

        search_query = build_search_query(args)
        debug(args, f"Starte IMAP-Suche: {search_query}")
        status, search_data = client.search(None, search_query)
        if status != "OK":
            raise RuntimeError("IMAP-Suche ist fehlgeschlagen.")

        message_ids = search_data[0].split()
        stats = DownloadStats(messages_seen=len(message_ids))
        debug(args, f"Suche erfolgreich: {len(message_ids)} Mails werden verarbeitet.")

        for position, message_id in enumerate(message_ids, start=1):
            debug(args, f"Lese Mail {position}/{len(message_ids)} mit IMAP-ID {message_id.decode(errors='replace')}.")
            status, fetch_data = client.fetch(message_id, "(RFC822)")
            if status != "OK" or not fetch_data:
                print(f"KI-AGENT: Mail {message_id!r} konnte nicht gelesen werden.", file=sys.stderr)
                continue

            raw_message = next((item[1] for item in fetch_data if isinstance(item, tuple)), None)
            if not raw_message:
                continue

            message = email.message_from_bytes(raw_message)
            debug(args, f'Mail gelesen: "{decode_mime_header(message.get("Subject"), "ohne Betreff")}".')
            target_dir = output_dir / message_folder(message, message_id) if args.group_by_message else output_dir
            target_dir.mkdir(parents=True, exist_ok=True)

            attachment_count = 0
            for part, filename in iter_attachment_parts(message):
                attachment_count += 1
                debug(args, f'Verarbeite Anhang "{filename}" aus Mail {position}.')
                payload = part.get_payload(decode=True)
                if payload is None:
                    stats = DownloadStats(
                        stats.messages_seen,
                        stats.attachments_saved,
                        stats.attachments_skipped + 1,
                    )
                    continue

                target = unique_path(target_dir / filename, args.overwrite)
                target.write_bytes(payload)
                stats = DownloadStats(
                    stats.messages_seen,
                    stats.attachments_saved + 1,
                    stats.attachments_skipped,
                )
                print(f"KI-AGENT: Gespeichert: {target}")

            if attachment_count == 0:
                debug(args, f"Mail {position} enthält keinen Anhang.")

            if args.progress and position % args.progress == 0:
                print(f"KI-AGENT: {position}/{len(message_ids)} Mails verarbeitet.")

    debug(args, "IMAP-Verbindung ordnungsgemäß geschlossen.")
    return stats


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Lädt alle Anhänge aus einem Gmail-Postfach per IMAP herunter.",
    )
    parser.add_argument("--email", required=True, help="Gmail-Adresse, z. B. name@gmail.com")
    parser.add_argument(
        "--password",
        help="Gmail App-Passwort. Sicherer: Umgebungsvariable GMAIL_APP_PASSWORD verwenden.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("gmail-anhaenge"),
        help="Zielordner für Anhänge. Standard: gmail-anhaenge",
    )
    parser.add_argument(
        "--mailbox",
        help="IMAP-Postfach/Label. Standard: automatisch das mit \\All markierte Gmail-Postfach.",
    )
    parser.add_argument(
        "--list-mailboxes",
        action="store_true",
        help="Verfügbare IMAP-Postfächer anzeigen und beenden.",
    )
    parser.add_argument(
        "--search",
        default="ALL",
        help='IMAP-Suche. Beispiele: ALL, UNSEEN, FROM "person@example.com"',
    )
    parser.add_argument("--since", type=lambda value: parse_imap_date(value, "--since"), help="Ab Datum YYYY-MM-DD")
    parser.add_argument("--before", type=lambda value: parse_imap_date(value, "--before"), help="Vor Datum YYYY-MM-DD")
    parser.add_argument(
        "--group-by-message",
        action="store_true",
        help="Anhänge je Mail in einen eigenen Unterordner speichern.",
    )
    parser.add_argument("--overwrite", action="store_true", help="Bestehende Dateien überschreiben.")
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Detaillierte Statusausgaben ausgeben, ohne das Passwort anzuzeigen.",
    )
    parser.add_argument(
        "--progress",
        type=int,
        default=100,
        help="Fortschritt alle N Mails ausgeben. 0 deaktiviert die Ausgabe.",
    )
    return parser.parse_args()


def main() -> int:
    try:
        stats = download_attachments(parse_args())
    except KeyboardInterrupt:
        print("\nKI-AGENT: Abgebrochen.", file=sys.stderr)
        return 130
    except Exception as exc:
        print(f"KI-AGENT: Fehler: {exc}", file=sys.stderr)
        return 1

    print(
        "KI-AGENT: Fertig. "
        f"{stats.messages_seen} Mails geprüft, "
        f"{stats.attachments_saved} Anhänge gespeichert, "
        f"{stats.attachments_skipped} Anhänge übersprungen."
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())