parent
b4b6835010
commit
a786dca470
120
mail.py
120
mail.py
|
|
@ -20,7 +20,6 @@ def to_utf7_imap(text):
|
|||
"""
|
||||
if not text:
|
||||
return ""
|
||||
|
||||
# 匹配所有非 ASCII 字符
|
||||
def encode_match(match):
|
||||
return "&" + match.group(0).encode("utf-16be").hex().upper() + "-"
|
||||
|
|
@ -28,7 +27,6 @@ def to_utf7_imap(text):
|
|||
# 替换非 ASCII 字符为 UTF-7 格式
|
||||
return re.sub(r"[^\x20-\x7E]", encode_match, text)
|
||||
|
||||
|
||||
def build_search_criteria(from_email=None, subject=None, body_keyword=None, since=None, before=None, unseen=False):
|
||||
"""
|
||||
构建 IMAP 搜索条件字符串，支持简单查询逻辑。
|
||||
|
|
@ -67,7 +65,6 @@ def build_search_criteria(from_email=None, subject=None, body_keyword=None, sinc
|
|||
# 用空格拼接所有条件（IMAP 默认 AND 逻辑）
|
||||
return " ".join(criteria) if criteria else "ALL"
|
||||
|
||||
|
||||
def parse_email_date(date_str, default_tz=timezone.utc):
|
||||
"""
|
||||
安全解析邮件日期。
|
||||
|
|
@ -90,12 +87,12 @@ class EmailClient:
|
|||
self.username = username
|
||||
self.password = password
|
||||
self.connection = None
|
||||
|
||||
|
||||
def connect(self):
|
||||
self.connection = imaplib.IMAP4(self.host)
|
||||
self.connection.login(self.username, self.password)
|
||||
self.connection.select("inbox")
|
||||
|
||||
|
||||
def disconnect(self):
|
||||
if self.connection:
|
||||
self.connection.logout()
|
||||
|
|
@ -118,16 +115,15 @@ class EmailClient:
|
|||
result, data = self.connection.search(None, search_criteria)
|
||||
if result != "OK":
|
||||
raise Exception(f"Failed to search emails with criteria: {search_criteria}")
|
||||
|
||||
|
||||
email_ids = data[0].split()
|
||||
if not email_ids:
|
||||
print("Debug: No matching emails found.")
|
||||
return result, [] # 返回空列表以便后续处理
|
||||
|
||||
|
||||
return result, data
|
||||
|
||||
def fetch_recent_emails(self, from_email=None, subject=None, body_keyword=None, since=None, before=None,
|
||||
unseen=False, max_count=100):
|
||||
|
||||
def fetch_recent_emails(self, from_email=None, subject=None, body_keyword=None, since=None, before=None, unseen=False, max_count=100):
|
||||
"""
|
||||
使用构建的搜索条件查询最近的邮件。
|
||||
|
||||
|
|
@ -162,8 +158,9 @@ class EmailClient:
|
|||
email_ids = data[0].split()[-max_count:]
|
||||
return list(reversed(email_ids))
|
||||
|
||||
def fetch_all_matching_emails(self, email_ids, sender_pattern=None, keyword_pattern=None, subject_pattern=None,
|
||||
start_time=None, max_results=None):
|
||||
|
||||
|
||||
def fetch_all_matching_emails(self, email_ids, sender_pattern=None, keyword_pattern=None, subject_pattern=None, start_time=None, max_results=None):
|
||||
if max_results is not None and max_results <= 0:
|
||||
raise ValueError("max_results must be a positive integer or None")
|
||||
|
||||
|
|
@ -181,42 +178,47 @@ class EmailClient:
|
|||
|
||||
msg = email.message_from_bytes(data[0][1])
|
||||
|
||||
# 检查发件人地址是否匹配
|
||||
from_email = email.utils.parseaddr(msg["From"])[1] if msg["From"] else ""
|
||||
if sender_regex and not sender_regex.search(from_email):
|
||||
continue
|
||||
|
||||
# 检查标题是否匹配
|
||||
subject, encoding = decode_header(msg["Subject"])[0]
|
||||
subject = subject.decode(encoding or "utf-8") if isinstance(subject, bytes) else subject
|
||||
if subject_regex and not subject_regex.search(subject):
|
||||
continue
|
||||
|
||||
print("1没报错")
|
||||
|
||||
# 处理邮件内容
|
||||
content = ""
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type()
|
||||
print("2没报错")
|
||||
try:
|
||||
if content_type == "text/plain":
|
||||
content += part.get_payload(decode=True).decode(part.get_content_charset() or "utf-8")
|
||||
print("3没报错")
|
||||
elif content_type == "text/html" and not content:
|
||||
content += part.get_payload(decode=True).decode(part.get_content_charset() or "utf-8")
|
||||
print("4没报错")
|
||||
except Exception as e:
|
||||
print(f"Warning: Failed to decode {content_type} content: {e}")
|
||||
payload = part.get_payload(decode=True)
|
||||
|
||||
# 增加对 payload 类型的判断
|
||||
if isinstance(payload, bytes):
|
||||
if content_type == "text/plain":
|
||||
content += payload.decode(part.get_content_charset() or "utf-8")
|
||||
elif content_type == "text/html" and not content:
|
||||
content += payload.decode(part.get_content_charset() or "utf-8")
|
||||
else:
|
||||
print(f"Skipping unexpected payload type: {type(payload)}, content_type: {content_type}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error processing MIME part: {e}")
|
||||
|
||||
# 检查内容关键字是否匹配
|
||||
if keyword_regex and not keyword_regex.search(content):
|
||||
continue
|
||||
|
||||
|
||||
# 检查邮件日期是否在范围内
|
||||
date_str = msg.get("Date")
|
||||
email_date = parse_email_date(date_str) if date_str else None
|
||||
# print('email_date:',email_date)
|
||||
# print('start_time:',start_time)
|
||||
if start_time and email_date and email_date < start_time:
|
||||
continue
|
||||
|
||||
# 保存匹配的邮件
|
||||
matched_email = {
|
||||
"subject": subject,
|
||||
"from": from_email,
|
||||
|
|
@ -225,6 +227,7 @@ class EmailClient:
|
|||
}
|
||||
all_matched_emails.append(matched_email)
|
||||
|
||||
# 如果达到了最大匹配数，停止处理
|
||||
if max_results and len(all_matched_emails) >= max_results:
|
||||
break
|
||||
|
||||
|
|
@ -233,8 +236,10 @@ class EmailClient:
|
|||
|
||||
return all_matched_emails
|
||||
|
||||
def filter_emails_by_sender_and_keyword(self, email_ids, sender_pattern=None, keyword_pattern=None,
|
||||
subject_pattern=None, start_time=None):
|
||||
|
||||
|
||||
|
||||
def filter_emails_by_sender_and_keyword(self, email_ids, sender_pattern=None, keyword_pattern=None, subject_pattern=None, start_time=None):
|
||||
matched_emails = self.fetch_all_matching_emails(
|
||||
email_ids=email_ids,
|
||||
sender_pattern=sender_pattern,
|
||||
|
|
@ -298,7 +303,6 @@ class GoogleCodeReceiver:
|
|||
body_keyword=username,
|
||||
since=since # 从 `start_time` 往前推 2 天开始检索
|
||||
)
|
||||
|
||||
matched_email = self.email_client.filter_emails_by_sender_and_keyword(
|
||||
email_ids=email_ids,
|
||||
# sender_pattern=sender_pattern,
|
||||
|
|
@ -306,7 +310,7 @@ class GoogleCodeReceiver:
|
|||
subject_pattern=subject_pattern,
|
||||
start_time=start_time # 确保只检索 `start_time` 之后的邮件
|
||||
)
|
||||
|
||||
|
||||
if matched_email:
|
||||
match = re.search(subject_pattern, matched_email["subject"])
|
||||
if match:
|
||||
|
|
@ -320,14 +324,16 @@ class GoogleCodeReceiver:
|
|||
raise TimeoutError("Timeout waiting for verification code.")
|
||||
|
||||
|
||||
|
||||
|
||||
# 老的通用邮箱测试
|
||||
def mailTest():
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
username = "gmailvinted@mailezu.com"
|
||||
password = "g1l2o0hld84"
|
||||
password = "g1l2o0hld84"
|
||||
client = EmailClient(server, username, password)
|
||||
client.connect()
|
||||
start_time = datetime(2024, 11, 22)
|
||||
start_time = datetime(2024, 11, 22)
|
||||
sender_pattern = r".*google.*" # 使用正则表达式匹配发件人邮箱
|
||||
keyword_pattern = r".*" # 替换为你想要匹配的关键字或正则表达式
|
||||
try:
|
||||
|
|
@ -338,7 +344,7 @@ def mailTest():
|
|||
# subject='Email verification code',#中文邮件叫 '电子邮件验证码‘
|
||||
body_keyword='RibeAchour875@gmail.com',
|
||||
since=start_time,
|
||||
)
|
||||
)
|
||||
|
||||
# 获取时间上最新的匹配项，应用起始时间过滤器
|
||||
latest_matched_email = client.filter_emails_by_sender_and_keyword(email_ids, sender_pattern, keyword_pattern)
|
||||
|
|
@ -347,7 +353,7 @@ def mailTest():
|
|||
print("主题:", latest_matched_email["subject"])
|
||||
print("发件人:", latest_matched_email["from"])
|
||||
print("内容:", latest_matched_email["content"])
|
||||
print("时间:", latest_matched_email["date"])
|
||||
print("时间:",latest_matched_email["date"])
|
||||
else:
|
||||
print("没有符合条件的时间上最新的匹配邮件")
|
||||
|
||||
|
|
@ -357,57 +363,53 @@ def mailTest():
|
|||
if all_matched_emails:
|
||||
print("\n所有匹配的邮件:")
|
||||
for idx, email in enumerate(all_matched_emails):
|
||||
print(f"邮件 {idx + 1}:")
|
||||
print(f"邮件 {idx+1}:")
|
||||
print("主题:", email["subject"])
|
||||
print("发件人:", email["from"])
|
||||
print("内容:", email["content"], "\n") # 显示内容
|
||||
print("时间:", email["date"])
|
||||
print("时间:",email["date"])
|
||||
else:
|
||||
print("没有符合条件的所有匹配邮件")
|
||||
|
||||
finally:
|
||||
client.disconnect()
|
||||
|
||||
|
||||
def codeTest():
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
username = "gmailvinted@mailezu.com"
|
||||
password = "g1l2o0hld84"
|
||||
password = "g1l2o0hld84"
|
||||
|
||||
client = EmailClient(server, username, password)
|
||||
client.connect()
|
||||
try:
|
||||
code_receiver = GoogleCodeReceiver(client)
|
||||
# 这里改成要捕获的目标邮件地址
|
||||
code = code_receiver.wait_code(
|
||||
username="RibeAchour875@gmail.com", timeout=300, interval=5,
|
||||
start_time=datetime(2024, 11, 10))
|
||||
# code = code_receiver.wait_code(username="RibeAchour875@gmail.com", timeout=300, interval=5)
|
||||
code_receiver = GoogleCodeReceiver(client)
|
||||
|
||||
code = ''
|
||||
try:
|
||||
code = code_receiver.wait_code(
|
||||
username="RibeAchour875@gmail.com", timeout=2, interval=1,
|
||||
start_time=datetime(2024, 11, 10))
|
||||
print(f"收到谷歌验证码: {code}")
|
||||
except TimeoutError as e:
|
||||
print(e)
|
||||
except Exception as e:
|
||||
print(f"An unexpected error occurred: {e}")
|
||||
finally:
|
||||
client.disconnect()
|
||||
|
||||
|
||||
if not code:
|
||||
print("并没有收到验证码")
|
||||
|
||||
def test3():
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
server = "server-10474.cuiqiu.vip" # 替换为你的 IMAP 服务器地址
|
||||
username = "gmailvinted@mailezu.com"
|
||||
password = "g1l2o0hld84"
|
||||
password = "g1l2o0hld84"
|
||||
|
||||
# server = "imap.qq.com"
|
||||
# username = "bigemon@foxmail.com"
|
||||
# password = "ejudkkdfiuemcaaj"
|
||||
# password = "ejudkkdfiuemcaaj"
|
||||
|
||||
client = EmailClient(server, username, password)
|
||||
client.connect()
|
||||
|
||||
# mailTest()
|
||||
|
||||
ok, email_ids = client.search_emails('FROM "noreply@google.com" BODY "RibeAchour875@gmail.com" SINCE 11-Nov-2024')
|
||||
ok,email_ids = client.search_emails('FROM "noreply@google.com" BODY "RibeAchour875@gmail.com" SINCE 11-Nov-2024')
|
||||
if email_ids:
|
||||
print(email_ids)
|
||||
sender_pattern = r".*google.*" # 使用正则表达式匹配发件人邮箱
|
||||
|
|
@ -416,17 +418,15 @@ def test3():
|
|||
if all_matched_emails:
|
||||
print("\n所有匹配的邮件:")
|
||||
for idx, email in enumerate(all_matched_emails):
|
||||
print(f"邮件 {idx + 1}:")
|
||||
print(f"邮件 {idx+1}:")
|
||||
print("主题:", email["subject"])
|
||||
print("发件人:", email["from"])
|
||||
print("内容:", email["content"], "\n") # 显示内容
|
||||
print("时间:", email["date"])
|
||||
print("时间:",email["date"])
|
||||
else:
|
||||
print("没有符合条件的所有匹配邮件")
|
||||
else:
|
||||
print("查不到")
|
||||
|
||||
|
||||
# 使用示例
|
||||
if __name__ == "__main__":
|
||||
# test3()
|
||||
|
|
|
|||
23
main.py
23
main.py
|
|
@ -583,7 +583,7 @@ def save_log(random_number, log_content):
|
|||
f.write("\n")
|
||||
|
||||
|
||||
def retry_with_recovery(tab, hwnd, email_account, MAX_RETRIES=4, wait_time=15):
|
||||
def retry_with_recovery(tab, hwnd, email_account, MAX_RETRIES=4, wait_time=5):
|
||||
"""
|
||||
封装重试逻辑，最多尝试 `MAX_RETRIES` 次，检查是否存在重试按钮，并重新输入邮箱地址。
|
||||
|
||||
|
|
@ -650,10 +650,7 @@ def logutGoogle(tab) -> bool:
|
|||
|
||||
|
||||
def main(email_account, email_password, old_recovery_email, new_password, new_recovery_email, proxy_host, proxy_port, proxy_username, proxy_password,region, row_index):
|
||||
# 初始化
|
||||
client = EmailClient(server, username, password)
|
||||
client.connect()
|
||||
code_receiver = GoogleCodeReceiver(client)
|
||||
|
||||
|
||||
print("邮箱账户:", email_account) # 邮箱账户
|
||||
print("邮箱密码:", email_password) # 邮箱密码
|
||||
|
|
@ -832,7 +829,7 @@ def main(email_account, email_password, old_recovery_email, new_password, new_re
|
|||
print("输入邮箱账号运行完毕")
|
||||
save_log(random_number, "输入邮箱账号运行完毕")
|
||||
|
||||
wrong = tab.ele('@text()=出了点问题', timeout=15)
|
||||
wrong = tab.ele('@text()=出了点问题', timeout=5)
|
||||
if wrong:
|
||||
next_span = tab.ele("@@tag()=span@@text()=下一步", timeout=15)
|
||||
next_span.click(by_js=True)
|
||||
|
|
@ -858,7 +855,7 @@ def main(email_account, email_password, old_recovery_email, new_password, new_re
|
|||
save_log(random_number, "检查是否出现了wrong运行完毕")
|
||||
|
||||
# 看下是否出现了手机号
|
||||
telephone = tab.ele("@text()=请输入电话号码，以便通过短信接收验证码。", timeout=15)
|
||||
telephone = tab.ele("@text()=请输入电话号码，以便通过短信接收验证码。", timeout=5)
|
||||
if telephone:
|
||||
update_status_in_db(email_account, '接码')
|
||||
browser.quit()
|
||||
|
|
@ -926,7 +923,7 @@ def main(email_account, email_password, old_recovery_email, new_password, new_re
|
|||
|
||||
|
||||
# 看下是否出现了手机号
|
||||
telephone = tab.ele("@text()=请输入电话号码，以便通过短信接收验证码。", timeout=15)
|
||||
telephone = tab.ele("@text()=请输入电话号码，以便通过短信接收验证码。", timeout=5)
|
||||
if telephone:
|
||||
update_status_in_db(email_account, '接码')
|
||||
browser.quit()
|
||||
|
|
@ -1091,17 +1088,23 @@ def main(email_account, email_password, old_recovery_email, new_password, new_re
|
|||
update_status_in_db(email_account, '已更改')
|
||||
return email_account
|
||||
else:
|
||||
# 初始化
|
||||
client = EmailClient(server, username, password)
|
||||
client.connect()
|
||||
code_receiver = GoogleCodeReceiver(client)
|
||||
|
||||
# 最多尝试3次循环
|
||||
for attempt in range(3):
|
||||
print(f"第 {attempt + 1} 次尝试检查验证码...")
|
||||
|
||||
# 获取当前程序运行的时间
|
||||
current_time = datetime.now(timezone.utc)
|
||||
print(f"当前时间为{current_time}")
|
||||
|
||||
# 检查验证码，超时时间为300秒，使用当前时间作为 start_time
|
||||
try:
|
||||
code = code_receiver.wait_code(username=f"{email_account}", timeout=300, interval=5,
|
||||
start_time=current_time)
|
||||
start_time=datetime(2024, 11, 27))
|
||||
print(f"检查的验证码是{code}")
|
||||
save_log(random_number, f"检查的验证码是{code}")
|
||||
if code:
|
||||
|
|
@ -1150,8 +1153,6 @@ def main(email_account, email_password, old_recovery_email, new_password, new_re
|
|||
sent_code_button = tab.ele('@text():Send a new code.', timeout=15)
|
||||
if sent_code_button:
|
||||
sent_code_button.click(by_js=True)
|
||||
|
||||
|
||||
except ElementNotFoundError as e:
|
||||
print(f"更改辅助邮箱账号的时候找不到元素：{e}")
|
||||
save_log(random_number, f"更改辅助邮箱账号的时候找不到元素：{e}")
|
||||
|
|
|
|||
Loading…
Reference in New Issue