203 lines
6.6 KiB
Python
203 lines
6.6 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
测试IMAP邮件获取功能
|
||
单独测试邮箱取件功能
|
||
"""
|
||
|
||
import imaplib
|
||
import email
|
||
import email.header
|
||
import sys
|
||
from email.parser import BytesParser
|
||
from email.policy import default
|
||
import re
|
||
import time
|
||
from urllib.parse import urlparse, parse_qs
|
||
|
||
# 邮箱配置
|
||
EMAIL_ADDRESS = "5ulxb79e@evnmail.com"
|
||
EMAIL_PASSWORD = "dr!mueh15ahG"
|
||
IMAP_SERVER = "mail.evnmail.com"
|
||
USE_SSL = True
|
||
|
||
def decode_mime_words(text):
|
||
"""解码MIME编码的文本"""
|
||
if not text:
|
||
return ""
|
||
try:
|
||
decoded = email.header.decode_header(text)
|
||
return " ".join([
|
||
(str(t[0], t[1] or 'utf-8') if isinstance(t[0], bytes) else str(t[0]))
|
||
for t in decoded
|
||
])
|
||
except:
|
||
return text
|
||
|
||
def get_body_text(msg):
|
||
"""获取邮件正文文本内容"""
|
||
if msg.is_multipart():
|
||
# 如果邮件包含多个部分，递归处理
|
||
text_parts = []
|
||
for part in msg.get_payload():
|
||
if part.get_content_type() == 'text/plain':
|
||
charset = part.get_content_charset() or 'utf-8'
|
||
try:
|
||
text_parts.append(part.get_payload(decode=True).decode(charset))
|
||
except:
|
||
text_parts.append("[无法解码的内容]")
|
||
elif part.is_multipart():
|
||
# 递归处理多部分
|
||
text_parts.append(get_body_text(part))
|
||
return "\n".join(text_parts)
|
||
else:
|
||
# 单部分邮件
|
||
content_type = msg.get_content_type()
|
||
if content_type == 'text/plain':
|
||
charset = msg.get_content_charset() or 'utf-8'
|
||
try:
|
||
return msg.get_payload(decode=True).decode(charset)
|
||
except:
|
||
return "[无法解码的内容]"
|
||
else:
|
||
return f"[内容类型: {content_type}]"
|
||
|
||
def extract_urls_from_body(body, pattern):
|
||
"""从邮件正文中提取URL"""
|
||
found_urls = re.findall(pattern, body)
|
||
return [url.replace('&', '&').replace("=3D", "=").replace("=\r\n", "")
|
||
.replace("\r\n", "").replace("=\n", "").replace("\n", "")
|
||
for url in found_urls]
|
||
|
||
def test_imap_connection():
|
||
"""测试IMAP连接"""
|
||
print(f"正在测试连接到 {IMAP_SERVER}...")
|
||
try:
|
||
# 连接到IMAP服务器
|
||
if USE_SSL:
|
||
mail = imaplib.IMAP4_SSL(IMAP_SERVER)
|
||
else:
|
||
mail = imaplib.IMAP4(IMAP_SERVER)
|
||
|
||
# 登录
|
||
print(f"正在登录邮箱 {EMAIL_ADDRESS}...")
|
||
mail.login(EMAIL_ADDRESS, EMAIL_PASSWORD)
|
||
print("登录成功!")
|
||
|
||
# 关闭连接
|
||
mail.logout()
|
||
return True
|
||
except Exception as e:
|
||
print(f"连接失败: {str(e)}")
|
||
return False
|
||
|
||
def fetch_emails():
|
||
"""获取邮件"""
|
||
urls = []
|
||
try:
|
||
# 连接到IMAP服务器
|
||
if USE_SSL:
|
||
mail = imaplib.IMAP4_SSL(IMAP_SERVER)
|
||
else:
|
||
mail = imaplib.IMAP4(IMAP_SERVER)
|
||
|
||
# 登录
|
||
print(f"正在登录邮箱 {EMAIL_ADDRESS}...")
|
||
mail.login(EMAIL_ADDRESS, EMAIL_PASSWORD)
|
||
print("登录成功!")
|
||
|
||
# 只检查INBOX文件夹
|
||
steam_verification_pattern = r'https://store\.steampowered\.com/account/newaccountverification\?[^\s"\'<>]+'
|
||
|
||
try:
|
||
print("\n检查文件夹: INBOX")
|
||
mail.select("INBOX")
|
||
status, messages = mail.search(None, "ALL")
|
||
if status != "OK" or not messages[0]:
|
||
print(" INBOX为空或无法访问")
|
||
else:
|
||
# 获取邮件ID列表，从最新的开始
|
||
message_ids = messages[0].split()
|
||
message_count = len(message_ids)
|
||
print(f" 找到 {message_count} 封邮件")
|
||
|
||
# 最多检查10封最新邮件
|
||
check_count = min(10, message_count)
|
||
message_ids = message_ids[-check_count:]
|
||
message_ids.reverse() # 最新的邮件放在前面
|
||
|
||
for i, msg_id in enumerate(message_ids):
|
||
print(f"\n 正在检查邮件 {i+1}/{check_count}, ID: {msg_id.decode()}...")
|
||
status, msg_data = mail.fetch(msg_id, "(RFC822)")
|
||
if status != "OK":
|
||
print(" 无法获取邮件内容")
|
||
continue
|
||
|
||
# 解析邮件内容
|
||
raw_email = msg_data[0][1]
|
||
msg = email.message_from_bytes(raw_email, policy=default)
|
||
|
||
# 提取信息
|
||
subject = decode_mime_words(msg["Subject"])
|
||
from_addr = decode_mime_words(msg["From"])
|
||
date = decode_mime_words(msg["Date"])
|
||
|
||
print(f" 主题: {subject}")
|
||
print(f" 发件人: {from_addr}")
|
||
print(f" 日期: {date}")
|
||
|
||
# 获取正文
|
||
body = get_body_text(msg)
|
||
body_preview = body[:100] + "..." if len(body) > 100 else body
|
||
print(f" 正文预览: {body_preview}")
|
||
|
||
# 查找Steam验证链接
|
||
found_urls = extract_urls_from_body(body, steam_verification_pattern)
|
||
if found_urls:
|
||
print(f" 发现 {len(found_urls)} 个可能的Steam验证链接:")
|
||
for url in found_urls:
|
||
print(f" {url[:80]}...")
|
||
urls.append(url)
|
||
else:
|
||
print(" 未发现Steam验证链接")
|
||
except Exception as e:
|
||
print(f"处理INBOX出错: {str(e)}")
|
||
|
||
# 尝试关闭连接
|
||
try:
|
||
mail.close()
|
||
except:
|
||
pass
|
||
mail.logout()
|
||
|
||
except Exception as e:
|
||
print(f"IMAP连接错误: {str(e)}")
|
||
|
||
return urls
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("IMAP邮件获取测试")
|
||
print("=" * 60)
|
||
|
||
# 测试连接
|
||
if not test_imap_connection():
|
||
print("IMAP连接测试失败，程序退出")
|
||
return 1
|
||
|
||
print("\n开始获取邮件...")
|
||
urls = fetch_emails()
|
||
|
||
if urls:
|
||
print("\n" + "=" * 60)
|
||
print(f"共找到 {len(urls)} 个可能的Steam验证链接:")
|
||
for i, url in enumerate(urls):
|
||
print(f"{i+1}. {url}")
|
||
else:
|
||
print("\n没有找到任何Steam验证链接")
|
||
|
||
return 0
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |