#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
bulk_send_keys.py
-----------------
Read an Excel exported by imap_curator_export.py, allocate Steam keys from a TXT pool (one per line),
and send HTML emails to curators using a fixed template.
This version:
- Logs ONE ROW PER KEY (not per email).
- Log column order (front):
1) No.
2) Send Date (Beijing, YYYY-MM-DD)
3) Channel (auto from social link; default "Steam")
4) Curator/Name
5) Purpose (fixed "评测")
6) Social Link
- Other columns kept (and adjusted for per-key row):
Key, Mailbox Key, To, Requested Key Count, Subject, Status, Sent At, Test Mode, Actual Recipient
- Subject uses: --subject + " RE: " + original email subject (if present)
- Test mode does NOT consume keys; dry-run does not send/consume.
- Optional external HTML template via --template (fallback to built-in).
Usage examples:
python bulk_send_keys.py --excel curator_requests.xlsx --keys key_pool.txt --out send_log.xlsx \
--subject "Steam Keys for Such A Guy" --dry-run --limit 2
python bulk_send_keys.py --excel curator_requests.xlsx --keys key_pool.txt --out send_log.xlsx \
--subject "Steam Keys for Such A Guy" --test --test-email krystic@such-one.com --limit 1
"""
import os
import sys
import argparse
import smtplib
import ssl
import imaplib
from email.message import EmailMessage
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
from datetime import datetime
from zoneinfo import ZoneInfo
# Built-in fallback template (can be overridden by --template file)
FALLBACK_TEMPLATE = """\
Hi {curator_name},
Thank you for your interest in Such A Guy!
We’re happy to provide you with {key_num} Steam key(s):
{keys_block}
Included materials:
Press kit (capsules, screenshots, short trailer):
Google Drive Press Kit
Review notes (features, estimated length, content warnings):
Google Docs Review Notes
We’d really appreciate your honest impressions after you’ve tried the game —
both on your Steam Curator page and as a store review on Steam.
If you enjoyed the experience, even a short recommendation would help more players discover it. 💫
Best,
Krystic
SUCH ONE STUDIO
"""
def load_template(path: Optional[str]) -> str:
if path and os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
return FALLBACK_TEMPLATE
def load_keys(path: str) -> List[str]:
with open(path, "r", encoding="utf-8") as f:
keys = [line.strip() for line in f if line.strip()]
return keys
def save_remaining_keys(path: str, keys: List[str]) -> None:
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
for k in keys:
f.write(k + "\n")
os.replace(tmp, path)
def render_email_html(template_html: str, curator_name: str, key_num: int, keys: List[str]) -> str:
keys_lines = "
\n".join([k for k in keys])
return template_html.format(curator_name=curator_name or "there",
key_num=key_num,
keys_block=keys_lines)
def send_email(smtp_host: str, smtp_port: int, smtp_user: str, smtp_pass: str,
from_name: str, from_email: str, replyto: str,
to_email: str, subject: str, html: str, dry_run: bool = False) -> None:
if dry_run:
print(f"\n--- DRY RUN (no send) ---\nTO: {to_email}\nSUBJECT: {subject}\nHTML:\n{html}\n-------------------------\n")
return
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email
msg["To"] = to_email
msg["In-Reply-To"] = replyto
msg.set_content("HTML email - please view in a mail client that supports HTML.")
msg.add_alternative(html, subtype="html")
context = ssl.create_default_context()
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.ehlo()
if smtp_port == 587:
server.starttls(context=context)
server.ehlo()
server.login(smtp_user, smtp_pass)
server.send_message(msg)
def parse_channel_and_link(links_str: str) -> Tuple[str, str]:
"""Return (channel, link) based on first recognizable URL; default ('Steam','')."""
if not isinstance(links_str, str) or not links_str.strip():
return ("Steam", "")
# Take first URL if comma-separated
first = links_str.split(",")[0].strip()
low = first.lower()
mapping = [
("store.steampowered.com/curator", "Steam"),
("steamcommunity.com", "Steam"),
("youtu.be", "YouTube"),
("youtube.com", "YouTube"),
("x.com", "Twitter"),
("twitter.com", "Twitter"),
("twitch.tv", "Twitch"),
("discord.gg", "Discord"),
("discord.com", "Discord"),
("facebook.com", "Facebook"),
("instagram.com", "Instagram"),
("tiktok.com", "TikTok"),
("bilibili.com", "Bilibili"),
("weibo.com", "Weibo"),
]
for needle, label in mapping:
if needle in low:
return (label, first)
return ("Steam", first) # default channel name
def _norm_uid(u) -> str:
if isinstance(u, (bytes, bytearray)):
return u.decode("utf-8", errors="ignore").strip()
return str(u).strip()
def _chunked(seq, n):
for i in range(0, len(seq), n):
yield seq[i:i+n]
def imap_mark_answered_batch(host: str, port: int, user: str, pwd: str, mailbox: str,
uids: list[str], batch_size: int = 500) -> int:
"""把一批 UID 标记为 \\Answered，返回成功提交的数量。"""
uids = [_norm_uid(u) for u in uids if _norm_uid(u)]
if not uids:
return 0
M = imaplib.IMAP4_SSL(host, port)
M.login(user, pwd)
typ, _ = M.select(mailbox, readonly=False)
if typ != "OK":
try:
M.close()
except Exception:
pass
M.logout()
return 0
total_ok = 0
for batch in _chunked(uids, batch_size):
seqset = ",".join(batch) # e.g. "444,445,446"
typ1, _ = M.uid("STORE", seqset, "+FLAGS.SILENT", r"(\Answered)")
if typ1 != "OK":
typ2, resp2 = M.uid("STORE", seqset, "+FLAGS", r"(\Answered)")
if typ2 == "OK":
total_ok += len(batch)
else:
# 逐封回退
for uid in batch:
t3, _ = M.uid("STORE", uid, "+FLAGS.SILENT", r"(\Answered)")
if t3 == "OK":
total_ok += 1
else:
t4, _ = M.uid("STORE", uid, "+FLAGS", r"(\Answered)")
if t4 == "OK":
total_ok += 1
else:
total_ok += len(batch)
try:
M.close()
except Exception:
pass
M.logout()
return total_ok
def main():
parser = argparse.ArgumentParser(description="Bulk send Steam keys to curators from Excel (one row per key).")
parser.add_argument("--excel", required=True, help="Path to input Excel (curator_requests.xlsx).")
parser.add_argument("--keys", required=True, help="Path to key pool TXT (one key per line).")
parser.add_argument("--out", default="send_log.xlsx", help="Path to output Excel log (per key rows).")
parser.add_argument("--subject", required=True, help="Base email subject (will prefix ' RE: original subject').")
parser.add_argument("--template", default="email_template.html", help="Path to HTML email template file (optional).")
parser.add_argument("--limit", type=int, default=None, help="Max rows to process from Excel.")
parser.add_argument("--dry-run", action="store_true", help="Render emails only; do not send.")
parser.add_argument("--test", action="store_true", help="Send to test address instead of recipients.")
parser.add_argument("--test-email", default=os.environ.get("TEST_EMAIL", ""), help="Test recipient (with --test).")
parser.add_argument("--skip-sent", action="store_true", help="Skip rows already present (by UID) in the output log.")
parser.add_argument("--no-consume", action="store_true", help="Do not modify key pool file (do not remove used keys).")
# SMTP config (env or CLI)
parser.add_argument("--smtp-host", default=os.environ.get("SMTP_HOST", ""))
parser.add_argument("--smtp-port", type=int, default=int(os.environ.get("SMTP_PORT", "587")))
parser.add_argument("--smtp-user", default=os.environ.get("SMTP_USER", ""))
parser.add_argument("--smtp-pass", default=os.environ.get("SMTP_PASS", ""))
parser.add_argument("--from-name", default=os.environ.get("FROM_NAME", "Krystic"))
parser.add_argument("--from-email", default=os.environ.get("FROM_EMAIL", os.environ.get("SMTP_USER", "")))
# IMAP config for optional marking as answered
parser.add_argument("--mark-answered", action="store_true",
help="After real sends, mark original messages as \\Answered via IMAP.")
parser.add_argument("--imap-host", default=os.environ.get("IMAP_HOST", ""))
parser.add_argument("--imap-port", type=int, default=int(os.environ.get("IMAP_PORT", "993")))
parser.add_argument("--imap-user", default=os.environ.get("EMAIL_USER", os.environ.get("SMTP_USER","")))
parser.add_argument("--imap-pass", default=os.environ.get("EMAIL_PASS", os.environ.get("SMTP_PASS", "")))
parser.add_argument("--imap-mailbox", default=os.environ.get("MAILBOX", "INBOX"))
args = parser.parse_args()
# Validate SMTP when not dry-run
if not args.dry_run:
for vname in ["smtp_host", "smtp_user", "smtp_pass", "from_email"]:
if not getattr(args, vname):
print(f"ERROR: Missing SMTP config --{vname.replace('_','-')} (or env var). Use --dry-run to preview.", file=sys.stderr)
sys.exit(2)
# Load template (external file if exists, otherwise fallback)
template_html = load_template(args.template)
# Load Excel
df = pd.read_excel(args.excel, sheet_name=0)
required_cols = ["Mailbox Key", "Email", "Curator/Name", "Requested Key Count", "Subject", "Curator/Social Links"]
for c in required_cols:
if c not in df.columns:
print(f"ERROR: Excel missing column: {c}", file=sys.stderr)
sys.exit(3)
# Load existing log to support --skip-sent (by UID)
sent_uids = set()
if args.skip_sent and os.path.exists(args.out):
try:
logdf = pd.read_excel(args.out, sheet_name=0)
if "Mailbox Key" in logdf.columns:
sent_uids = set(str(x) for x in logdf["Mailbox Key"].astype(str).tolist())
except Exception:
pass
# Load key pool
pool = load_keys(args.keys)
# Prepare per-key logging
log_rows: List[Dict[str, Any]] = []
row_no = 1
processed = 0
# ===== 进度与汇总统计 =====
# 估算计划处理的“邮件行”总数（考虑 --limit 与 --skip-sent）
if args.limit is not None:
total_target = min(len(df), args.limit)
else:
total_target = len(df)
if args.skip_sent and os.path.exists(args.out):
try:
# 粗略估算，已发过的行会被跳过（只是估算，实际略有出入也没关系）
total_target = max(0, total_target - len(sent_uids))
except Exception:
pass
attempt_rows = 0 # 实际尝试发送的“邮件行”（有成功分配到 key 才算一次尝试）
emails_ok = 0 # 发送成功（SENT 或 SENT_TEST）
emails_fail = 0 # 失败/跳过（ERROR 或 SKIPPED_NO_KEYS 等）
keys_assigned_total = 0 # 实际分配（写进邮件里的）key 数（dry-run/test 也会统计）
uids_to_mark_answered: list[str] = []
# Iterate over Excel rows (one email row)
for idx, row in df.iterrows():
try:
uid = str(row.get("Mailbox Key", "")).strip()
email_to = str(row.get("Email", "")).strip()
name_val = row.get("Curator/Name", "")
if pd.isna(name_val) or not str(name_val).strip():
curator_name = "Curator"
else:
curator_name = str(name_val).strip()
if args.test:
curator_name = curator_name + "(" + email_to + ")"
req_num = row.get("Requested Key Count")
try:
key_num = int(req_num) if pd.notna(req_num) else 2
except Exception:
key_num = 2
if key_num <= 0:
key_num = 2
if args.skip_sent and uid and uid in sent_uids:
continue
if args.limit is not None and processed >= args.limit:
break
# Channel & link detection (robust against NaN)
val = row.get("Curator/Social Links", "")
if pd.isna(val):
links_str = ""
else:
links_str = str(val).strip()
channel, chosen_link = parse_channel_and_link(links_str)
# ✅ 一律将 None/NaN/空白 归一为 ""
safe_link = "" if (chosen_link is None or (isinstance(chosen_link, float) and pd.isna(chosen_link)) or not str(chosen_link).strip()) else str(chosen_link).strip()
# Build email
if len(pool) < key_num:
print(f"WARNING: Not enough keys left for UID {uid}. Needed {key_num}, have {len(pool)}. Skipping.", file=sys.stderr)
status = "SKIPPED_NO_KEYS"
assigned = []
to_addr = ""
final_subject = args.subject
else:
assigned = pool[:key_num]
# subject: --subject + ' RE: ' + original
orig_subject = str(row.get("Subject", "")).strip()
final_subject = args.subject
if orig_subject:
final_subject = f"{args.subject} RE: {orig_subject}"
html = render_email_html(template_html, curator_name, key_num, assigned)
# Decide recipient
to_addr = args.test_email if args.test and args.test_email else (os.environ.get("TEST_EMAIL") if args.test else email_to)
if args.test and not to_addr:
print("ERROR: --test specified but no test email provided. Use --test-email or TEST_EMAIL env.", file=sys.stderr)
sys.exit(4)
# Send (or dry-run)
send_email(
smtp_host=args.smtp_host, smtp_port=args.smtp_port,
smtp_user=args.smtp_user, smtp_pass=args.smtp_pass,
from_name=args.from_name, from_email=args.from_email,
replyto=uid,
to_email=to_addr, subject=final_subject, html=html,
dry_run=args.dry_run
)
status = "SENT_TEST" if args.test else ("DRY_RUN" if args.dry_run else "SENT")
if status == "SENT":
uids_to_mark_answered.append(uid)
# ===== 进度统计（有分配到 key 才算一次尝试）=====
if assigned:
attempt_rows += 1
if status in ("SENT", "SENT_TEST"):
emails_ok += 1
keys_assigned_total += len(assigned)
else:
emails_fail += 1
# ===== 实时进度反馈（不是 dry-run 就打印）=====
if not args.dry_run:
# processed 是你脚本里原本就有的计数器：处理了多少“邮件行”
# total_target 是开始前估算的目标处理量（用于百分比展示）
pct = (processed / total_target * 100) if total_target else 0
print(f"[{processed}/{total_target} | {pct:.1f}%] {status} UID={uid} to={to_addr or email_to} keys={len(assigned)}")
# Consume keys ONLY when real send (not dry-run, not test) and not --no-consume
if not args.no_consume:
pool = pool[key_num:]
# Prepare BJ timestamps
now_bj = datetime.now(ZoneInfo("Asia/Shanghai"))
send_date = now_bj.strftime("%Y-%m-%d") # YYYY-MM-DD
sent_at = now_bj.strftime("%Y-%m-%d %H:%M:%S") # detailed
# Log ONE ROW PER KEY
if assigned:
for k in assigned:
log_rows.append({
"No.": row_no,
"Send Date": send_date,
"Channel": channel or "Steam",
"Curator/Name": curator_name,
"Purpose": "评测",
"Social Link": safe_link,
"Key": k,
"Mailbox Key": uid,
"To": email_to,
"Requested Key Count": key_num,
"Subject": final_subject,
"Status": status,
"Sent At": sent_at,
"Test Mode": bool(args.test),
"Actual Recipient": to_addr if (args.test or args.dry_run) else email_to,
})
row_no += 1
else:
# Even if skipped/no keys, write a single row for traceability (without Key)
log_rows.append({
"No.": row_no,
"Send Date": send_date,
"Channel": channel or "Steam",
"Curator/Name": curator_name,
"Purpose": "评测",
"Social Link": chosen_link,
"Key": "",
"Mailbox Key": uid,
"To": email_to,
"Requested Key Count": key_num,
"Subject": final_subject,
"Status": status,
"Sent At": sent_at,
"Test Mode": bool(args.test),
"Actual Recipient": to_addr if (args.test or args.dry_run) else email_to,
})
row_no += 1
processed += 1
except Exception as e:
now_bj = datetime.now(ZoneInfo("Asia/Shanghai"))
log_rows.append({
"No.": row_no,
"Send Date": now_bj.strftime("%Y-%m-%d"),
"Channel": "Steam",
"Curator/Name": str(row.get("Curator/Name", "")),
"Purpose": "评测",
"Social Link": str(row.get("Curator/Social Links", "")),
"Key": "",
"Mailbox Key": str(row.get("Mailbox Key", "")),
"To": str(row.get("Email", "")),
"Requested Key Count": row.get("Requested Key Count"),
"Subject": str(row.get("Subject", "")),
"Status": f"ERROR:{e}",
"Sent At": now_bj.strftime("%Y-%m-%d %H:%M:%S"),
"Test Mode": bool(args.test),
"Actual Recipient": "",
})
row_no += 1
print(f"ERROR processing row {idx}: {e}", file=sys.stderr)
continue
if args.mark_answered and not args.dry_run and not args.test and uids_to_mark_answered:
done = imap_mark_answered_batch(
host=args.imap_host, port=args.imap_port,
user=args.imap_user, pwd=args.imap_pass,
mailbox=args.imap_mailbox, uids=uids_to_mark_answered, batch_size=500
)
print(f"[INFO] Marked {done} message(s) as \\Answered.")
# Build DataFrame with desired column order
columns = [
"No.", "Send Date", "Channel", "Curator/Name", "Purpose", "Social Link",
"Key", "Mailbox Key", "To", "Requested Key Count", "Subject",
"Status", "Sent At", "Test Mode", "Actual Recipient"
]
log_df = pd.DataFrame(log_rows, columns=columns)
log_path = os.path.abspath(args.out)
with pd.ExcelWriter(log_path, engine="openpyxl") as writer:
log_df.to_excel(writer, sheet_name="SendLog", index=False)
# ===== 汇总报告 =====
mode = "REAL"
if args.dry_run:
mode = "DRY-RUN"
elif args.test:
mode = "TEST"
print("\n=== RUN SUMMARY ===")
print(f"Mode: {mode}")
print(f"Email rows processed: {processed}")
print(f"Attempted sends: {attempt_rows} (rows that had keys assigned)")
print(f"Successful sends: {emails_ok}")
print(f"Failed/Skipped: {emails_fail}")
print(f"Keys assigned total: {keys_assigned_total}")
print(f"Log file: {log_path}")
# Save remaining keys if consuming (not dry-run/test)
if not args.no_consume and not args.dry_run and not args.test:
save_remaining_keys(args.keys, pool)
print(f"Done. Processed {processed} email row(s). Logged {len(log_df)} key row(s). File: {log_path}")
if not args.no_consume:
print(f"Current in-memory remaining keys (not saved if test/dry-run): {len(pool)}")
if __name__ == "__main__":
main()