Clear bad IMAP Junk Flags from a folder (to quiet mbsync). #py #isync #mbsync #imap

clear_imap_junk_flags.py

#!/usr/bin/env python3
r"""Find and clear \Junk / $Junk / Junk flags from iCloud INBOX.

Reads the app-specific password from ~/.authinfo.gpg via gpg.
"""

import argparse
import imaplib
import os
import subprocess
import re
import sys

DEFAULT_IMAP_HOST = "imap.mail.me.com"
DEFAULT_IMAP_PORT = 993

JUNK_FLAGS = [r"\Junk", "$Junk", "Junk"]


def get_password(host, user):
    result = subprocess.run(
        ["gpg", "--quiet", "--for-your-eyes-only", "--no-tty", "--decrypt",
         os.path.expanduser("~/.authinfo.gpg")],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        sys.exit(f"gpg failed: {result.stderr.strip()}")
    for line in result.stdout.splitlines():
        if f"{host} login {user}" in line:
            return line.strip().split()[-1]
    sys.exit(f"No credentials found for {user}@{host}")


def find_junk_messages(conn):
    """Return sequence numbers of all messages carrying any junk flag."""
    typ, data = conn.fetch("1:*", "(FLAGS)")
    if typ != "OK":
        sys.exit("Failed to fetch flags")

    junk_msgs = []
    for item in data:
        if not isinstance(item, bytes):
            continue
        line = item.decode("utf-8", errors="replace")
        if re.search(r"junk", line, re.IGNORECASE):
            match = re.match(r"\* (\d+) FETCH", line)
            if match:
                junk_msgs.append(match.group(1))
    return junk_msgs


def fetch_headers(conn, seq):
    """Fetch From/Subject/Date for a message."""
    typ, data = conn.fetch(seq, "(BODY[HEADER.FIELDS (FROM SUBJECT DATE)])")
    if typ != "OK":
        return "(could not fetch headers)"
    for part in data:
        if isinstance(part, tuple):
            return part[1].decode("utf-8", errors="replace").strip()
    return "(no headers)"


def clear_junk_flags(conn, seq):
    """Remove all junk flag variants from a message."""
    flag_str = " ".join(JUNK_FLAGS)
    typ, data = conn.store(seq, "-FLAGS", f"({flag_str})")
    return typ == "OK"


def parse_args():
    parser = argparse.ArgumentParser(
        description=("Find and clear \\Junk / $Junk / Junk flags "
                     "from an IMAP INBOX.")
    )
    parser.add_argument("user", help="IMAP username (login)")
    parser.add_argument("--host", default=DEFAULT_IMAP_HOST,
                        help=f"IMAP host (default: {DEFAULT_IMAP_HOST})")
    parser.add_argument("--port", type=int, default=DEFAULT_IMAP_PORT,
                        help=f"IMAP SSL port (default: {DEFAULT_IMAP_PORT})")
    parser.add_argument("--folder", default="INBOX",
                        help="Mailbox/folder to scan (default: INBOX)")
    return parser.parse_args()


def main():
    args = parse_args()
    password = get_password(args.host, args.user)

    print(f"Connecting to {args.host}...")
    conn = imaplib.IMAP4_SSL(args.host, args.port)
    conn.login(args.user, password)
    typ, _ = conn.select(args.folder)
    if typ != "OK":
        sys.exit(f"Failed to select folder: {args.folder}")

    print("Scanning for junk-flagged messages...")
    junk_msgs = find_junk_messages(conn)

    if not junk_msgs:
        print("No junk-flagged messages found.")
        conn.logout()
        return

    print(f"Found {len(junk_msgs)} message(s) with junk flags:\n")

    for seq in junk_msgs:
        headers = fetch_headers(conn, seq)
        print(f"  [{seq}] {headers}")

    print()
    answer = input(
        "Clear junk flags from these messages? [y/N] ").strip().lower()
    if answer != "y":
        print("Aborted.")
        conn.logout()
        return

    for seq in junk_msgs:
        if clear_junk_flags(conn, seq):
            print(f"  Cleared flags on message {seq}")
        else:
            print(f"  FAILED to clear flags on message {seq}")

    print("\nDone.")
    conn.logout()


if __name__ == "__main__":
    main()