wienen
5c0e345fd8
Prompts used since the previous commit: - Hieronder staat je oorspronkelijke plan. Volgens mij moet stap 'Add VPS automation' nog worden uitgevoerd, inclusief de testen. Special observations: - Added a testable IMAP sorting daemon, shared IMAP operation helpers, systemd service template, and SSH deployment script. - Local tests passed; no live VPS deployment or mailbox-affecting daemon test was run. - verplaats_log.json remains modified from the user's run and was intentionally not staged.
247 lines
8.5 KiB
Python
Executable File
247 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""IMAP IDLE daemon for routing new mail through mailcat rules."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import imaplib
|
|
import json
|
|
import logging
|
|
import select
|
|
import signal
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import config
|
|
from imap_utils import quote_mailbox
|
|
from mail_imap_ops import HEADER_FETCH, destination_from_header, ensure_mailbox, fetch_body, move_message
|
|
|
|
DEFAULT_STATE_FILE = Path(__file__).parent / "sort_mail_daemon_state.json"
|
|
DEFAULT_LOG_FILE = Path(__file__).parent / "sort_mail_daemon.log"
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Sort new IMAP mail using mailcat routes.")
|
|
parser.add_argument("--account", default="backup", help="config.json account name")
|
|
parser.add_argument("--folder", default="INBOX", help="folder to watch")
|
|
parser.add_argument("--state-file", type=Path, default=DEFAULT_STATE_FILE)
|
|
parser.add_argument("--log-file", type=Path, default=DEFAULT_LOG_FILE)
|
|
parser.add_argument("--idle-timeout", type=int, default=1740)
|
|
parser.add_argument("--reconnect-delay", type=int, default=30)
|
|
parser.add_argument("--once", action="store_true", help="process once and exit")
|
|
parser.add_argument("--dry-run", action="store_true", help="log intended moves without changing IMAP")
|
|
return parser
|
|
|
|
|
|
def load_state(path: Path) -> dict:
|
|
if not path.exists():
|
|
return {"folder_state": {}}
|
|
try:
|
|
state = json.loads(path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
return {"folder_state": {}}
|
|
state.setdefault("folder_state", {})
|
|
return state
|
|
|
|
|
|
def save_state(path: Path, state: dict):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def folder_state(state: dict, folder: str) -> dict:
|
|
folders = state.setdefault("folder_state", {})
|
|
entry = folders.setdefault(folder, {})
|
|
entry.setdefault("uidvalidity", None)
|
|
entry.setdefault("processed_uids", [])
|
|
return entry
|
|
|
|
|
|
def reset_on_uidvalidity_change(state: dict, folder: str, uidvalidity: str | None):
|
|
entry = folder_state(state, folder)
|
|
if entry.get("uidvalidity") != uidvalidity:
|
|
entry["uidvalidity"] = uidvalidity
|
|
entry["processed_uids"] = []
|
|
|
|
|
|
def processed_set(state: dict, folder: str) -> set[str]:
|
|
return set(folder_state(state, folder).get("processed_uids", []))
|
|
|
|
|
|
def mark_processed(state: dict, folder: str, uid: str):
|
|
entry = folder_state(state, folder)
|
|
processed = set(entry.get("processed_uids", []))
|
|
processed.add(uid)
|
|
entry["processed_uids"] = sorted(processed, key=lambda value: int(value) if value.isdigit() else value)
|
|
|
|
|
|
def setup_logging(log_file: Path):
|
|
log_file.parent.mkdir(parents=True, exist_ok=True)
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
handlers=[logging.FileHandler(log_file, encoding="utf-8"), logging.StreamHandler(sys.stdout)],
|
|
)
|
|
|
|
|
|
def connect(account_name: str):
|
|
cfg = config.load(account_name)
|
|
mail = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
|
|
mail.login(cfg.username, cfg.password)
|
|
logging.info("connected account=%s username=%s host=%s", cfg.name, cfg.username, cfg.imap_host)
|
|
return mail
|
|
|
|
|
|
def select_folder(mail, folder: str) -> str | None:
|
|
status, data = mail.select(quote_mailbox(folder))
|
|
if status != "OK":
|
|
raise RuntimeError(f"Cannot select {folder}: {data!r}")
|
|
_status, data = mail.response("UIDVALIDITY")
|
|
if data and data[0]:
|
|
return data[0].decode(errors="replace") if isinstance(data[0], bytes) else str(data[0])
|
|
return None
|
|
|
|
|
|
def uid_search_all(mail) -> list[bytes]:
|
|
status, data = mail.uid("SEARCH", None, "ALL")
|
|
if status != "OK" or not data or not data[0]:
|
|
return []
|
|
return [uid for uid in data[0].split() if uid]
|
|
|
|
|
|
def process_folder(mail, state: dict, folder: str, dry_run: bool) -> dict[str, int]:
|
|
uidvalidity = select_folder(mail, folder)
|
|
reset_on_uidvalidity_change(state, folder, uidvalidity)
|
|
processed = processed_set(state, folder)
|
|
counts = {"seen": 0, "matched": 0, "moved": 0, "unmatched": 0, "failed": 0, "skipped": 0}
|
|
|
|
for uid in uid_search_all(mail):
|
|
uid_text = uid.decode(errors="replace")
|
|
counts["seen"] += 1
|
|
if uid_text in processed:
|
|
counts["skipped"] += 1
|
|
continue
|
|
|
|
status, data = mail.uid("FETCH", uid, HEADER_FETCH)
|
|
raw_header = fetch_body(data)
|
|
if status != "OK" or raw_header is None:
|
|
counts["failed"] += 1
|
|
logging.warning("fetch_failed folder=%s uid=%s status=%s data=%r", folder, uid_text, status, data)
|
|
continue
|
|
|
|
destination = destination_from_header(raw_header)
|
|
if not destination or destination == folder:
|
|
counts["unmatched"] += 1
|
|
mark_processed(state, folder, uid_text)
|
|
logging.info("no_match folder=%s uid=%s", folder, uid_text)
|
|
continue
|
|
|
|
counts["matched"] += 1
|
|
if dry_run:
|
|
logging.info("dry_move folder=%s uid=%s destination=%s", folder, uid_text, destination)
|
|
mark_processed(state, folder, uid_text)
|
|
continue
|
|
|
|
if not ensure_mailbox(mail, destination):
|
|
counts["failed"] += 1
|
|
logging.error("ensure_destination_failed folder=%s uid=%s destination=%s", folder, uid_text, destination)
|
|
continue
|
|
|
|
ok, failure = move_message(mail, uid, folder, destination)
|
|
if ok:
|
|
counts["moved"] += 1
|
|
mark_processed(state, folder, uid_text)
|
|
logging.info("moved folder=%s uid=%s destination=%s", folder, uid_text, destination)
|
|
else:
|
|
counts["failed"] += 1
|
|
logging.error("move_failed folder=%s uid=%s destination=%s failure=%r", folder, uid_text, destination, failure)
|
|
|
|
if not dry_run:
|
|
mail.expunge()
|
|
return counts
|
|
|
|
|
|
def wait_for_mail(mail, timeout: int) -> bool:
|
|
if hasattr(mail, "idle"):
|
|
with mail.idle(duration=timeout) as idler:
|
|
for typ, _data in idler:
|
|
if typ == "EXISTS":
|
|
return True
|
|
return False
|
|
return wait_for_mail_private_idle(mail, timeout)
|
|
|
|
|
|
def wait_for_mail_private_idle(mail, timeout: int) -> bool:
|
|
tag = mail._new_tag()
|
|
mail.send(tag + b" IDLE\r\n")
|
|
continuation = mail.readline()
|
|
if not continuation.startswith(b"+"):
|
|
raise RuntimeError(f"IDLE rejected: {continuation!r}")
|
|
|
|
changed = False
|
|
try:
|
|
ready, _, _ = select.select([mail.sock], [], [], timeout)
|
|
if ready:
|
|
line = mail.readline()
|
|
changed = b"EXISTS" in line or b"RECENT" in line
|
|
finally:
|
|
mail.send(b"DONE\r\n")
|
|
while True:
|
|
line = mail.readline()
|
|
if line.startswith(tag):
|
|
break
|
|
return changed
|
|
|
|
|
|
def run(args) -> int:
|
|
setup_logging(args.log_file)
|
|
stop = {"requested": False}
|
|
|
|
def request_stop(_signum, _frame):
|
|
stop["requested"] = True
|
|
|
|
signal.signal(signal.SIGTERM, request_stop)
|
|
signal.signal(signal.SIGINT, request_stop)
|
|
|
|
state = load_state(args.state_file)
|
|
|
|
while not stop["requested"]:
|
|
mail = None
|
|
try:
|
|
mail = connect(args.account)
|
|
counts = process_folder(mail, state, args.folder, args.dry_run)
|
|
save_state(args.state_file, state)
|
|
logging.info("scan_complete folder=%s counts=%s", args.folder, counts)
|
|
if args.once:
|
|
return 0
|
|
|
|
while not stop["requested"]:
|
|
select_folder(mail, args.folder)
|
|
changed = wait_for_mail(mail, args.idle_timeout)
|
|
if changed:
|
|
counts = process_folder(mail, state, args.folder, args.dry_run)
|
|
save_state(args.state_file, state)
|
|
logging.info("idle_scan_complete folder=%s counts=%s", args.folder, counts)
|
|
else:
|
|
logging.info("idle_timeout folder=%s", args.folder)
|
|
except Exception:
|
|
logging.exception("daemon_error reconnecting_after=%ss", args.reconnect_delay)
|
|
if args.once:
|
|
return 1
|
|
time.sleep(args.reconnect_delay)
|
|
finally:
|
|
if mail is not None:
|
|
try:
|
|
mail.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
save_state(args.state_file, state)
|
|
logging.info("stopped")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(run(build_parser().parse_args()))
|