janssen/mailstore/mailstore_map_v1.0.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
================================================================================
 Nazev:   mailstore_map_v1.0.py
 Verze:   1.0
 Datum:   2026-06-11
 Autor:   Vladimir Buzalka (asistovano Claude)
 Popis:   Vykresli "mapu" jedne MailStore schranky - strom slozek z
          Administration API (GetChildFolders) + celkovy pocet zprav schranky
          z GetFolderStatistics.

          Argument = nazev schranky (top-level slozka v MailStore archivu),
          napr. "vladimir.buzalka@buzalka.cz" nebo "lenka.hanzalova".
          Seznam dostupnych schranek: --list (vola GetUsers/GetChildFolders root).

 Zdroj:   MailStore Server Administration API, HTTPS port 8463.
          Auth: admin / heslo (Basic). Parametry jako form-body. Async operace
          (GetFolderStatistics) se poluji pres /api/get-status.

 Pozn.:   API umi jen strukturu + souhrnne pocty per schranka. Pocty zprav per
          jednotliva slozka API levne nedava - to bude dalsi krok (IMAP STATUS).

 Spusteni:
   python mailstore_map_v1.0.py "lenka.hanzalova"
   python mailstore_map_v1.0.py "vladimir.buzalka@buzalka.cz" --no-stats
   python mailstore_map_v1.0.py --list
================================================================================
"""

from __future__ import annotations

import argparse
import json
import ssl
import sys
import time
import urllib.parse
import urllib.request
from base64 import b64encode

# --- konfigurace ------------------------------------------------------------
HOST = "192.168.1.53"
PORT = 8463
USER = "admin"
PASS = "*$N(B)vMUym!%"

BASE = f"https://{HOST}:{PORT}/api"
_AUTH = "Basic " + b64encode(f"{USER}:{PASS}".encode()).decode()
_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE


# --- API helper -------------------------------------------------------------

def _post(path: str, params: dict | None = None) -> dict:
    """Jeden POST na API, vrati naparsovany JSON (odstrani BOM)."""
    data = urllib.parse.urlencode(params or {}).encode()
    req = urllib.request.Request(
        f"{BASE}/{path}", data=data, method="POST",
        headers={"Authorization": _AUTH,
                 "Content-Type": "application/x-www-form-urlencoded"},
    )
    with urllib.request.urlopen(req, context=_CTX, timeout=30) as resp:
        raw = resp.read().decode("utf-8-sig")  # utf-8-sig sezere BOM
    return json.loads(raw)


def api(method: str, params: dict | None = None, poll_timeout: int = 120) -> dict:
    """Zavola API funkci. Pokud je async (statusCode=running), poluje
    /api/get-status az do dokonceni. Vrati cely objekt odpovedi."""
    r = _post(f"invoke/{method}", params)
    if r.get("statusCode") != "running":
        return r
    token = r.get("token")
    sv = r.get("statusVersion", 0)
    t0 = time.time()
    while r.get("statusCode") == "running":
        if time.time() - t0 > poll_timeout:
            raise TimeoutError(f"{method}: polling prekrocil {poll_timeout}s")
        r = _post("get-status", {"token": token,
                                 "lastKnownStatusVersion": sv,
                                 "millisecondsTimeout": 5000})
        sv = r.get("statusVersion", sv)
    return r


def api_result(method: str, params: dict | None = None):
    r = api(method, params)
    if r.get("statusCode") != "succeeded":
        err = (r.get("error") or {}).get("message", "neznama chyba")
        raise RuntimeError(f"{method} selhalo: {err}")
    return r.get("result")


# --- formatovani ------------------------------------------------------------

def human_size(n: int) -> str:
    f = float(n)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if f < 1024 or unit == "TB":
            return f"{f:.1f} {unit}"
        f /= 1024


def print_tree(node: dict, indent: int = 0) -> int:
    """Rekurzivne vypise strom slozek. Vrati pocet vypsanych slozek."""
    count = 0
    for ch in node.get("childFolders") or []:
        marker = "+" if ch.get("hasChildFolders") else "-"
        print(f"   {'  ' * indent}{marker} {ch.get('name')}")
        count += 1
        count += print_tree(ch, indent + 1)
    return count


# --- akce -------------------------------------------------------------------

def list_mailboxes() -> None:
    """Vypise top-level slozky (schranky) v archivu."""
    root = api_result("GetChildFolders", {"maxLevels": 1})
    print("Dostupne schranky (top-level slozky archivu):")
    for ch in root.get("childFolders") or []:
        print(f"  - {ch.get('name')}")


def map_mailbox(mailbox: str, with_stats: bool = True) -> None:
    # 1) celkovy pocet zprav schranky (volitelne - GetFolderStatistics je ~20s)
    total = size = None
    if with_stats:
        print("Nacitam statistiky (GetFolderStatistics, muze trvat ~20s)...",
              file=sys.stderr, flush=True)
        stats = api_result("GetFolderStatistics") or []
        for s in stats:
            if s.get("folder") == mailbox:
                total, size = s.get("count"), s.get("size")
                break

    # 2) strom slozek
    tree = api_result("GetChildFolders", {"folder": mailbox, "maxLevels": 20})

    print("=" * 64)
    print(f"MAILSTORE MAPA SCHRANKY: {mailbox}")
    if total is not None:
        print(f"Celkem zprav: {total:,}   Velikost: {human_size(size)}")
    print("=" * 64)
    n = print_tree(tree)
    print("-" * 64)
    print(f"Slozek celkem: {n}")


def main() -> int:
    ap = argparse.ArgumentParser(description="MailStore mapa schranky (API)")
    ap.add_argument("mailbox", nargs="?", help="Nazev schranky (top-level slozka)")
    ap.add_argument("--list", action="store_true",
                    help="Vypsat dostupne schranky a skoncit")
    ap.add_argument("--no-stats", action="store_true",
                    help="Preskocit celkovy pocet zprav (rychlejsi, bez ~20s GetFolderStatistics)")
    args = ap.parse_args()

    if args.list:
        list_mailboxes()
        return 0
    if not args.mailbox:
        ap.error("zadej nazev schranky, nebo --list pro seznam")
    map_mailbox(args.mailbox, with_stats=not args.no_stats)
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nPreruseno", file=sys.stderr)
        sys.exit(1)