主题QQ邮箱的实时监控和自动回复实现QQ邮箱的实时监控和自动回复思路代码1. 获取QQ邮箱授权码只有开启了QQ邮箱的IMAP SMTP服务才能路径登录QQ邮箱-设置-账号与安全-开启IMAP/SMTP服务大概在这个位置你会看到一个 ““POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务”类似的内容点击下方的“开启服务”我这边因为已经开启过了所以看不到补充了解即可开启IMAP/SMTP服务按照提示生成一串16位的授权码务必保存好2. 读取配置文件创建配置文件读取配置文件def load_config(config_file): 加载配置文件如果需要 try: with open(config_file, r, encodingutf-8) as f: config {} for line in f: if line.strip() and not line.startswith(#): key, value line.split(, 1) config[key.strip()] value.strip() return config except FileNotFoundError: print(f配置文件 {config_file} 未找到。使用默认配置。) return {} CONFIG load_config(r./QQ_mail.txt) # ---------- 配置区域 ---------- IMAP_SERVER CONFIG.get(IMAP_SERVER, imap.qq.com) EMAIL_ACCOUNT CONFIG.get(EMAIL_ACCOUNT) # 你的QQ邮箱 EMAIL_PASSWORD CONFIG.get(EMAIL_PASSWORD) # 替换成16位授权码 CHECK_INTERVAL CONFIG.get(CHECK_INTERVAL, 180) # 检查间隔秒 SMTP_SERVER CONFIG.get(SMTP_SERVER, smtp.qq.com) # ---------------------------3. 解码邮件头部信息def decode_email_header(header_value): 解码邮件头部信息处理中文乱码 if header_value is None: return decoded_parts [] for part, encoding in decode_header(header_value): if isinstance(part, bytes): try: if encoding: decoded_parts.append(part.decode(encoding)) else: decoded_parts.append(part.decode(utf-8, errorsignore)) except: try: decoded_parts.append(part.decode(gbk, errorsignore)) except: decoded_parts.append(part.decode(utf-8, errorsignore)) else: decoded_parts.append(part) return .join(decoded_parts)4. 从发件人字符串中提取邮箱地址def get_sender_email(sender_string): 从发件人字符串中提取邮箱地址 # 匹配邮箱地址的正则表达式 email_pattern r[\w\.-][\w\.-]\.\w match re.search(email_pattern, sender_string) if match: return match.group(0) return sender_string5. 判断哪些邮箱应该自动回复有的邮件是系统邮件可以不用回复def should_auto_reply(sender_email): 判断是否应该自动回复 # 检查黑名单 if sender_email in BLACKLIST: return False # 检查是否是系统邮件 system_keywords [postmaster, mailer-daemon, noreply, no-reply, service] sender_lower sender_email.lower() for keyword in system_keywords: if keyword in sender_lower: return False # 检查是否是自己的邮件 if sender_email EMAIL_ACCOUNT: return False return True6. 提取邮件正文def get_email_body(msg): 提取邮件正文 body try: if msg.is_multipart(): for part in msg.walk(): content_type part.get_content_type() content_disposition str(part.get(Content-Disposition)) # 只提取纯文本内容跳过附件 if content_type text/plain and attachment not in content_disposition: try: payload part.get_payload(decodeTrue) if payload: body payload.decode(utf-8, errorsignore) break except: try: body part.get_payload(decodeTrue).decode(gbk, errorsignore) break except: pass else: # 非 multipart 邮件 payload msg.get_payload(decodeTrue) if payload: try: body payload.decode(utf-8, errorsignore) except: body payload.decode(gbk, errorsignore) except Exception as e: print(f 提取正文时出错: {e}) return body[:500] # 限制长度7. 生成回复内容def generate_reply_content(original_sender, original_subject, original_content): 生成回复内容 reply f您好 感谢您的来信 我是自动回复机器人已收到您的邮件 主题{original_subject} 您的邮件对我们很重要我会尽快人工回复您。 祝好 {EMAIL_ACCOUNT} 自动回复请勿直接回复此邮件 return reply8. 发送回复邮件def send_reply(to_email, subject, content): 发送回复邮件 - 增强错误处理 try: # 创建邮件对象 msg MIMEMultipart() msg[From] EMAIL_ACCOUNT msg[To] to_email msg[Subject] fRe: {subject} # 添加邮件正文 msg.attach(MIMEText(content, plain, utf-8)) # 使用更稳定的SMTP连接方式 server smtplib.SMTP_SSL(SMTP_SERVER, 465, timeout30) # 添加调试信息可选 # server.set_debuglevel(1) # 登录 server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) # 发送邮件 server.send_message(msg) # 关闭连接 server.quit() print(f ✓ 已回复: {to_email}) return True except smtplib.SMTPServerDisconnected as e: print(f ✗ SMTP连接断开: {e}) return False except smtplib.SMTPAuthenticationError as e: print(f ✗ 认证失败请检查授权码: {e}) return False except Exception as e: print(f ✗ 发送失败: {e}) return False def mark_as_replied(mail, email_id): 标记邮件为已处理 try: # 标记为已读 mail.store(email_id, FLAGS, \\Seen) print(f 已标记邮件为已读) except Exception as e: print(f 标记失败: {e})9. 获取未读邮件并逐一回复def process_and_reply(): 主函数获取未读邮件并逐一回复 try: # 连接IMAP服务器 print(\n 正在连接IMAP服务器...) mail imaplib.IMAP4_SSL(IMAP_SERVER, timeout30) mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) mail.select(INBOX) # 搜索未读邮件 status, messages mail.search(None, UNSEEN) if status ! OK: return 0 email_ids messages[0].split() if not email_ids: print( 没有未读邮件) return 0 print(f\n 发现 {len(email_ids)} 封未读邮件) # 过滤掉系统邮件和黑名单 valid_emails [] for email_id in email_ids: status, data mail.fetch(email_id, (BODY.PEEK[HEADER.FIELDS (FROM SUBJECT)])) if status OK: raw_header data[0][1] msg email.message_from_bytes(raw_header) from_raw msg.get(From, ) sender_email get_sender_email(from_raw) if should_auto_reply(sender_email): valid_emails.append(email_id) print(f ✓ 有效邮件: {sender_email}) else: print(f ⏭ 跳过黑名单: {sender_email}) # 标记为已读避免重复处理 mail.store(email_id, FLAGS, \\Seen) if not valid_emails: print( 没有需要回复的有效邮件) return 0 print(f\n 开始回复 {len(valid_emails)} 封有效邮件...) replied_count 0 for email_id in valid_emails: print(f\n--- 处理邮件 ID: {email_id.decode()} ---) # 获取完整邮件内容 status, data mail.fetch(email_id, (RFC822)) if status ! OK: print(f 无法获取邮件内容) continue raw_email data[0][1] msg email.message_from_bytes(raw_email) # 提取邮件信息 from_raw msg.get(From, ) from_addr decode_email_header(from_raw) subject_raw msg.get(Subject, 无主题) subject decode_email_header(subject_raw) sender_email get_sender_email(from_addr) print(f 发件人: {from_addr}) print(f 邮箱: {sender_email}) print(f 主题: {subject}) # 提取邮件正文 body get_email_body(msg) # 生成回复内容 reply_content generate_reply_content(sender_email, subject, body[:200]) # 发送回复 if send_reply(sender_email, subject, reply_content): replied_count 1 # 标记已处理 mark_as_replied(mail, email_id) # 等待一下避免发送过快 time.sleep(2) print(f\n✅ 处理完成共回复 {replied_count} 封邮件) return replied_count except imaplib.IMAP4.error as e: print(f❌ IMAP连接错误: {e}) return 0 except Exception as e: print(f❌ 处理过程中出错: {e}) import traceback traceback.print_exc() return 0 finally: try: mail.close() mail.logout() except: pass10. 定时监控并自动回复在循环内定时执行 process_and_reply函数def auto_reply_loop(): 持续监控并自动回复 print( * 60) print(f 自动回复系统已启动) print(f 监控邮箱: {EMAIL_ACCOUNT}) print(f⏱️ 检查间隔: {CHECK_INTERVAL}秒) print(f 黑名单: {, .join(BLACKLIST)}) print( * 60) while True: try: process_and_reply() print(f\n⏰ 等待 {CHECK_INTERVAL} 秒后继续监控...) time.sleep(CHECK_INTERVAL) except KeyboardInterrupt: print(\n\n 用户中断自动回复系统已停止) break except Exception as e: print(f⚠️ 运行错误: {e}) time.sleep(CHECK_INTERVAL)11. 测试if __name__ __main__: print(load_config(r./QQ_mail.txt) ) print(f开始监控邮箱: {EMAIL_ACCOUNT}) auto_reply_loop()12. 程序运行结果PS D:\workspace\email-automation uv run python monitor.py {IMAP_SERVER: imap.qq.com, EMAIL_ACCOUNT: spz_0911qq.com, EMAIL_PASSWORD: jiugrsrwpmddbadi} 开始监控邮箱: spz_0911qq.com 自动回复系统已启动 监控邮箱: spz_0911qq.com ⏱️ 检查间隔: 180秒 黑名单: PostMasterqq.com, mailer-daemonqq.com, noreplyqq.com, serviceqq.com, spz_0911qq.com 正在连接IMAP服务器... 发现 32 封未读邮件 ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ✓ 有效邮件: campusmailf.wisdomore.com ✓ 有效邮件: campusmailf.wisdomore.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: servicequickjobs.51job.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ⏭ 跳过黑名单: noreplyredditmail.com ✓ 有效邮件: MyClawaisecret.us ⏭ 跳过黑名单: no-replygithub.com ✓ 有效邮件: hicursor.com ✓ 有效邮件: position-trackingsemrush.com ✓ 有效邮件: productmailsend1.jijyun.cn ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com ⏭ 跳过黑名单: PostMasterqq.com 开始回复 6 封有效邮件... --- 处理邮件 ID: 155 --- 发件人: 智联招聘 campusmailf.wisdomore.com 邮箱: campusmailf.wisdomore.com 主题: 尊敬的曾同学【智联推荐】盒马前置仓校招专项来啦 ✓ 已回复: campusmailf.wisdomore.com 已标记邮件为已读 --- 处理邮件 ID: 156 --- 发件人: 智联招聘 campusmailf.wisdomore.com 邮箱: campusmailf.wisdomore.com 主题: 尊敬的曾同学【智联推荐】盒马前置仓校招专项来啦 ✓ 已回复: campusmailf.wisdomore.com 已标记邮件为已读 --- 处理邮件 ID: 198 --- 发件人: MyClaw Newsletter MyClawaisecret.us 邮箱: MyClawaisecret.us 主题: OpenClaw Instances Widely Exposed ✓ 已回复: MyClawaisecret.us 已标记邮件为已读 --- 处理邮件 ID: 274 --- 发件人: Cursor hicursor.com 邮箱: hicursor.com 主题: Here to help ✓ 已回复: hicursor.com 已标记邮件为已读 --- 处理邮件 ID: 278 --- 发件人: Semrush Position Tracking position-trackingsemrush.com 邮箱: position-trackingsemrush.com 主题: Position Tracking Update (originmattress.com.au) ✓ 已回复: position-trackingsemrush.com 已标记邮件为已读 --- 处理邮件 ID: 279 --- 发件人: 集简云 productmailsend1.jijyun.cn 邮箱: productmailsend1.jijyun.cn 主题: 欢迎加入集简云 ✓ 已回复: productmailsend1.jijyun.cn 已标记邮件为已读 ✅ 处理完成共回复 6 封邮件 ⏰ 等待 180 秒后继续监控...完整代码import imaplib import smtplib import time import email from email.header import decode_header from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import re def load_config(config_file): 加载配置文件如果需要 try: with open(config_file, r, encodingutf-8) as f: config {} for line in f: if line.strip() and not line.startswith(#): key, value line.split(, 1) config[key.strip()] value.strip() return config except FileNotFoundError: print(f配置文件 {config_file} 未找到。使用默认配置。) return {} CONFIG load_config(r./QQ_mail.txt) # ---------- 配置区域 ---------- IMAP_SERVER CONFIG.get(IMAP_SERVER, imap.qq.com) EMAIL_ACCOUNT CONFIG.get(EMAIL_ACCOUNT) # 你的QQ邮箱 EMAIL_PASSWORD CONFIG.get(EMAIL_PASSWORD) # 替换成16位授权码 CHECK_INTERVAL CONFIG.get(CHECK_INTERVAL, 180) # 检查间隔秒 SMTP_SERVER CONFIG.get(SMTP_SERVER, smtp.qq.com) # --------------------------- BLACKLIST [ PostMasterqq.com, mailer-daemonqq.com, noreplyqq.com, serviceqq.com, spz_0911qq.com # 自己的邮箱 ] def decode_email_header(header_value): 解码邮件头部信息处理中文乱码 if header_value is None: return decoded_parts [] for part, encoding in decode_header(header_value): if isinstance(part, bytes): try: if encoding: decoded_parts.append(part.decode(encoding)) else: decoded_parts.append(part.decode(utf-8, errorsignore)) except: try: decoded_parts.append(part.decode(gbk, errorsignore)) except: decoded_parts.append(part.decode(utf-8, errorsignore)) else: decoded_parts.append(part) return .join(decoded_parts) def get_sender_email(sender_string): 从发件人字符串中提取邮箱地址 # 匹配邮箱地址的正则表达式 email_pattern r[\w\.-][\w\.-]\.\w match re.search(email_pattern, sender_string) if match: return match.group(0) return sender_string def should_auto_reply(sender_email): 判断是否应该自动回复 # 检查黑名单 if sender_email in BLACKLIST: return False # 检查是否是系统邮件 system_keywords [postmaster, mailer-daemon, noreply, no-reply, service] sender_lower sender_email.lower() for keyword in system_keywords: if keyword in sender_lower: return False # 检查是否是自己的邮件 if sender_email EMAIL_ACCOUNT: return False return True def generate_reply_content(original_sender, original_subject, original_content): 生成回复内容 reply f您好 感谢您的来信 我是自动回复机器人已收到您的邮件 主题{original_subject} 您的邮件对我们很重要我会尽快人工回复您。 祝好 {EMAIL_ACCOUNT} 自动回复请勿直接回复此邮件 return reply def send_reply(to_email, subject, content): 发送回复邮件 - 增强错误处理 try: # 创建邮件对象 msg MIMEMultipart() msg[From] EMAIL_ACCOUNT msg[To] to_email msg[Subject] fRe: {subject} # 添加邮件正文 msg.attach(MIMEText(content, plain, utf-8)) # 使用更稳定的SMTP连接方式 server smtplib.SMTP_SSL(SMTP_SERVER, 465, timeout30) # 添加调试信息可选 # server.set_debuglevel(1) # 登录 server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) # 发送邮件 server.send_message(msg) # 关闭连接 server.quit() print(f ✓ 已回复: {to_email}) return True except smtplib.SMTPServerDisconnected as e: print(f ✗ SMTP连接断开: {e}) return False except smtplib.SMTPAuthenticationError as e: print(f ✗ 认证失败请检查授权码: {e}) return False except Exception as e: print(f ✗ 发送失败: {e}) return False def mark_as_replied(mail, email_id): 标记邮件为已处理 try: # 标记为已读 mail.store(email_id, FLAGS, \\Seen) print(f 已标记邮件为已读) except Exception as e: print(f 标记失败: {e}) def get_email_body(msg): 提取邮件正文 body try: if msg.is_multipart(): for part in msg.walk(): content_type part.get_content_type() content_disposition str(part.get(Content-Disposition)) # 只提取纯文本内容跳过附件 if content_type text/plain and attachment not in content_disposition: try: payload part.get_payload(decodeTrue) if payload: body payload.decode(utf-8, errorsignore) break except: try: body part.get_payload(decodeTrue).decode(gbk, errorsignore) break except: pass else: # 非 multipart 邮件 payload msg.get_payload(decodeTrue) if payload: try: body payload.decode(utf-8, errorsignore) except: body payload.decode(gbk, errorsignore) except Exception as e: print(f 提取正文时出错: {e}) return body[:500] # 限制长度 def process_and_reply(): 主函数获取未读邮件并逐一回复 try: # 连接IMAP服务器 print(\n 正在连接IMAP服务器...) mail imaplib.IMAP4_SSL(IMAP_SERVER, timeout30) mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) mail.select(INBOX) # 搜索未读邮件 status, messages mail.search(None, UNSEEN) if status ! OK: return 0 email_ids messages[0].split() if not email_ids: print( 没有未读邮件) return 0 print(f\n 发现 {len(email_ids)} 封未读邮件) # 过滤掉系统邮件和黑名单 valid_emails [] for email_id in email_ids: status, data mail.fetch(email_id, (BODY.PEEK[HEADER.FIELDS (FROM SUBJECT)])) if status OK: raw_header data[0][1] msg email.message_from_bytes(raw_header) from_raw msg.get(From, ) sender_email get_sender_email(from_raw) if should_auto_reply(sender_email): valid_emails.append(email_id) print(f ✓ 有效邮件: {sender_email}) else: print(f ⏭ 跳过黑名单: {sender_email}) # 标记为已读避免重复处理 mail.store(email_id, FLAGS, \\Seen) if not valid_emails: print( 没有需要回复的有效邮件) return 0 print(f\n 开始回复 {len(valid_emails)} 封有效邮件...) replied_count 0 for email_id in valid_emails: print(f\n--- 处理邮件 ID: {email_id.decode()} ---) # 获取完整邮件内容 status, data mail.fetch(email_id, (RFC822)) if status ! OK: print(f 无法获取邮件内容) continue raw_email data[0][1] msg email.message_from_bytes(raw_email) # 提取邮件信息 from_raw msg.get(From, ) from_addr decode_email_header(from_raw) subject_raw msg.get(Subject, 无主题) subject decode_email_header(subject_raw) sender_email get_sender_email(from_addr) print(f 发件人: {from_addr}) print(f 邮箱: {sender_email}) print(f 主题: {subject}) # 提取邮件正文 body get_email_body(msg) # 生成回复内容 reply_content generate_reply_content(sender_email, subject, body[:200]) # 发送回复 if send_reply(sender_email, subject, reply_content): replied_count 1 # 标记已处理 mark_as_replied(mail, email_id) # 等待一下避免发送过快 time.sleep(2) print(f\n✅ 处理完成共回复 {replied_count} 封邮件) return replied_count except imaplib.IMAP4.error as e: print(f❌ IMAP连接错误: {e}) return 0 except Exception as e: print(f❌ 处理过程中出错: {e}) import traceback traceback.print_exc() return 0 finally: try: mail.close() mail.logout() except: pass def auto_reply_loop(): 持续监控并自动回复 print( * 60) print(f 自动回复系统已启动) print(f 监控邮箱: {EMAIL_ACCOUNT}) print(f⏱️ 检查间隔: {CHECK_INTERVAL}秒) print(f 黑名单: {, .join(BLACKLIST)}) print( * 60) while True: try: process_and_reply() print(f\n⏰ 等待 {CHECK_INTERVAL} 秒后继续监控...) time.sleep(CHECK_INTERVAL) except KeyboardInterrupt: print(\n\n 用户中断自动回复系统已停止) break except Exception as e: print(f⚠️ 运行错误: {e}) time.sleep(CHECK_INTERVAL) if __name__ __main__: print(load_config(r./QQ_mail.txt) ) print(f开始监控邮箱: {EMAIL_ACCOUNT}) auto_reply_loop()