#!/usr/bin/env python3 import argparse import email import imaplib import os import smtplib import ssl import sys from email.header import decode_header, make_header from email.message import EmailMessage from email.utils import parsedate_to_datetime def env(name, default=None, required=False): value = os.getenv(name, default) if required and not value: print(f"Missing environment variable: {name}", file=sys.stderr) sys.exit(1) return value QQMAIL_ADDRESS = env("QQMAIL_ADDRESS", required=True) QQMAIL_AUTH_CODE = env("QQMAIL_AUTH_CODE", required=True) IMAP_HOST = env("QQMAIL_IMAP_HOST", "imap.qq.com") IMAP_PORT = int(env("QQMAIL_IMAP_PORT", "993")) SMTP_HOST = env("QQMAIL_SMTP_HOST", "smtp.qq.com") SMTP_PORT = int(env("QQMAIL_SMTP_PORT", "465")) def decode_text(value): if value is None: return "" try: return str(make_header(decode_header(value))) except Exception: return str(value) def connect_imap(): imap = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) imap.login(QQMAIL_ADDRESS, QQMAIL_AUTH_CODE) imap.select("INBOX") return imap def fetch_uids(imap, criteria, limit): typ, data = imap.uid("search", None, criteria) if typ != "OK": raise RuntimeError(f"search failed: {typ}") uids = data[0].split() return list(reversed(uids))[:limit] def fetch_message_meta(imap, uid): typ, data = imap.uid("fetch", uid, "(BODY.PEEK[HEADER.FIELDS (FROM TO SUBJECT DATE)])") if typ != "OK": raise RuntimeError(f"fetch failed for UID {uid.decode()}") msg = email.message_from_bytes(data[0][1]) date_raw = msg.get("Date", "") try: dt = parsedate_to_datetime(date_raw) date_fmt = dt.isoformat() except Exception: date_fmt = date_raw return { "uid": uid.decode(), "from": decode_text(msg.get("From", "")), "to": decode_text(msg.get("To", "")), "subject": decode_text(msg.get("Subject", "")), "date": date_fmt, } def get_body_text(msg): if msg.is_multipart(): parts = [] for part in msg.walk(): ctype = part.get_content_type() disp = str(part.get("Content-Disposition", "")) if "attachment" in disp.lower(): continue if ctype == "text/plain": payload = part.get_payload(decode=True) charset = part.get_content_charset() or "utf-8" if payload: try: parts.append(payload.decode(charset, errors="replace")) except Exception: parts.append(payload.decode("utf-8", errors="replace")) if parts: return "\n".join(parts).strip() for part in msg.walk(): if part.get_content_type() == "text/html": payload = part.get_payload(decode=True) if payload: return payload.decode(part.get_content_charset() or "utf-8", errors="replace").strip() return "" payload = msg.get_payload(decode=True) if not payload: return "" charset = msg.get_content_charset() or "utf-8" try: return payload.decode(charset, errors="replace").strip() except Exception: return payload.decode("utf-8", errors="replace").strip() def cmd_list(args, unread=False): imap = connect_imap() try: criteria = '(UNSEEN)' if unread else 'ALL' uids = fetch_uids(imap, criteria, args.limit) for uid in uids: meta = fetch_message_meta(imap, uid) print(f"UID: {meta['uid']}") print(f"From: {meta['from']}") print(f"Subject: {meta['subject']}") print(f"Date: {meta['date']}") print("---") finally: try: imap.logout() except Exception: pass def cmd_search(args): imap = connect_imap() try: criteria = [] if args.from_addr: criteria += ['FROM', f'"{args.from_addr}"'] if args.subject: criteria += ['SUBJECT', f'"{args.subject}"'] if not criteria: print("Provide --from or --subject", file=sys.stderr) sys.exit(1) search_criteria = "(" + " ".join(criteria) + ")" uids = fetch_uids(imap, search_criteria, args.limit) for uid in uids: meta = fetch_message_meta(imap, uid) print(f"UID: {meta['uid']}") print(f"From: {meta['from']}") print(f"Subject: {meta['subject']}") print(f"Date: {meta['date']}") print("---") finally: try: imap.logout() except Exception: pass def cmd_read(args): imap = connect_imap() try: typ, data = imap.uid("fetch", args.uid.encode(), "(RFC822)") if typ != "OK": raise RuntimeError(f"fetch failed for UID {args.uid}") msg = email.message_from_bytes(data[0][1]) print(f"From: {decode_text(msg.get('From', ''))}") print(f"To: {decode_text(msg.get('To', ''))}") print(f"Subject: {decode_text(msg.get('Subject', ''))}") print(f"Date: {decode_text(msg.get('Date', ''))}") print("---") print(get_body_text(msg)) finally: try: imap.logout() except Exception: pass def cmd_send(args): msg = EmailMessage() msg["From"] = QQMAIL_ADDRESS msg["To"] = args.to msg["Subject"] = args.subject msg.set_content(args.body) context = ssl.create_default_context() with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context) as smtp: smtp.login(QQMAIL_ADDRESS, QQMAIL_AUTH_CODE) smtp.send_message(msg) print("SEND_OK") def main(): parser = argparse.ArgumentParser(description="Read and send QQ Mail via IMAP/SMTP") sub = parser.add_subparsers(dest="cmd", required=True) p_list = sub.add_parser("list") p_list.add_argument("--limit", type=int, default=10) p_unread = sub.add_parser("unread") p_unread.add_argument("--limit", type=int, default=10) p_search = sub.add_parser("search") p_search.add_argument("--from", dest="from_addr") p_search.add_argument("--subject") p_search.add_argument("--limit", type=int, default=10) p_read = sub.add_parser("read") p_read.add_argument("--uid", required=True) p_send = sub.add_parser("send") p_send.add_argument("--to", required=True) p_send.add_argument("--subject", required=True) p_send.add_argument("--body", required=True) args = parser.parse_args() if args.cmd == "list": cmd_list(args, unread=False) elif args.cmd == "unread": cmd_list(args, unread=True) elif args.cmd == "search": cmd_search(args) elif args.cmd == "read": cmd_read(args) elif args.cmd == "send": cmd_send(args) if __name__ == "__main__": main()