support-email/fetch_requests.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
imap_curator_export.py
----------------------
Fetch emails via IMAP, find Steam curator key requests since a chosen date,
and export results to an Excel file.

Features:
- Connectivity test (--test) that does NOT read/write state.
- Stateful resume via UID bookmarking (JSON). Next run scans only newer UIDs.
- Extracts curator/social links and requested key counts (defaults to 2 if not found).
- Filters OUT reply/forward messages (replies to you), detected via headers and subject prefixes.
- Outputs Excel with requested column order:
    1) No. (auto-increment)
    2) Mailbox Key (UID of the message)
    3) Requested Key Count
    4) Date (Beijing time, "YYYY-MM-DD HH:MM")
    5) Curator/Name
    6) Email
    7) Subject
    8) Curator/Social Links
    9) Body (preview)
- Prints a summary after export:
    * Total keys requested
    * Keys with social links present
    * Keys without social links

USAGE
-----
1) Install deps (ideally in a virtualenv):
   pip install -r requirements.txt

2) Set environment variables (examples for macOS/Linux bash/zsh):
   export IMAP_HOST="imap.gmail.com"         # e.g. imap.gmail.com, outlook.office365.com
   export IMAP_PORT="993"                    # usually 993
   export EMAIL_USER="your_email@example.com"
   export EMAIL_PASS="your_app_password"     # For Gmail/Outlook, use an App Password or OAuth token string
   export MAILBOX="INBOX"                    # optional, default INBOX
   # Start date used ONLY for the first run (no state yet). IMAP format DD-Mon-YYYY.
   export START_DATE="18-Oct-2025"
   # Optional: where to store/read bookmark state per mailbox
   export STATE_FILE="curator_state.json"

3) Run a connectivity test (no state read/write):
  python fetch_requests.py --test

4) Run extraction:
  python fetch_requests.py

5) Reset bookmark (force full scan from START_DATE next time):
  python fetch_requests.py --reset-state

Output:
   ./curator_requests.xlsx
"""

import os
import re
import sys
import json
import imaplib
import email
import argparse
from email.header import decode_header, make_header
from email.policy import default as default_policy
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from typing import List, Dict, Any, Optional

# Third-party
from bs4 import BeautifulSoup  # for HTML -> text
from openpyxl import load_workbook
import pandas as pd

import re

CURATOR_LINK_PATTERNS = [
    r"store\.steampowered\.com/curator/",
    r"steamcommunity\.com/groups/",
    r"steamcommunity\.com/id/",
    r"x\.com/|twitter\.com/",
    r"youtube\.com/|youtu\.be/",
    r"twitch\.tv/",
    r"discord\.gg/|discord\.com/invite/",
    r"facebook\.com/",
    r"instagram\.com/",
    r"tiktok\.com/",
    r"bilibili\.com/",
    r"weibo\.com/",
]

# Heuristics for detecting a request for Steam keys
KEY_REQUEST_PATTERNS = [
    r"\b(\d{1,3})\s*(?:keys?|key codes?|steam\s*keys?)\b",
    r"(?:需要|申请|索取|来点|要)\s*(\d{1,3})\s*(?:个)?\s*(?:key|激活码|序列号|钥匙)",
    r"\bup to\s*(\d{1,3})\s*keys?\b",
    r"\brequest(?:ing)?\s*(\d{1,3})\s*keys?\b",
    r"\b(\d{1,3})\s*x\s*keys?\b",
    r"\b(\d{1,3})-(\d{1,3})\s*keys?\b",
    r"\b(\d{1,3})\s*(?:steam\s*)?(?:activation\s+(?:codes?|keys?))\b",
]

CURATOR_KEYWORDS = [
    "curator", "steam curator", "reviewer",
    "鉴赏家", "评测", "媒体", "KOL", "influencer", "press",
    "key", "keys", "激活码", "序列号", "steam key",
]

# Subject prefixes indicating replies/forwards (EN/ZH)
REPLY_FWD_PREFIX = re.compile(
    r"^\s*(re(\[\d+\])?|回复|答复|答覆|转发|fw|fwd)\s*[:：]\s*",
    re.IGNORECASE
)

DEFAULT_CONFIG_PATH = os.path.join("configs", "settings.json")

def load_settings(config_path: Optional[str]) -> Dict[str, Any]:
    path = config_path or DEFAULT_CONFIG_PATH
    if not os.path.isabs(path):
        path = os.path.abspath(path)
    if not os.path.exists(path):
        print(f"ERROR: settings file not found: {path}", file=sys.stderr)
        sys.exit(2)
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        print(f"ERROR: failed to read settings.json: {e}", file=sys.stderr)
        sys.exit(2)
    files = data.get("files", {})
    for key in ["excel", "state", "template", "keys", "log"]:
        if key in files and not os.path.isabs(files[key]):
            files[key] = os.path.abspath(files[key])
    data["files"] = files
    return data

def ensure_parent_dir(path: str) -> None:
    directory = os.path.dirname(path)
    if directory:
        os.makedirs(directory, exist_ok=True)

def extract_name_from_body(body: str) -> str:
    """
    从正文中根据常见短语提取名字：
    例如 "My name is Alex", "I am John", "This is Lily"。
    """
    if not body:
        return ""

    text = body.strip().replace("\r", "").replace("\n", " ")

    # 常见英语自我介绍
    patterns = [
        r"\bmy name is ([A-Z][a-z]+(?: [A-Z][a-z]+)?)",
        r"\bi am ([A-Z][a-z]+(?: [A-Z][a-z]+)?)",
        r"\bthis is ([A-Z][a-z]+(?: [A-Z][a-z]+)?)",
    ]

    for pat in patterns:
        m = re.search(pat, text, flags=re.IGNORECASE)
        if m:
            # 返回匹配到的名字（首字母大写格式化）
            name = m.group(1).strip()
            return name.title()

    # 末尾签名行的简易提取（如 “Thanks, Alex”）
    m = re.search(r"(?:regards|thanks|cheers)[,:\s]+([A-Z][a-z]+(?: [A-Z][a-z]+)?)", text, flags=re.IGNORECASE)
    if m:
        return m.group(1).strip().title()

    return ""

def norm_email(addr: str) -> str:
    return (addr or "").strip().lower()

def chunked(seq, n):
    for i in range(0, len(seq), n):
        yield seq[i:i+n]

def mark_seen_batch(M, mailbox: str, uids: list[str], batch_size: int = 500) -> int:
    """
    以批为单位，把同一 mailbox 中的多封邮件标记为已读。
    返回成功提交 STORE 的 UID 总数（不逐一校验）。
    """
    if not uids:
        return 0
    # 切换到可写
    typ, _ = M.select(mailbox, readonly=False)
    if typ != "OK":
        print(f"[WARN] cannot select {mailbox} in write mode; skip mark-read")
        return 0

    total = 0
    for batch in chunked(uids, batch_size):
        # UID 序列（逗号拼接）
        seqset = ",".join(str(u.decode() if isinstance(u, (bytes, bytearray)) else u) for u in batch)

        # 先静默，减少服务器返回
        typ, _ = M.uid("STORE", seqset, "+FLAGS.SILENT", r"(\Seen)")
        if typ != "OK":
            # 兼容某些服务器
            typ, _ = M.uid("STORE", seqset, "+FLAGS", r"(\Seen)")
        if typ == "OK":
            total += len(batch)
        else:

            print(f"[WARN] batch STORE failed for {len(batch)} UIDs (len={len(seqset)})")

    return total


def imap_login(conf: Dict[str, Any]) -> imaplib.IMAP4_SSL:
    host = conf.get("host")
    port = int(conf.get("port", 993))
    user = conf.get("user")
    pwd  = conf.get("pass")
    if not (host and user and pwd):
        print("ERROR: Missing IMAP configuration in settings.json", file=sys.stderr)
        sys.exit(1)
    M = imaplib.IMAP4_SSL(host, port)
    M.login(user, pwd)
    return M


def select_mailbox(M: imaplib.IMAP4_SSL, mailbox: str) -> str:
    typ, data = M.select(mailbox, readonly=True)
    if typ != "OK":
        raise RuntimeError(f"Cannot select mailbox {mailbox}: {typ} {data}")
    return mailbox


def imap_date_string(start_date: str) -> str:
    try:
        datetime.strptime(start_date, "%d-%b-%Y")
    except Exception:
        print("WARNING: START_DATE is not in 'DD-Mon-YYYY' (e.g., 18-Oct-2025). Using 18-Oct-2025 by default.", file=sys.stderr)
        start_date = "18-Oct-2025"
    return start_date


def load_state(path: str) -> Dict[str, Any]:
    if not os.path.exists(path):
        return {"_path": path, "mailboxes": {}}
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        data["_path"] = path
        if "mailboxes" not in data or not isinstance(data["mailboxes"], dict):
            data["mailboxes"] = {}
        return data
    except Exception:
        return {"_path": path, "mailboxes": {}}


def save_state(state: Dict[str, Any]) -> None:
    path = state.get("_path")
    if not path:
        return
    ensure_parent_dir(path)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump({k: v for k, v in state.items() if k != "_path"}, f, ensure_ascii=False, indent=2)
    os.replace(tmp, path)


def get_mailbox_key(host: str, user: str, mailbox: str) -> str:
    return f"{host}|{user}|{mailbox}"


def uid_search(M: imaplib.IMAP4_SSL, criterion: str) -> List[bytes]:
    typ, data = M.uid("search", None, criterion)
    if typ != "OK":
        raise RuntimeError(f"UID SEARCH failed: {typ} {data}")
    return data[0].split()


def search_initial(M: imaplib.IMAP4_SSL, since_date: str) -> List[bytes]:
    typ, data = M.search(None, f'(SINCE "{since_date}")')
    if typ != "OK":
        raise RuntimeError(f"SEARCH failed: {typ} {data}")
    ids = data[0].split()
    uids: List[bytes] = []
    if not ids:
        return uids
    for i in range(0, len(ids), 500):
        batch = ids[i:i+500]
        typ2, data2 = M.fetch(b",".join(batch), "(UID)")
        if typ2 != "OK":
            continue
        for part in data2:
            if not isinstance(part, tuple):
                continue
            header = part[0].decode("utf-8", "ignore")
            m = re.search(r"UID\s+(\d+)", header)
            if m:
                uids.append(m.group(1).encode())
    return uids


def decode_str(s: Optional[str]) -> str:
    if not s:
        return ""
    try:
        return str(make_header(decode_header(s)))
    except Exception:
        return s


def get_address(msg: email.message.Message) -> (str, str):
    from_raw = msg.get("From", "")
    name = email.utils.parseaddr(from_raw)[0]
    addr = email.utils.parseaddr(from_raw)[1]
    return decode_str(name).strip(), addr.strip()


def get_subject(msg: email.message.Message) -> str:
    return decode_str(msg.get("Subject", "")).strip()


def get_payload_text(msg: email.message.Message) -> (str, str):
    plain_parts = []
    html_parts = []

    if msg.is_multipart():
        for part in msg.walk():
            ctype = part.get_content_type()
            disp = str(part.get_content_disposition() or "").lower()
            if disp == "attachment":
                continue
            try:
                payload = part.get_payload(decode=True)
            except Exception:
                payload = None

            if payload is None:
                continue

            charset = part.get_content_charset() or "utf-8"
            try:
                text = payload.decode(charset, errors="replace")
            except Exception:
                text = payload.decode("utf-8", errors="replace")

            if ctype == "text/plain":
                plain_parts.append(text)
            elif ctype == "text/html":
                html_parts.append(text)
    else:
        payload = msg.get_payload(decode=True) or b""
        charset = msg.get_content_charset() or "utf-8"
        try:
            text = payload.decode(charset, errors="replace")
        except Exception:
            text = payload.decode("utf-8", errors="replace")
        if msg.get_content_type() == "text/html":
            html_parts.append(text)
        else:
            plain_parts.append(text)

    return ("\n".join(plain_parts).strip(), "\n".join(html_parts).strip())


def html_to_text(html: str) -> str:
    if not html:
        return ""
    soup = BeautifulSoup(html, "html.parser")
    for tag in soup(["script", "style"]):
        tag.decompose()
    return soup.get_text(separator="\n").strip()


def extract_links(text: str) -> List[str]:
    if not text:
        return []
    urls = re.findall(r"https?://[^\s<>()\"\']+", text, flags=re.IGNORECASE)
    seen = set()
    out = []
    for u in urls:
        if u not in seen:
            seen.add(u)
            out.append(u)
    return out


def filter_curator_links(urls: List[str]) -> List[str]:
    if not urls:
        return []
    combined = "|".join(CURATOR_LINK_PATTERNS)
    pat = re.compile(combined, re.IGNORECASE)
    return [u for u in urls if pat.search(u)]


def detect_key_count(text: str) -> Optional[int]:
    if not text:
        return None
    best = None
    for pat in KEY_REQUEST_PATTERNS:
        m = re.search(pat, text, flags=re.IGNORECASE)
        if m:
            if m.lastindex and m.lastindex >= 2 and m.group(1) and m.group(2):
                try:
                    a = int(m.group(1))
                    b = int(m.group(2))
                    best = max(a, b)
                    break
                except Exception:
                    continue
            else:
                nums = [int(g) for g in m.groups() if g and g.isdigit()]
                if nums:
                    best = max(nums)
                    break
    return best


def looks_like_curator_request(subject: str, body_text: str) -> bool:
    blob = f"{subject}\n{body_text}".lower()
    return any(k.lower() in blob for k in CURATOR_KEYWORDS)


def fetch_by_uid(M: imaplib.IMAP4_SSL, uid: bytes) -> email.message.Message:
    typ, data = M.uid("fetch", uid, "(RFC822)")
    if typ != "OK" or not data or not isinstance(data[0], tuple):
        raise RuntimeError(f"UID FETCH failed for {uid!r}: {typ} {data}")
    raw = data[0][1]
    msg = email.message_from_bytes(raw, policy=default_policy)
    return msg


def parse_msg_date_bj(msg: email.message.Message) -> str:
    raw = msg.get("Date") or ""
    try:
        dt = parsedate_to_datetime(raw)
        if dt is None:
            raise ValueError("parsedate_to_datetime returned None")
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        dt_cst = dt.astimezone(ZoneInfo("Asia/Shanghai"))
        return dt_cst.strftime("%Y-%m-%d %H:%M")
    except Exception:
        return ""


def is_reply_or_forward(msg: email.message.Message, subject: str) -> bool:
    # Header-based
    if msg.get("In-Reply-To") or msg.get("References"):
        return True
    # Subject-based
    if REPLY_FWD_PREFIX.search(subject):
        return True
    return False


def fetch_and_parse(M: imaplib.IMAP4_SSL, uid: bytes) -> Dict[str, Any]:
    msg = fetch_by_uid(M, uid)
    name, addr = get_address(msg)
    subject = get_subject(msg)
    date_local = parse_msg_date_bj(msg)
    reply_flag = is_reply_or_forward(msg, subject)
    message_id = (msg.get("Message-ID") or "").strip()

    plain, html = get_payload_text(msg)
    merged_text = plain.strip()
    if html and (not merged_text or len(merged_text) < 20):
        merged_text = html_to_text(html)

    links_all = extract_links(plain + "\n" + html)
    curator_links = filter_curator_links(links_all)

    key_count = detect_key_count(merged_text)
    if key_count is None:
        key_count = 2  # default when not specified

    return {
        "uid": uid.decode() if isinstance(uid, (bytes, bytearray)) else str(uid),
        "from_name": name or "",
        "from_email": addr or "",
        "subject": subject,
        "message_id": message_id,
        "date_local": date_local,
        "body_preview": (merged_text[:3000] + ("..." if len(merged_text) > 3000 else "")),
        "curator_links": curator_links,
        "key_count": int(key_count),
        "is_reply": reply_flag,
    }


def connectivity_test(settings: Dict[str, Any]) -> int:
    try:
        imap_conf = settings.get("imap", {})
        mailbox_name = imap_conf.get("mailbox", "INBOX")
        M = imap_login(imap_conf)
        try:
            mailbox = select_mailbox(M, mailbox_name)
            uids = uid_search(M, "ALL")
            count = len(uids)
            if count == 0:
                print(f"[TEST] Connected to {mailbox}, but it has no messages.")
                return 0
            latest_uid = uids[-1]
            msg = fetch_by_uid(M, latest_uid)
            subject = get_subject(msg)
            from_name, from_addr = get_address(msg)
            date_local = parse_msg_date_bj(msg)
            print("[TEST] IMAP OK.")
            print(f"[TEST] Mailbox: {mailbox}")
            print(f"[TEST] Total messages: {count}")
            print(f"[TEST] Latest UID: {latest_uid.decode()}")
            print(f"[TEST] Latest From: {from_name} <{from_addr}>")
            print(f"[TEST] Latest Subject: {subject}")
            print(f"[TEST] Latest Date (BJ): {date_local}")
            return 0
        finally:
            try:
                M.logout()
            except Exception:
                pass
    except Exception as e:
        print(f"[TEST] IMAP failed: {e}", file=sys.stderr)
        return 2


def run_export(settings: Dict[str, Any], reset_state: bool = False, mark_read: bool = False) -> int:
    files_conf = settings.get("files", {})
    state_path = os.path.abspath(files_conf.get("state", os.path.join("config", "curator_state.json")))
    out_path = os.path.abspath(files_conf.get("excel", os.path.join("records", "curator_requests.xlsx")))
    start_date = imap_date_string(settings.get("start_date", "18-Oct-2025"))
    imap_conf = settings.get("imap", {})
    mailbox_name = imap_conf.get("mailbox", "INBOX")
    host = imap_conf.get("host", "")
    user = imap_conf.get("user", "")
    min_uid = int(settings.get("min_uid", 125))

    M = imap_login(imap_conf)
    state = load_state(state_path)
    try:
        mailbox = select_mailbox(M, mailbox_name)
        mailbox_key = get_mailbox_key(host, user, mailbox)

        if reset_state and mailbox_key in state["mailboxes"]:
            del state["mailboxes"][mailbox_key]
            save_state(state)
            print(f"[STATE] Reset bookmark for {mailbox_key}")

        last_uid = None
        if mailbox_key in state["mailboxes"]:
            last_uid = state["mailboxes"][mailbox_key].get("last_uid")

        if last_uid:
            criterion = f"(UID {int(last_uid)+1}:*)"
            uids = uid_search(M, criterion)
            print(f"[SCAN] Using bookmark last_uid={last_uid}; new UIDs found: {len(uids)}")
        else:
            uids = search_initial(M, start_date)
            print(f"[SCAN] Initial run since {start_date}; candidate UIDs: {len(uids)}")

        rows: List[Dict[str, Any]] = []
        all_uids_for_mark: list[str] = []
        max_seen_uid = int(last_uid) if last_uid else 0
        row_no = 1

        best_by_email: dict[str, dict] = {}
        dup_skipped = 0

        interrupted = False
        try:
            for i, uid in enumerate(uids, 1):
                try:
                    uid_int = int(uid)
                    if uid_int < min_uid:
                        continue

                    rec = fetch_and_parse(M, uid)
                    if uid_int > max_seen_uid:
                        max_seen_uid = uid_int

                    # Exclude replies/forwards
                    if rec["is_reply"]:
                        continue

                    has_links = False
                    if looks_like_curator_request(rec["subject"], rec["body_preview"]):
                        has_links = bool(rec["curator_links"])
                        curator_email = norm_email(rec["from_email"])
                        dedup_key = curator_email if curator_email else f"uid:{rec['uid']}"


                        # 首选：邮件头中的名字
                        if rec["from_name"]:
                            curator_name = rec["from_name"].strip()
                        else:
                            # 尝试从正文里提取名字
                            extracted = extract_name_from_body(rec["body_preview"])
                            if extracted:
                                curator_name = extracted
                            else:
                                curator_name = "Curator"

                        candidate = {
                            "uid_int": int(uid),
                            "record": {
                                "Mailbox Key": rec["uid"],                 # UID
                                "Requested Key Count": rec["key_count"],
                                "Date": rec["date_local"],
                                "Curator/Name": curator_name,
                                "Email": rec["from_email"],
                                "Subject": rec["subject"],
                                "Curator/Social Links": ", ".join(rec["curator_links"]) if has_links else "",
                                "Body (preview)": rec["body_preview"],
                                "Original Message-ID": rec.get("message_id", ""),
                                "_has_links": has_links,
                            }
                        }
                        all_uids_for_mark.append(int(uid))
                        prev = best_by_email.get(dedup_key)
                        if prev is None or candidate["uid_int"] > prev["uid_int"]:
                            best_by_email[dedup_key] = candidate
                        else:
                            dup_skipped += 1

                        if i % 10 == 0:
                            pct = (i / len(uids)) * 100
                            print(f"  Processed {pct:.1f}% ({i}/{len(uids)})")
                except Exception as e:
                    print(f"[WARN] Failed to parse UID {uid!r}: {e}", file=sys.stderr)
                    continue
        except KeyboardInterrupt:
            interrupted = True
            print("\n[INTERRUPTED] Stopping scan early. Partial results will be written.")

        if mark_read:
            done = mark_seen_batch(M, mailbox, all_uids_for_mark, batch_size=500)
            print(f"[INFO] Marked {done} message(s) as read in batches.")

        rows = []
        row_no = 1
        selected_uids_for_mark = []

        for _, v in best_by_email.items():
            rec = v["record"]
            rows.append({
                "No.": row_no,
                "Mailbox Key": rec["Mailbox Key"],
                "Requested Key Count": rec["Requested Key Count"],
                "Date": rec["Date"],
                "Curator/Name": rec["Curator/Name"],
                "Email": rec["Email"],
                "Subject": rec["Subject"],
                "Curator/Social Links": rec["Curator/Social Links"],
                "Body (preview)": rec["Body (preview)"],
                "Original Message-ID": rec["Original Message-ID"],
                "_has_links": rec["_has_links"],
            })
            selected_uids_for_mark.append(rec["Mailbox Key"])
            row_no += 1

        # Save bookmark even if no rows matched, so daily runs skip already-seen messages
        state["mailboxes"][mailbox_key] = {"last_uid": str(max_seen_uid)}
        save_state(state)
        print(f"[STATE] Updated last_uid={max_seen_uid} for {mailbox_key}")

        columns = [
            "No.", "Mailbox Key", "Requested Key Count", "Date", "Curator/Name",
            "Email", "Subject", "Curator/Social Links", "Body (preview)", "Original Message-ID"
        ]

        if not rows:
            print("No curator key requests matched the filters.")
            df = pd.DataFrame(columns=columns)
            total_keys = with_links = without_links = 0
        else:
            df = pd.DataFrame(rows)
            df["Requested Key Count"] = pd.to_numeric(df["Requested Key Count"], errors="coerce").fillna(0).astype(int)
            total_keys = int(df["Requested Key Count"].sum())
            with_links = int(df.loc[df["_has_links"], "Requested Key Count"].sum()) if "_has_links" in df.columns else 0
            without_links = total_keys - with_links
            # drop helper
            if "_has_links" in df.columns:
                df = df.drop(columns=["_has_links"])
            for col in columns:
                if col not in df.columns:
                    df[col] = ""
            df = df[columns]

        ensure_parent_dir(out_path)
        date_tag = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
        sheet_prefix = f"curtor_{date_tag}"
        existing_sheets = set()
        if os.path.exists(out_path):
            try:
                wb = load_workbook(out_path, read_only=True)
                existing_sheets = set(wb.sheetnames)
                wb.close()
            except Exception:
                existing_sheets = set()

        suffix = 1
        while True:
            sheet_name = f"{sheet_prefix}{suffix:02d}"
            if sheet_name not in existing_sheets:
                break
            suffix += 1

        writer_args: Dict[str, Any] = {"engine": "openpyxl"}
        if os.path.exists(out_path):
            writer_args["mode"] = "a"
        with pd.ExcelWriter(out_path, **writer_args) as writer:
            df.to_excel(writer, sheet_name=sheet_name, index=False)
            book = writer.book
            if sheet_name in book.sheetnames and book.sheetnames[0] != sheet_name:
                ws = book[sheet_name]
                sheets = book._sheets  # type: ignore[attr-defined]
                sheets.insert(0, sheets.pop(sheets.index(ws)))

        # Summary printout
        print("\n=== SUMMARY ===")
        print(f"Total requested keys: {total_keys}")
        print(f"With social links:    {with_links}")
        print(f"Without social links: {without_links}")
        if interrupted:
            print("[INTERRUPTED] Export stopped early; only the processed portion is included.")

        print(f"\nExported {len(df)} row(s) to {out_path}")
        print("\nDRY-RUN 命令：")
        print(f"  python send_keys.py \\")
        print(f"    --excel {os.path.basename(out_path)} \\")
        print(f"    --sheet {sheet_name} \\")
        print(f"    --template \"email_template.html\" \\")
        print(f"    --keys steam_key.txt \\")
        print(f"    --out send_log.xlsx \\")
        print(f"    --subject \"Steam Keys for Such A Guy - \" \\")
        print(f"    --dry-run")
        print("\nTEST 命令：")
        print(f"  python send_keys.py \\")
        print(f"    --excel {os.path.basename(out_path)} \\")
        print(f"    --sheet {sheet_name} \\")
        print(f"    --template \"email_template.html\" \\")
        print(f"    --keys steam_key.txt \\")
        print(f"    --out send_log.xlsx \\")
        print(f"    --subject \"Steam Keys for Such A Guy -\" \\")
        print(f"    --test \\")
        print(f"    --no-consume --no-sentemail")
        print("\n正式执行命令：")
        print(f"  python send_keys.py \\")
        print(f"    --excel {os.path.basename(out_path)} \\")
        print(f"    --sheet {sheet_name} \\")
        print(f"    --template \"email_template.html\" \\")
        print(f"    --keys steam_key.txt \\")
        print(f"    --out send_log.xlsx \\")
        print(f"    --subject \"Steam Keys for Such A Guy -\" \\")
        print(f"    --mark-answered")
        return 0
    finally:
        try:
            M.logout()
        except Exception:
            pass


def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Export Steam curator key requests to Excel via IMAP.")
    p.add_argument("--config", help="Path to settings.json (default: configs/settings.json).")
    p.add_argument("--test", action="store_true", help="Run a quick IMAP connectivity test and exit (does not read/write state).")
    p.add_argument("--reset-state", action="store_true", help="Reset stored UID bookmark before running.")
    p.add_argument("--mark-read", action="store_true", help="After exporting, mark those emails as read on the IMAP server.")
    return p.parse_args(argv)


def main(argv: Optional[List[str]] = None) -> int:
    args = parse_args(argv)
    config_path = args.config or os.environ.get("SETTINGS_FILE") or DEFAULT_CONFIG_PATH
    settings = load_settings(config_path)
    if args.test:
        return connectivity_test(settings)   # no state read/write
    return run_export(settings=settings, reset_state=args.reset_state, mark_read=args.mark_read)


if __name__ == "__main__":
    raise SystemExit(main())