notebook
This commit is contained in:
@@ -0,0 +1,176 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user