panleicim/appointment_request
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimization on mail_reader_all_contacts.py

Browse Source
This commit is contained in:
Lei PAN
2025-12-01 17:14:56 +01:00
parent a689e66635
commit 0cf94d44fc
4 changed files with 291 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mail/mail_reader_all_contacts.py
Executable → Regular
+278 -157
View File
@@ -2,11 +2,10 @@ import datetime
import email
import logging
import re
from builtins import list
from concurrent.futures import ThreadPoolExecutor
from email.header import decode_header
from email.message import Message
from typing import Union
from typing import Union, List
from imapclient import IMAPClient
from db.mongo_manager import MONGO_STORE_MANAGER
@@ -15,74 +14,87 @@ from mail.mail_constants import DOMAIN_HOTMAIL, create_imap
from models.ReserveResultPojo import ReserveResultPojo
from models.mail_pojo import MailPojo, MailAddress
VALIDATION_URL_SUBJECT_fr = 'Validation de votre demande de rendez-vous'
# 定义常量
VALIDATION_URL_SUBJECT_FR = 'Validation de votre demande de rendez-vous'
VALIDATION_URL_SUBJECT_EN = 'Please confirm your appointment request'
VALIDATION_URL_REGEX = """https:\/\/rendezvousparis.hermes.com\/client\/register\/[A-Z0-9]+\/validate.code=[A-Z0-9]+"""
PART_VALIDATION_URL_REGEX = """client\/register\/[A-Z0-9]+\/validate.code=[A-Z0-9]+"""
VALIDATION_URL_REGEX = r"https:\/\/rendezvousparis.hermes.com\/client\/register\/[A-Z0-9]+\/validate.code=[A-Z0-9]+"
PART_VALIDATION_URL_REGEX = r"client\/register\/[A-Z0-9]+\/validate.code=[A-Z0-9]+"
HERMES_EMAIL = "no-reply@hermes.com"
EMAIL_ADDRESS_REGEX = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
date_format = "%d-%b-%Y" # DD-Mon-YYYY e.g., 3-Mar-2014
# 日期格式
DATE_FORMAT = "%d-%b-%Y"
# 邮箱列表（简化为常量）
REDIRECTION_MAILS = "appointment2022@aol.com, chenpeijun@aol.com,hongjiang176@aol.com,ciyuexie@aol.com,rutger.62@aol.com,ciccidaniel@aol.com,armasgoodman@aol.com,wknd.gemerine@aol.com,rafmail1981@aol.com,tonovichivanenaki@aol.com,hetland.ari@aol.com,mateusiversen@aol.com,lacerdaraffaello@aol.com,anasida76@aol.com,liamolinari@aol.com,sen70zib@aol.com,mezeiderrick@aol.com,stanisl49avchic@aol.com,damcvrobaneuron@aol.com,suyzanna_fleona@aol.com,dxealing.dissa@aol.com,hogg.karen@aol.com,obocharovamarina@aol.com,buchholzjohann@aol.com,orn.cecchini@aol.com,percivaltorgersen@aol.com,candalgudrun@aol.com,filimonis.76@aol.com,bengann_100@aol.com,axelhanne@aol.com,tiffanylarochelle@aol.com,nicoleta.r@aol.com,eichenbaum.1963@aol.com,kotensasharev@aol.com,samognat32@aol.com,edem_headshot@aol.com,kozmakuzmich1960@aol.com,damonsvensson@aol.com,anders.riva@aol.com,caiminwei123@gmail.com,yulingguo086@gmail.com,yingxiaolu086@gmail.com,lijiazhen0035@gmail.com,fangp370@gmail.com,huangyayu10086@gmail.com,fuziyuan110@gmail.com,xinyingdu886@gmail.com,yasiaforever.1971@aol.com,lukaszfidalgo@aol.com,zaichi29@aol.com,prostotakitak.1974@aol.com,mo90nroe@aol.com,blonde.87@aol.com,dimidrol.1969@aol.com"
def check_email_address(email):
# pass the regular expression
# and the string into the fullmatch() method
if (re.fullmatch(EMAIL_ADDRESS_REGEX, email)):
print("Valid Email")
return True
else:
print("Invalid Email:" + email)
return False
# 邮件处理相关函数
def is_valid_email(email: str) -> bool:
"""验证邮箱地址是否有效"""
return re.fullmatch(EMAIL_ADDRESS_REGEX, email) is not None
def find_from_mail(param):
def extract_email_from_from_address(content: str) -> str:
"""从邮件地址中提取邮箱"""
match = re.search(r'[\w.+-]+@[\w-]+\.[\w.-]+', content)
return match.group(0) if match else ""
def find_from_mail(param) -> str:
"""解析邮件地址"""
from_address, encoded_algo = param[0]
# 处理字节编码
if isinstance(from_address, bytes):
from_address = from_address.decode(encoded_algo)
if not check_email_address(from_address) and len(param) == 2:
# 如果邮箱地址无效，尝试另一种编码
if not is_valid_email(from_address) and len(param) == 2:
from_address, new_encode = param[1]
if new_encode is None:
new_encode = encoded_algo
if isinstance(from_address, bytes):
from_address = from_address.decode(new_encode)
return from_address.strip(" ").strip(">").strip("<")
# 清理邮箱地址
return from_address.strip(" ").strip(">").strip("<")
def extract_email_from_from_address(content: str):
_match = re.search(r'[\w.+-]+@[\w-]+\.[\w.-]+', content)
return _match.group(0)
class MailReader:
"""邮件读取器类"""
class MailReader():
def __init__(self, login, password):
def __init__(self, login: str, password: str):
self.login = login
self.password = password
@staticmethod
def show_folders(imap) -> list:
def show_folders(imap) -> List[str]:
"""获取邮箱文件夹列表"""
folders = []
isImapClient = isinstance(imap, IMAPClient)
if not isImapClient:
is_imap_client = isinstance(imap, IMAPClient)
if not is_imap_client:
# 处理非IMAPClient对象
for i in imap.list()[1]:
l = i.decode().split(' "/" ')
folders.append(l[1])
return folders
else:
list = imap.list_folders()
for i in list:
# 处理IMAPClient对象
folder_list = imap.list_folders()
for i in folder_list:
name = i[-1]
folders.append(name)
return folders
def read_emails(self, mails_messages: list) -> list:
return folders
def read_emails(self, mails_messages: List[MailPojo]) -> List[MailPojo]:
"""读取邮件"""
imap = create_imap(self.login)
isImapClient = isinstance(imap, IMAPClient)
print("isImapClient is " + str(isImapClient))
if isImapClient:
# authenticate
is_imap_client = isinstance(imap, IMAPClient)
# 登录邮箱
if is_imap_client:
dat = imap.login(self.login, str(self.password))
print("type is {} for {}".format(dat, self.login))
else:
@@ -91,224 +103,333 @@ class MailReader():
mail_list = []
print("read mails from {}".format(self.login))
if not isImapClient:
folder_list = self.show_folders(imap)
for folder in folder_list:
print("folder is {}".format(folder))
mail_list.extend(self._get_messages_from_folder(imap, subject=VALIDATION_URL_SUBJECT_fr,
folder=folder))
mail_list.extend(self._get_messages_from_folder(imap, subject=VALIDATION_URL_SUBJECT_EN,
folder=folder))
else:
folder_list = self.show_folders(imap)
for folder in folder_list:
print("folder is " + folder)
if folder == "Sent" or folder == "Drafts":
pass
else:
mail_list.extend(self._get_messages_from_folder_for_imapclient(imap, folder=folder))
if not isImapClient:
# 获取文件夹列表
folder_list = self.show_folders(imap)
# 处理每个文件夹
for folder in folder_list:
print("folder is {}".format(folder))
# 跳过Sent和Drafts文件夹
if folder in ["Sent", "Drafts"]:
continue
if is_imap_client:
# 使用IMAPClient处理
mail_list.extend(self._get_messages_from_folder_for_imapclient(imap, folder))
else:
# 使用传统IMAP处理
mail_list.extend(self._get_messages_from_folder(imap, subject=VALIDATION_URL_SUBJECT_FR, folder=folder))
mail_list.extend(self._get_messages_from_folder(imap, subject=VALIDATION_URL_SUBJECT_EN, folder=folder))
# 关闭连接
if not is_imap_client:
imap.close()
imap.logout()
# 添加邮件到结果列表
mails_messages.extend(mail_list)
return mail_list
def _get_messages_from_folder(self, imap, subject, folder="INBOX") -> list:
def _get_messages_from_folder(self, imap, subject: str, folder: str = "INBOX") -> List[MailPojo]:
"""从指定文件夹获取邮件（传统IMAP方式）"""
imap.select(folder)
mail_messages = []
typ, data = imap.search(None, '(SUBJECT "{}" SINCE "{}")'.format(subject,
datetime.datetime.today().strftime(
date_format)))
# 搜索邮件
search_query = '(SUBJECT "{}" SINCE "{}")'.format(subject, datetime.datetime.today().strftime(DATE_FORMAT))
typ, data = imap.search(None, search_query)
for i in data[0].split():
# fetch the email message by ID
res, msg = imap.fetch(i.decode("utf-8"), "(RFC822)")
body = ''
for response in msg:
if isinstance(response, tuple):
# parse a bytes email into a message object
msg = email.message_from_bytes(response[1])
# decode the email subject
subject, subject_encoded = decode_header(msg["Subject"])[0]
received_date = msg["Date"]
if isinstance(subject, bytes):
# if it's a bytes, decode to str
subject = subject.decode(subject_encoded)
# decode email sender
from_address = find_from_mail(decode_header(msg.get("From")))
to_email = find_from_mail(decode_header(msg.get("To")))
print("Email:", self.login)
print("From:", from_address)
print("To:", to_email)
print("Subject:", subject)
# if the email message is multipart
if msg.is_multipart():
# iterate over email parts
for part in msg.walk():
try:
# get the email body
payloads = part.get_payload()
if isinstance(payloads, list):
for payload in payloads:
if isinstance(payload, Message):
body = body + payload.get_payload(decode=True).decode("iso-8859-1")
# print(body)
except Exception as Error:
print(Error)
else:
body = msg.get_payload(decode=True).decode()
print(body)
if VALIDATION_URL_SUBJECT_fr in subject or VALIDATION_URL_SUBJECT_EN in subject:
mail = MailPojo(subject=subject, body=body, from_address=from_address)
if to_email is None:
mail.to_address = self.login
else:
mail.to_address = to_email
mail.mail_address = self.login
mail_messages.append(mail)
try:
# 获取邮件内容
res, msg = imap.fetch(i.decode("utf-8"), "(RFC822)")
# 解析邮件
for response in msg:
if isinstance(response, tuple):
email_message = email.message_from_bytes(response[1])
# 解码主题
subject, subject_encoded = decode_header(email_message["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(subject_encoded)
# 解码发件人地址
from_address = find_from_mail(decode_header(email_message.get("From")))
# 解码收件人地址
to_email = find_from_mail(decode_header(email_message.get("To")))
print("Email:", self.login)
print("From:", from_address)
print("To:", to_email)
print("Subject:", subject)
# 获取邮件正文
body = self._extract_body(email_message)
# 检查是否是预约验证邮件
if VALIDATION_URL_SUBJECT_FR in subject or VALIDATION_URL_SUBJECT_EN in subject:
mail = MailPojo(
subject=subject,
body=body,
from_address=from_address
)
# 设置收件人地址
if to_email is None:
mail.to_address = self.login
else:
mail.to_address = to_email
mail.mail_address = self.login
mail_messages.append(mail)
except Exception as error:
print("Error processing email: {}".format(error))
return mail_messages
def _get_messages_from_folder_for_imapclient(self, imap, folder="INBOX") -> list:
def _extract_body(self, email_message: Message) -> str:
"""提取邮件正文"""
body = ""
# 遍历邮件部分
for part in email_message.walk():
try:
content_type = part.get_content_type()
if content_type == "text/html":
# 处理HTML内容
payload = part.get_payload(decode=True)
if payload:
body += payload.decode("utf-8", errors="ignore")
elif content_type == "text/plain":
# 处理纯文本内容
payload = part.get_payload()
if payload:
body += payload
except Exception as error:
print("Error extracting body part: {}".format(error))
return body
def _get_messages_from_folder_for_imapclient(self, imap, folder: str = "INBOX") -> List[MailPojo]:
"""从指定文件夹获取邮件（IMAPClient方式）"""
mail_messages = []
# 搜索邮件
search_terms = 'SINCE "{}"'.format(
datetime.datetime.today().strftime(
date_format))
datetime.datetime.today().strftime(DATE_FORMAT))
print("{}: search terms is {}".format(self.login, search_terms))
imap.select_folder(folder)
messages = imap.search(['SINCE', datetime.datetime.today()])
print("{}: {} messages from our best friend".format(self.login, len(messages)))
if len(messages) == 0:
return mail_messages
# 处理每封邮件
for uid, message_data in imap.fetch(messages, 'RFC822').items():
try:
email_message = email.message_from_bytes(message_data[b'RFC822'])
# 获取发件人和主题
from_address = email_message.get('FROM')
subject = email_message.get('subject')
body = ""
hermes_mail_adress = "no-reply@hermes.com"
if hermes_mail_adress in from_address or "outlook.com" in from_address or "hotmail" in from_address:
for part in email_message.walk():
print(part.get_content_type())
if part.get_content_type() == "text/html":
body = body + part.get_payload(decode=True).decode("utf-8")
elif part.get_content_type() == "text/plain":
body = body + part.get_payload()
if VALIDATION_URL_SUBJECT_fr in subject or VALIDATION_URL_SUBJECT_EN in subject or "Votre=20demande=20de=20rendez-vous" in subject or "Votre demande de rendez-vous" in body:
mail = MailPojo(subject=subject, body=body, from_address=from_address)
# 检查是否是Hermes邮件
hermes_mail_address = "no-reply@hermes.com"
if (hermes_mail_address in from_address or
"outlook.com" in from_address or
"hotmail" in from_address):
# 提取邮件正文
body = self._extract_body_for_imapclient(email_message)
# 检查是否是预约验证邮件
if (VALIDATION_URL_SUBJECT_FR in subject or
VALIDATION_URL_SUBJECT_EN in subject or
"Votre=20demande=20de=20rendez-vous" in subject or
"Votre demande de rendez-vous" in body):
mail = MailPojo(
subject=subject,
body=body,
from_address=from_address
)
mail.isImapClient = True
print("email is {}".format(self.login))
print("body is {}".format(body))
print("subject is {}".format(subject))
# 设置收件人地址
if len(mail.to_address) == 0:
if "outlook.com" in from_address or "hotmail.com" in from_address:
# it is a transferred email
# 转发邮件
mail.to_address = extract_email_from_from_address(from_address)
else:
mail.to_address = self.login
mail_messages.append(mail)
except Exception as error:
print(error)
print("error trying to read email_Message for {}".format(self.login))
print("Error trying to read email_Message for {}: {}".format(self.login, error))
return mail_messages
def _extract_body_for_imapclient(self, email_message: Message) -> str:
"""提取IMAPClient邮件正文"""
body = ""
#
# Find the ReserveResultPojo object from persisted items of DB
#
for part in email_message.walk():
content_type = part.get_content_type()
if content_type == "text/html":
payload = part.get_payload(decode=True)
if payload:
body += payload.decode("utf-8", errors="ignore")
elif content_type == "text/plain":
payload = part.get_payload()
if payload:
body += payload
return body
# 邮件处理相关函数
def find_item_by_url(url: str, successful_items) -> Union[None, ReserveResultPojo]:
"""根据URL查找预约结果对象"""
print("url is :" + url)
parts = url.split('/')
_id = parts[5]
if len(_id) == 6:
for item in successful_items:
if item.id == _id:
return item
return None
def need_to_valid_url(url: str, item: Union[ReserveResultPojo, None]) -> bool:
"""判断是否需要验证URL"""
print("url is :" + url)
parts = url.split('/')
id = parts[5]
if len(id) == 6:
_id = parts[5]
if len(_id) == 6:
if item:
if item.url_validated is not None:
return not item.url_validated
else:
# if url_validated is None
# 如果url_validated为None，需要验证
return True
return True
else:
print("id not valid:{}".format(id))
return False
print("id not valid:{}".format(_id))
return False
def need_to_check_email(mail: str, successful_items) -> bool:
"""判断是否需要检查邮件"""
print("successful_items size is " + str(len(successful_items)))
# 特殊处理
if mail == "saigecong1990@pissmail.com":
return True
filtered_items = list(filter(lambda item: item.email == mail, successful_items))
# has validated value
if len(filtered_items) > 0:
validated_items = list(filter(
lambda filtered_item: filtered_item.url_validated is not None and filtered_item.url_validated is True,
filtered_items))
if len(validated_items) > 0:
return False
else:
return True
else:
return True
# 过滤已验证的项目
filtered_items = [item for item in successful_items if item.email == mail]
# 检查是否有已验证的项目
validated_items = [item for item in filtered_items
if item.url_validated is not None and item.url_validated is True]
return len(validated_items) == 0
def find_links_to_validate_from_mail_list(mail_list: list, logger):
def find_links_to_validate_from_mail_list(mail_list: List[MailAddress], logger) -> None:
"""从邮件列表中查找需要验证的链接"""
if not mail_list:
return
# check time before start checking emails
# 检查时间前开始检查邮件
contact_to_book_list = MONGO_STORE_MANAGER.get_all_contact_to_book_list()
successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
mails_messages = []
with ThreadPoolExecutor(max_workers=200) as executor:
# 使用线程池处理邮件
with ThreadPoolExecutor(max_workers=20) as executor:
futures = []
for mail in mail_list:
# check whether we need to read mail
# 检查是否需要读取邮件
if need_to_check_email(mail.mail, successful_items):
mail_reader = MailReader(mail.mail, mail.password)
executor.submit(mail_reader.read_emails, mails_messages)
future = executor.submit(mail_reader.read_emails, mails_messages)
futures.append(future)
# 等待所有任务完成
for future in futures:
try:
future.result()
except Exception as e:
print("Error processing mail: {}".format(e))
# 刷新成功的项目
_refreshed_successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
# 处理邮件中的链接
for mail in mails_messages:
match = re.search(VALIDATION_URL_REGEX, mail.body)
if match:
url = match.group(0)
_item = find_item_by_url(url, _refreshed_successful_items)
if need_to_valid_url(url, _item):
logger.info("need to validate url: " + url)
_model = ""
_used_ip = ""
if _item:
_model = _item.model
MONGO_STORE_MANAGER.save_links_to_validate(url, mail.to_address, model=_model,
_all_contact_list=contact_to_book_list)
_used_ip = _item.current_ip
MONGO_STORE_MANAGER.save_links_to_validate(
url,
mail.to_address,
model=_model,
_all_contact_list=contact_to_book_list, _used_ip= _used_ip)
else:
logger.info("do not need to click url --> {}".format(mail.mail_address))
# 主函数
if __name__ == '__main__':
# mail_address1 = MailAddress(mail="tinagonzales685585@aol.com", password="yhihvdkrbxnksema")
# mail_list = [mail_address1]
# 读取联系人列表
contact_to_book_list = read_contacts(
# file_name="/Users/rdv/Desktop/contact_list_not_used_contacts.xlsx")
# file_name="/Users/lpan/Desktop/contact_list_not_used_contacts.xlsx")
# file_name="/Users/rdv/Desktop/contact_list_2025-10-30.xlsx")
# file_name="~/Desktop/contact_list_all.xlsx")
file_name="~/Desktop/contact_list_2025-11-06.xlsx")
# file_name="/Users/rdv/Desktop/contact_list_all_studo_gmx_us.xlsx")
# file_name="/Users/rdv/Desktop/contact_list_2025-05-24.xlsx")
# 获取目标邮箱列表
all_mail_list = MONGO_STORE_MANAGER.get_destination_emails()
# 筛选需要检查的邮件列表
mail_list_to_check = []
for contact in contact_to_book_list:
for mail in all_mail_list:
if contact.mail == mail.mail:
mail_list_to_check.append(mail)
# 设置日志记录器
logger = logging.getLogger()
# 获取已验证的链接列表
_all_links = MONGO_STORE_MANAGER.get_links_to_validate()
# 过滤掉已处理的邮件
filter_mail = []
for mail_pojo in mail_list_to_check:
_to_add = True
@@ -317,6 +438,6 @@ if __name__ == '__main__':
_to_add = False
if _to_add:
filter_mail.append(mail_pojo)
# filter_mail.append(MailAddress("saigecong1990@pissmail.com", "cvExXKOP8oY1D@"))
# filter_mail = [MailAddress("saigecong1990@pissmail.com", "cvExXKOP8oY1D@")]
# filter_mail = [MailAddress("utatapi@gmx.net", "RSAzHAFek8s")]
# 处理邮件
find_links_to_validate_from_mail_list(filter_mail, logger)