codex-oauth-automation-extension/scripts/hotmail_helper.py

import email
import html
import imaplib
import json
import os
import re
import threading
import time
import traceback
from datetime import datetime, timezone
from email.header import decode_header
from email.utils import getaddresses, parseaddr, parsedate_to_datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen


HOST = os.environ.get("HOTMAIL_HELPER_HOST", "127.0.0.1").strip() or "127.0.0.1"
try:
    PORT = int(os.environ.get("HOTMAIL_HELPER_PORT", "17373") or 17373)
except Exception:
    PORT = 17373
LIVE_TOKEN_URL = "https://login.live.com/oauth20_token.srf"
ENTRA_COMMON_TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
ENTRA_CONSUMERS_TOKEN_URL = "https://login.microsoftonline.com/consumers/oauth2/v2.0/token"
GRAPH_API_ORIGIN = "https://graph.microsoft.com"
OUTLOOK_API_ORIGIN = "https://outlook.office.com"
GRAPH_SCOPES = "offline_access https://graph.microsoft.com/Mail.Read https://graph.microsoft.com/User.Read"
GRAPH_DEFAULT_SCOPE = "https://graph.microsoft.com/.default"
TOKEN_ENDPOINTS = {
    "live": {
        "name": "live",
        "url": LIVE_TOKEN_URL,
        "extra_data": {},
    },
    "entra-consumers-delegated": {
        "name": "entra-consumers-delegated",
        "url": ENTRA_CONSUMERS_TOKEN_URL,
        "extra_data": {
            "scope": GRAPH_SCOPES,
        },
    },
    "entra-common-delegated": {
        "name": "entra-common-delegated",
        "url": ENTRA_COMMON_TOKEN_URL,
        "extra_data": {
            "scope": GRAPH_SCOPES,
        },
    },
    "entra-common-default": {
        "name": "entra-common-default",
        "url": ENTRA_COMMON_TOKEN_URL,
        "extra_data": {
            "scope": GRAPH_DEFAULT_SCOPE,
        },
    },
    "entra-common-outlook": {
        "name": "entra-common-outlook",
        "url": ENTRA_COMMON_TOKEN_URL,
        "extra_data": {},
    },
}
IMAP_HOST = "outlook.office365.com"
IMAP_PORT = 993
REQUEST_TIMEOUT_SECONDS = 45
FETCH_LIMIT_DEFAULT = 5
FETCH_LIMIT_MAX = 120
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
ACCOUNT_LOG_PATH = os.path.join(BASE_DIR, "data", "account-run-history.txt")
ACCOUNT_RECORDS_SNAPSHOT_PATH = os.path.join(BASE_DIR, "data", "account-run-history.json")
A4SKY_IMAP_CONFIG_PATH = os.path.join(BASE_DIR, "data", "a4sky-imap.local.json")
ACCOUNT_RECORDS_LOCK = threading.Lock()


def json_response(handler, status, payload):
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json; charset=utf-8")
    handler.send_header("Content-Length", str(len(body)))
    handler.send_header("Access-Control-Allow-Origin", "*")
    handler.send_header("Access-Control-Allow-Headers", "Content-Type")
    handler.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
    handler.end_headers()
    handler.wfile.write(body)


def read_json_payload(handler):
    length = int(handler.headers.get("Content-Length", "0") or 0)
    raw = handler.rfile.read(length) if length > 0 else b"{}"
    try:
        return json.loads(raw.decode("utf-8"))
    except Exception as exc:
        raise RuntimeError(f"Invalid JSON payload: {exc}") from exc


def post_form(url, data):
    encoded = urlencode(data).encode("utf-8")
    request = Request(url, data=encoded, headers={"Content-Type": "application/x-www-form-urlencoded"})
    with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
        return json.loads(response.read().decode("utf-8"))


def get_json(url, headers=None):
    request = Request(url, headers=headers or {})
    with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
        return response.getcode(), json.loads(response.read().decode("utf-8"))


def mask_secret(value, keep=6):
    raw = str(value or "")
    if not raw:
        return ""
    if len(raw) <= keep:
        return "*" * len(raw)
    return raw[:keep] + "..." + raw[-keep:]


def compact_text(value, limit=400):
    text = str(value or "").replace("\r", " ").replace("\n", " ").strip()
    return text[:limit]


def log_info(message):
    print(f"[HotmailHelper] {message}", flush=True)


def append_account_log(email_addr, password, status, recorded_at="", reason=""):
    normalized_email = str(email_addr or "").strip()
    normalized_password = str(password or "").strip()
    normalized_status = str(status or "").strip().lower()
    normalized_recorded_at = str(recorded_at or "").strip() or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    normalized_reason = str(reason or "").strip().replace("\r", " ").replace("\n", " ")

    if not normalized_email or not normalized_password or not normalized_status:
        raise RuntimeError("Missing email/password/status for account log append")

    os.makedirs(os.path.dirname(ACCOUNT_LOG_PATH), exist_ok=True)
    line = f"{normalized_recorded_at}\t{normalized_email}\t{normalized_password}\t{normalized_status}\t{normalized_reason}\n"
    with ACCOUNT_RECORDS_LOCK:
        with open(ACCOUNT_LOG_PATH, "a", encoding="utf-8") as handle:
            handle.write(line)
    return ACCOUNT_LOG_PATH


def normalize_account_run_snapshot_record(record):
    if not isinstance(record, dict):
        return None

    email_addr = str(record.get("email") or "").strip()
    password = str(record.get("password") or "").strip()
    final_status = str(record.get("finalStatus") or "").strip().lower()
    if not email_addr or not password or final_status not in {"success", "failed", "stopped"}:
        return None

    finished_at = str(record.get("finishedAt") or "").strip() or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    retry_count = max(0, int(record.get("retryCount") or 0))
    failed_step_raw = record.get("failedStep")
    try:
        failed_step = int(failed_step_raw)
    except (TypeError, ValueError):
        failed_step = None
    if failed_step is not None and failed_step <= 0:
        failed_step = None

    auto_run_context = record.get("autoRunContext") if isinstance(record.get("autoRunContext"), dict) else None
    normalized_auto_run_context = None
    if auto_run_context:
        normalized_auto_run_context = {
            "currentRun": max(0, int(auto_run_context.get("currentRun") or 0)),
            "totalRuns": max(0, int(auto_run_context.get("totalRuns") or 0)),
            "attemptRun": max(0, int(auto_run_context.get("attemptRun") or 0)),
        }
        if not any(normalized_auto_run_context.values()):
            normalized_auto_run_context = None

    source = "auto" if str(record.get("source") or "").strip().lower() == "auto" else "manual"

    return {
        "recordId": str(record.get("recordId") or email_addr).strip() or email_addr,
        "email": email_addr,
        "password": password,
        "finalStatus": final_status,
        "finishedAt": finished_at,
        "retryCount": retry_count,
        "failureLabel": str(record.get("failureLabel") or "").strip(),
        "failureDetail": str(record.get("failureDetail") or "").strip(),
        "failedStep": failed_step,
        "source": source,
        "autoRunContext": normalized_auto_run_context,
    }


def summarize_account_run_snapshot(records):
    summary = {
        "total": 0,
        "success": 0,
        "failed": 0,
        "retryTotal": 0,
    }
    for item in records:
        summary["total"] += 1
        if item.get("finalStatus") == "success":
            summary["success"] += 1
        elif item.get("finalStatus") == "failed":
            summary["failed"] += 1
        summary["retryTotal"] += max(0, int(item.get("retryCount") or 0))
    return summary


def normalize_account_run_snapshot_payload(payload):
    if not isinstance(payload, dict):
        raise RuntimeError("Invalid account run snapshot payload")

    normalized_records = []
    for item in payload.get("records") if isinstance(payload.get("records"), list) else []:
        normalized = normalize_account_run_snapshot_record(item)
        if normalized:
            normalized_records.append(normalized)

    return {
        "generatedAt": str(payload.get("generatedAt") or "").strip() or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        "summary": summarize_account_run_snapshot(normalized_records),
        "records": normalized_records,
    }


def sync_account_run_records(payload):
    normalized_payload = normalize_account_run_snapshot_payload(payload)
    os.makedirs(os.path.dirname(ACCOUNT_RECORDS_SNAPSHOT_PATH), exist_ok=True)
    with ACCOUNT_RECORDS_LOCK:
        with open(ACCOUNT_RECORDS_SNAPSHOT_PATH, "w", encoding="utf-8") as handle:
            json.dump(normalized_payload, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
    return ACCOUNT_RECORDS_SNAPSHOT_PATH


def try_refresh_access_token(endpoint, client_id, refresh_token):
    request_data = {
        "client_id": client_id,
        "refresh_token": refresh_token,
        "grant_type": "refresh_token",
        **(endpoint.get("extra_data") or {}),
    }
    started_at = time.monotonic()
    try:
        payload = post_form(endpoint["url"], request_data)
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="ignore")
        return {
            "ok": False,
            "endpoint": endpoint["name"],
            "url": endpoint["url"],
            "status": getattr(exc, "code", None),
            "error": compact_text(detail or str(exc)),
            "elapsed_ms": int((time.monotonic() - started_at) * 1000),
        }
    except URLError as exc:
        return {
            "ok": False,
            "endpoint": endpoint["name"],
            "url": endpoint["url"],
            "status": None,
            "error": compact_text(f"Token request failed: {exc}"),
            "elapsed_ms": int((time.monotonic() - started_at) * 1000),
        }

    access_token = str(payload.get("access_token") or "").strip()
    if not access_token:
        return {
            "ok": False,
            "endpoint": endpoint["name"],
            "url": endpoint["url"],
            "status": 200,
            "error": compact_text(payload.get("error_description") or payload.get("error") or json.dumps(payload, ensure_ascii=False)),
            "elapsed_ms": int((time.monotonic() - started_at) * 1000),
        }

    return {
        "ok": True,
        "endpoint": endpoint["name"],
        "url": endpoint["url"],
        "elapsed_ms": int((time.monotonic() - started_at) * 1000),
        "payload": {
            "access_token": access_token,
            "next_refresh_token": str(payload.get("refresh_token") or "").strip(),
        },
    }


def refresh_access_token(client_id, refresh_token, strategy_names=None):
    errors = []
    selected_endpoints = [
        TOKEN_ENDPOINTS[name]
        for name in (strategy_names or ["live", "entra-consumers-delegated", "entra-common-delegated"])
        if name in TOKEN_ENDPOINTS
    ]
    log_info(
        "token refresh start "
        f"clientId={mask_secret(client_id)} "
        f"refreshToken={mask_secret(refresh_token)} "
        f"strategies={[item['name'] for item in selected_endpoints]}"
    )

    for endpoint in selected_endpoints:
        result = try_refresh_access_token(endpoint, client_id, refresh_token)
        if result["ok"]:
            log_info(
                "token refresh success "
                f"endpoint={result['endpoint']} "
                f"elapsedMs={result['elapsed_ms']}"
            )
            return {
                "access_token": result["payload"]["access_token"],
                "next_refresh_token": result["payload"]["next_refresh_token"],
                "token_endpoint": result["endpoint"],
                "token_url": result["url"],
            }

        errors.append(result)
        log_info(
            "token refresh failed "
            f"endpoint={result['endpoint']} "
            f"status={result['status']} "
            f"elapsedMs={result['elapsed_ms']} "
            f"detail={result['error']}"
        )

    details = " | ".join(
        f"{item['endpoint']}({item['status']}): {item['error']}"
        for item in errors
    )
    raise RuntimeError(f"Token refresh failed on all endpoints: {details}")


def load_local_imap_config():
    if not os.path.exists(A4SKY_IMAP_CONFIG_PATH):
        return {}
    try:
        with open(A4SKY_IMAP_CONFIG_PATH, "r", encoding="utf-8") as handle:
            payload = json.load(handle)
            return payload if isinstance(payload, dict) else {}
    except Exception as exc:
        raise RuntimeError(f"Invalid local IMAP config: {exc}") from exc


def resolve_basic_imap_settings(payload):
    local_config = load_local_imap_config()
    host = str(payload.get("host") or local_config.get("host") or "").strip()
    username = str(payload.get("username") or local_config.get("username") or "").strip()
    password = str(payload.get("password") or local_config.get("password") or "").strip()
    port_raw = payload.get("port") if payload.get("port") is not None else local_config.get("port")
    try:
        port = int(port_raw or 993)
    except Exception as exc:
        raise RuntimeError(f"Invalid IMAP port: {exc}") from exc

    if not host or not username or not password:
        raise RuntimeError("Missing IMAP host/username/password. Please fill data/a4sky-imap.local.json or pass credentials explicitly.")

    return {
        "host": host,
        "port": max(1, port),
        "username": username,
        "password": password,
    }


def open_basic_imap_mailbox(host, port, username, password):
    client = imaplib.IMAP4_SSL(host, port)
    client.login(username, password)
    return client


def build_xoauth2(email_addr, access_token):
    return f"user={email_addr}\x01auth=Bearer {access_token}\x01\x01".encode("utf-8")


def open_mailbox(email_addr, access_token):
    client = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
    client.authenticate("XOAUTH2", lambda _: build_xoauth2(email_addr, access_token))
    return client


def decode_mime_header(value):
    if not value:
        return ""
    parts = []
    for chunk, charset in decode_header(value):
        if isinstance(chunk, bytes):
            parts.append(chunk.decode(charset or "utf-8", errors="ignore"))
        else:
            parts.append(str(chunk))
    return "".join(parts).strip()


def extract_text_part(message):
    if message.is_multipart():
        for part in message.walk():
            if part.get_content_maintype() == "multipart":
                continue
            if "attachment" in str(part.get("Content-Disposition") or "").lower():
                continue
            payload = part.get_payload(decode=True) or b""
            charset = part.get_content_charset() or "utf-8"
            text = payload.decode(charset, errors="ignore").strip()
            if part.get_content_type() == "text/plain" and text:
                return text
            if part.get_content_type() == "text/html" and text:
                return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", html.unescape(text))).strip()
        return ""

    payload = message.get_payload(decode=True) or b""
    charset = message.get_content_charset() or "utf-8"
    text = payload.decode(charset, errors="ignore").strip()
    if message.get_content_type() == "text/html":
        return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", html.unescape(text))).strip()
    return text


def mailbox_candidates(mailbox):
    normalized = str(mailbox or "INBOX").strip().lower()
    if normalized in {"junk", "junk email", "junk e-mail", "junkemail"}:
        return ["Junk", "Junk Email", "Junk E-Mail"]
    return ["INBOX"]


def normalize_mailbox_label(mailbox):
    normalized = str(mailbox or "INBOX").strip().lower()
    if normalized in {"junk", "junk email", "junk e-mail", "junkemail"}:
        return "Junk"
    return "INBOX"


def normalize_mailbox_id(mailbox):
    normalized = str(mailbox or "INBOX").strip().lower()
    if normalized in {"junk", "junk email", "junk e-mail", "junkemail"}:
        return "junkemail"
    return "inbox"


def select_mailbox(client, mailbox):
    for candidate in mailbox_candidates(mailbox):
        status, _ = client.select(candidate)
        if status == "OK":
            return candidate
    raise RuntimeError(f"Mailbox not found: {mailbox}")


def to_timestamp_ms(raw_date):
    if not raw_date:
        return 0
    try:
        parsed = parsedate_to_datetime(raw_date)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return int(parsed.timestamp() * 1000)
    except Exception:
        return 0


def to_iso_string(timestamp_ms):
    if not timestamp_ms:
        return ""
    return datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc).isoformat().replace("+00:00", "Z")


def normalize_message(message_id, raw_bytes, mailbox):
    parsed = email.message_from_bytes(raw_bytes)
    sender_name, sender_addr = parseaddr(parsed.get("From", ""))
    subject = decode_mime_header(parsed.get("Subject", ""))
    body = extract_text_part(parsed)
    timestamp_ms = to_timestamp_ms(parsed.get("Date"))

    recipient_headers = []
    for header_name in ["To", "Delivered-To", "Envelope-To", "X-Original-To", "Cc"]:
        recipient_headers.extend(parsed.get_all(header_name, []))

    recipient_items = []
    recipient_addresses = []
    for recipient_name, recipient_addr in getaddresses(recipient_headers):
        normalized_addr = str(recipient_addr or "").strip().lower()
        if not normalized_addr:
            continue
        recipient_addresses.append(normalized_addr)
        recipient_items.append({
            "emailAddress": {
                "address": normalized_addr,
                "name": str(recipient_name or "").strip(),
            }
        })

    return {
        "id": str(message_id),
        "mailbox": mailbox,
        "subject": subject,
        "from": {
            "emailAddress": {
                "address": sender_addr.strip(),
                "name": sender_name.strip(),
            }
        },
        "toRecipients": recipient_items,
        "recipientAddresses": recipient_addresses,
        "bodyPreview": body[:500],
        "receivedDateTime": to_iso_string(timestamp_ms),
        "receivedTimestamp": timestamp_ms,
    }


def normalize_fetch_limit(top):
    try:
        numeric = int(top or FETCH_LIMIT_DEFAULT)
    except Exception:
        numeric = FETCH_LIMIT_DEFAULT
    return max(1, min(numeric, FETCH_LIMIT_MAX))


def search_imap_message_ids(client, target_email=""):
    normalized_target = str(target_email or "").strip().lower()
    if not normalized_target:
        return []

    matched_ids = []
    seen_ids = set()
    for header_name in ["To", "Delivered-To", "Envelope-To", "X-Original-To", "Cc"]:
        try:
            status, data = client.search(None, "HEADER", header_name, f'"{normalized_target}"')
        except Exception:
            continue
        if status != "OK" or not data or not data[0]:
            continue
        for message_id in data[0].split():
            if not message_id or message_id in seen_ids:
                continue
            seen_ids.add(message_id)
            matched_ids.append(message_id)

    matched_ids.sort(key=lambda item: int(item) if item.isdigit() else 0)
    return matched_ids


def load_selected_message_ids(client, top, target_email=""):
    limit = normalize_fetch_limit(top)
    target_ids = search_imap_message_ids(client, target_email)
    if target_ids:
        return list(reversed(target_ids[-limit:]))

    status, data = client.search(None, "ALL")
    if status != "OK" or not data or not data[0]:
        return []
    message_ids = data[0].split()
    return list(reversed(message_ids[-limit:]))


def fetch_messages(email_addr, access_token, mailbox="INBOX", top=FETCH_LIMIT_DEFAULT):
    client = None
    logical_mailbox = normalize_mailbox_label(mailbox)
    try:
        client = open_mailbox(email_addr, access_token)
        select_mailbox(client, mailbox)
        selected_ids = load_selected_message_ids(client, top)
        if not selected_ids:
            return {"mailbox": logical_mailbox, "messages": [], "count": 0}

        messages = []
        for message_id in selected_ids:
            fetch_status, fetch_data = client.fetch(message_id, "(RFC822)")
            if fetch_status != "OK" or not fetch_data:
                continue
            raw_bytes = b""
            for item in fetch_data:
                if isinstance(item, tuple) and len(item) >= 2:
                    raw_bytes = item[1]
                    break
            if not raw_bytes:
                continue
            messages.append(normalize_message(message_id.decode("utf-8", errors="ignore"), raw_bytes, logical_mailbox))
        return {"mailbox": logical_mailbox, "messages": messages, "count": len(messages)}
    finally:
        if client is not None:
            try:
                client.logout()
            except Exception:
                pass


def fetch_messages_for_mailboxes(email_addr, access_token, mailboxes, top):
    mailbox_results = []
    all_messages = []
    for mailbox in mailboxes or ["INBOX"]:
        result = fetch_messages(email_addr, access_token, mailbox=mailbox, top=top)
        mailbox_results.append(result)
        all_messages.extend(result["messages"])
    all_messages.sort(key=lambda item: int(item.get("receivedTimestamp") or 0), reverse=True)
    return {"mailboxResults": mailbox_results, "messages": all_messages}


def fetch_basic_imap_messages(host, port, username, password, mailbox="INBOX", top=FETCH_LIMIT_DEFAULT, target_email=""):
    client = None
    logical_mailbox = normalize_mailbox_label(mailbox)
    try:
        client = open_basic_imap_mailbox(host, port, username, password)
        select_mailbox(client, mailbox)
        selected_ids = load_selected_message_ids(client, top, target_email)
        if not selected_ids:
            return {"mailbox": logical_mailbox, "messages": [], "count": 0}

        messages = []
        for message_id in selected_ids:
            fetch_status, fetch_data = client.fetch(message_id, "(RFC822)")
            if fetch_status != "OK" or not fetch_data:
                continue
            raw_bytes = b""
            for item in fetch_data:
                if isinstance(item, tuple) and len(item) >= 2:
                    raw_bytes = item[1]
                    break
            if not raw_bytes:
                continue
            messages.append(normalize_message(message_id.decode("utf-8", errors="ignore"), raw_bytes, logical_mailbox))
        return {"mailbox": logical_mailbox, "messages": messages, "count": len(messages)}
    finally:
        if client is not None:
            try:
                client.logout()
            except Exception:
                pass


def fetch_basic_imap_messages_for_mailboxes(host, port, username, password, mailboxes, top, target_email=""):
    mailbox_results = []
    all_messages = []
    for mailbox in mailboxes or ["INBOX"]:
        result = fetch_basic_imap_messages(
            host,
            port,
            username,
            password,
            mailbox=mailbox,
            top=top,
            target_email=target_email,
        )
        mailbox_results.append(result)
        all_messages.extend(result["messages"])
    all_messages.sort(key=lambda item: int(item.get("receivedTimestamp") or 0), reverse=True)
    return {"mailboxResults": mailbox_results, "messages": all_messages}


def collect_basic_imap_messages(payload, mailboxes, top):
    settings = resolve_basic_imap_settings(payload)
    target_email = str(payload.get("targetEmail") or payload.get("email") or "").strip().lower()
    result = fetch_basic_imap_messages_for_mailboxes(
        settings["host"],
        settings["port"],
        settings["username"],
        settings["password"],
        mailboxes,
        top,
        target_email=target_email,
    )
    result["transport"] = "imap-basic"
    result["settings"] = {
        "host": settings["host"],
        "port": settings["port"],
        "username": settings["username"],
    }
    return result


def normalize_graph_message(message, mailbox):
    sender = message.get("from", {}) or {}
    email_addr = sender.get("emailAddress", {}) if isinstance(sender, dict) else {}
    received = str(message.get("receivedDateTime") or "").strip()
    return {
        "id": str(message.get("id") or message.get("internetMessageId") or "").strip(),
        "mailbox": mailbox,
        "subject": str(message.get("subject") or "").strip(),
        "from": {
            "emailAddress": {
                "address": str(email_addr.get("address") or "").strip(),
                "name": str(email_addr.get("name") or "").strip(),
            }
        },
        "bodyPreview": str(message.get("bodyPreview") or "").strip(),
        "receivedDateTime": received,
        "receivedTimestamp": int(datetime.fromisoformat(received.replace("Z", "+00:00")).timestamp() * 1000) if received else 0,
    }


def normalize_outlook_message(message, mailbox):
    sender = message.get("From", {}) or message.get("from", {}) or {}
    email_addr = sender.get("EmailAddress", {}) if isinstance(sender, dict) else {}
    if isinstance(sender, dict) and not email_addr:
        email_addr = sender.get("emailAddress", {}) if isinstance(sender, dict) else {}
    received = str(message.get("ReceivedDateTime") or message.get("receivedDateTime") or "").strip()
    return {
        "id": str(message.get("Id") or message.get("id") or "").strip(),
        "mailbox": mailbox,
        "subject": str(message.get("Subject") or message.get("subject") or "").strip(),
        "from": {
            "emailAddress": {
                "address": str(email_addr.get("Address") or email_addr.get("address") or "").strip(),
                "name": str(email_addr.get("Name") or email_addr.get("name") or "").strip(),
            }
        },
        "bodyPreview": str(message.get("BodyPreview") or message.get("bodyPreview") or "").strip(),
        "receivedDateTime": received,
        "receivedTimestamp": int(datetime.fromisoformat(received.replace("Z", "+00:00")).timestamp() * 1000) if received else 0,
    }


def fetch_graph_messages(access_token, mailbox="INBOX", top=FETCH_LIMIT_DEFAULT):
    mailbox_id = normalize_mailbox_id(mailbox)
    url = (
        f"{GRAPH_API_ORIGIN}/v1.0/me/mailFolders/{mailbox_id}/messages"
        f"?$top={max(1, min(int(top or FETCH_LIMIT_DEFAULT), 30))}"
        f"&$select=id,internetMessageId,subject,from,bodyPreview,receivedDateTime"
        f"&$orderby=receivedDateTime desc"
    )
    try:
        _, payload = get_json(url, headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {access_token}",
        })
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="ignore")
        raise RuntimeError(f"Graph request failed: {detail or exc}") from exc
    except URLError as exc:
        raise RuntimeError(f"Graph request failed: {exc}") from exc

    messages = [normalize_graph_message(item, normalize_mailbox_label(mailbox)) for item in (payload.get("value") or [])]
    return {"mailbox": normalize_mailbox_label(mailbox), "messages": messages, "count": len(messages)}


def fetch_outlook_api_messages(access_token, mailbox="INBOX", top=FETCH_LIMIT_DEFAULT):
    mailbox_id = normalize_mailbox_id(mailbox)
    url = (
        f"{OUTLOOK_API_ORIGIN}/api/v2.0/me/mailfolders/{mailbox_id}/messages"
        f"?$top={max(1, min(int(top or FETCH_LIMIT_DEFAULT), 30))}"
        f"&$select=Id,Subject,From,BodyPreview,ReceivedDateTime"
        f"&$orderby=ReceivedDateTime desc"
    )
    try:
        _, payload = get_json(url, headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {access_token}",
        })
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="ignore")
        raise RuntimeError(f"Outlook API request failed: {detail or exc}") from exc
    except URLError as exc:
        raise RuntimeError(f"Outlook API request failed: {exc}") from exc

    messages = [normalize_outlook_message(item, normalize_mailbox_label(mailbox)) for item in (payload.get("value") or [])]
    return {"mailbox": normalize_mailbox_label(mailbox), "messages": messages, "count": len(messages)}


def collect_imap_messages(email_addr, client_id, refresh_token, mailboxes, top):
    token_payload = refresh_access_token(client_id, refresh_token, [
        "live",
        "entra-consumers-delegated",
        "entra-common-delegated",
    ])
    result = fetch_messages_for_mailboxes(email_addr, token_payload["access_token"], mailboxes, top)
    result["transport"] = "imap"
    result["token_payload"] = token_payload
    return result


def collect_graph_messages(email_addr, client_id, refresh_token, mailboxes, top):
    token_payload = refresh_access_token(client_id, refresh_token, [
        "entra-common-delegated",
        "entra-consumers-delegated",
        "entra-common-default",
    ])
    mailbox_results = [fetch_graph_messages(token_payload["access_token"], mailbox=mailbox, top=top) for mailbox in mailboxes]
    messages = []
    for item in mailbox_results:
        messages.extend(item["messages"])
    messages.sort(key=lambda item: int(item.get("receivedTimestamp") or 0), reverse=True)
    return {
        "transport": "graph",
        "token_payload": token_payload,
        "mailboxResults": mailbox_results,
        "messages": messages,
    }


def collect_outlook_messages(email_addr, client_id, refresh_token, mailboxes, top):
    token_payload = refresh_access_token(client_id, refresh_token, [
        "entra-common-outlook",
        "entra-common-delegated",
    ])
    mailbox_results = [fetch_outlook_api_messages(token_payload["access_token"], mailbox=mailbox, top=top) for mailbox in mailboxes]
    messages = []
    for item in mailbox_results:
        messages.extend(item["messages"])
    messages.sort(key=lambda item: int(item.get("receivedTimestamp") or 0), reverse=True)
    return {
        "transport": "outlook",
        "token_payload": token_payload,
        "mailboxResults": mailbox_results,
        "messages": messages,
    }


def collect_messages(email_addr, client_id, refresh_token, mailboxes, top):
    errors = []
    collectors = [
        ("imap", collect_imap_messages),
        ("graph", collect_graph_messages),
        ("outlook", collect_outlook_messages),
    ]

    for transport_name, collector in collectors:
        try:
            log_info(f"message collection start transport={transport_name}")
            result = collector(email_addr, client_id, refresh_token, mailboxes, top)
            log_info(
                f"message collection success transport={transport_name} "
                f"tokenEndpoint={result['token_payload'].get('token_endpoint', '')}"
            )
            return result
        except Exception as exc:
            message = compact_text(str(exc), 600)
            errors.append(f"{transport_name}: {message}")
            log_info(f"message collection failed transport={transport_name} detail={message}")

    raise RuntimeError(f"Message collection failed on all transports: {' | '.join(errors)}")


def extract_code(text):
    source = str(text or "")
    patterns = [
        r"(?:代码为|验证码[^0-9]*?)[\s：:]*(\d{6})",
        r"code(?:\s+is|[\s:])+(\d{6})",
        r"\b(\d{6})\b",
    ]
    for pattern in patterns:
        match = re.search(pattern, source, flags=re.IGNORECASE)
        if match:
            return match.group(1)
    return ""


def select_latest_code(messages, sender_filters, subject_filters, exclude_codes, filter_after_timestamp, target_email=""):
    sender_keywords = [str(item).strip().lower() for item in sender_filters or [] if str(item).strip()]
    subject_keywords = [str(item).strip().lower() for item in subject_filters or [] if str(item).strip()]
    excluded = {str(item).strip() for item in exclude_codes or [] if str(item).strip()}

    normalized_target_email = str(target_email or "").strip().lower()

    def match_message(message, apply_time_filter):
        timestamp = int(message.get("receivedTimestamp") or 0)
        if apply_time_filter and filter_after_timestamp and timestamp and timestamp < int(filter_after_timestamp):
            return None

        sender = str(message.get("from", {}).get("emailAddress", {}).get("address", "")).lower()
        subject = str(message.get("subject", ""))
        preview = str(message.get("bodyPreview", ""))
        recipient_addresses = [
            str(item or "").strip().lower()
            for item in message.get("recipientAddresses", [])
            if str(item or "").strip()
        ]
        recipient_text = " ".join(recipient_addresses)
        combined = " ".join([sender, subject.lower(), preview.lower(), recipient_text])
        if normalized_target_email and recipient_addresses and normalized_target_email not in recipient_addresses:
            return None

        code = extract_code(" ".join([subject, preview, sender]))
        if not code or code in excluded:
            return None

        sender_ok = not sender_keywords or any(keyword in combined for keyword in sender_keywords)
        subject_ok = not subject_keywords or any(keyword in combined for keyword in subject_keywords)
        if not sender_ok and not subject_ok:
            return None

        return {"code": code, "message": message}

    for use_time_fallback in [False, True]:
        matched = []
        for message in messages:
            result = match_message(message, apply_time_filter=not use_time_fallback)
            if result:
                matched.append(result)
        if matched:
            matched.sort(key=lambda item: int(item["message"].get("receivedTimestamp") or 0), reverse=True)
            best = matched[0]
            return {
                "code": best["code"],
                "message": best["message"],
                "usedTimeFallback": use_time_fallback,
            }
    return {"code": "", "message": None, "usedTimeFallback": False}


class HotmailHelperHandler(BaseHTTPRequestHandler):
    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
        self.end_headers()

    def do_POST(self):
        try:
            payload = read_json_payload(self)

            if self.path == "/sync-account-run-records":
                file_path = sync_account_run_records(payload)
                json_response(self, 200, {
                    "ok": True,
                    "filePath": file_path,
                })
                return

            if self.path == "/append-account-log":
                file_path = append_account_log(
                    payload.get("email"),
                    payload.get("password"),
                    payload.get("status"),
                    payload.get("recordedAt"),
                    payload.get("reason"),
                )
                json_response(self, 200, {
                    "ok": True,
                    "filePath": file_path,
                })
                return

            top = max(1, min(int(payload.get("top") or FETCH_LIMIT_DEFAULT), 30))
            mailboxes = payload.get("mailboxes") if isinstance(payload.get("mailboxes"), list) else [payload.get("mailbox") or "INBOX"]

            if self.path == "/imap-messages":
                result = collect_basic_imap_messages(payload, mailboxes, top)
                json_response(self, 200, {
                    "ok": True,
                    "messages": result["messages"],
                    "mailboxResults": result["mailboxResults"],
                    "transport": result.get("transport") or "",
                    "settings": result.get("settings") or {},
                })
                return

            if self.path == "/imap-code":
                result = collect_basic_imap_messages(payload, mailboxes, top)
                selected = select_latest_code(
                    result["messages"],
                    payload.get("senderFilters") or [],
                    payload.get("subjectFilters") or [],
                    payload.get("excludeCodes") or [],
                    int(payload.get("filterAfterTimestamp") or 0),
                    payload.get("targetEmail") or payload.get("email") or "",
                )
                json_response(self, 200, {
                    "ok": True,
                    "code": selected["code"],
                    "message": selected["message"],
                    "usedTimeFallback": selected["usedTimeFallback"],
                    "transport": result.get("transport") or "",
                    "settings": result.get("settings") or {},
                })
                return

            email_addr = str(payload.get("email") or "").strip()
            client_id = str(payload.get("clientId") or "").strip()
            refresh_token = str(payload.get("refreshToken") or "").strip()
            if not email_addr or not client_id or not refresh_token:
                raise RuntimeError("Missing email/clientId/refreshToken")

            if self.path == "/messages":
                result = collect_messages(email_addr, client_id, refresh_token, mailboxes, top)
                json_response(self, 200, {
                    "ok": True,
                    "messages": result["messages"],
                    "mailboxResults": result["mailboxResults"],
                    "nextRefreshToken": result["token_payload"].get("next_refresh_token") or "",
                    "tokenEndpoint": result["token_payload"].get("token_endpoint") or "",
                    "transport": result.get("transport") or "",
                })
                return

            if self.path == "/code":
                result = collect_messages(email_addr, client_id, refresh_token, mailboxes, top)
                selected = select_latest_code(
                    result["messages"],
                    payload.get("senderFilters") or [],
                    payload.get("subjectFilters") or [],
                    payload.get("excludeCodes") or [],
                    int(payload.get("filterAfterTimestamp") or 0),
                    payload.get("targetEmail") or "",
                )
                json_response(self, 200, {
                    "ok": True,
                    "code": selected["code"],
                    "message": selected["message"],
                    "usedTimeFallback": selected["usedTimeFallback"],
                    "nextRefreshToken": result["token_payload"].get("next_refresh_token") or "",
                    "tokenEndpoint": result["token_payload"].get("token_endpoint") or "",
                    "transport": result.get("transport") or "",
                })
                return

            json_response(self, 404, {"ok": False, "error": f"Unsupported path: {self.path}"})
        except Exception as exc:
            traceback.print_exc()
            json_response(self, 500, {"ok": False, "error": str(exc)})


def main():
    server = ThreadingHTTPServer((HOST, PORT), HotmailHelperHandler)
    print(f"Hotmail helper listening on http://{HOST}:{PORT}", flush=True)
    print(f"Account log file: {ACCOUNT_LOG_PATH}", flush=True)
    print(f"Account snapshot file: {ACCOUNT_RECORDS_SNAPSHOT_PATH}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()