142 lines
5.8 KiB
Python
142 lines
5.8 KiB
Python
import email
|
|
import imaplib
|
|
import re
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from email.header import decode_header
|
|
from email.message import Message
|
|
|
|
from builtins import list
|
|
|
|
from src import params
|
|
from src.db.mongo_manager import MONGO_STORE_MANAGER
|
|
from src.pojo.mail.mail_pojo import MailPojo, MailAddress
|
|
from src.proxy.proxy_type import ProxyType
|
|
from src.workers.link_validator import LinkValidator
|
|
|
|
AOL_IMAP_SERVER = "imap.aol.com"
|
|
VALIDATION_URL_SUBJECT = 'Validation de votre demande de rendez-vous'
|
|
VALIDATION_URL_REGEX = """https:\/\/rendezvousparis.hermes.com\/client\/register\/[A-Z0-9]+\/validate.code=[A-Z0-9]+"""
|
|
|
|
|
|
class MailReader():
|
|
def __init__(self, login, password):
|
|
self.login = login
|
|
self.password = password
|
|
|
|
def read_emails(self, email_number=0) -> list:
|
|
# create an IMAP4 class with SSL
|
|
mail_list = []
|
|
imap = imaplib.IMAP4_SSL(AOL_IMAP_SERVER)
|
|
# authenticate
|
|
imap.login(self.login, self.password)
|
|
status, messages = imap.select("INBOX")
|
|
# total number of emails
|
|
messages = int(messages[0])
|
|
for i in range(messages, 0, -1):
|
|
# fetch the email message by ID
|
|
res, msg = imap.fetch(str(i), "(RFC822)")
|
|
# res, msg = imap.fetch(str(i))
|
|
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, subject_encoded = decode_header(msg.get("From"))[0]
|
|
if isinstance(from_address, bytes):
|
|
from_address = from_address.decode(subject_encoded)
|
|
print("From:", from_address)
|
|
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 in subject:
|
|
mail = MailPojo(subject=subject, body=body, from_address=from_address)
|
|
mail_list.append(mail)
|
|
# close the connection and logout
|
|
imap.close()
|
|
imap.logout()
|
|
return mail_list
|
|
|
|
|
|
hermes_email = "no-reply@hermes.com"
|
|
# account credentials
|
|
username = "appointment2022@aol.com"
|
|
password = "gyilpmvyyvlcaviq"
|
|
|
|
|
|
def clean(text):
|
|
# clean text for creating a folder
|
|
return "".join(c if c.isalnum() else "_" for c in text)
|
|
|
|
|
|
def need_to_valid_url(url: str, successful_items) -> bool:
|
|
print("url is :" + url)
|
|
parts = url.split('/')
|
|
id = parts[5]
|
|
if len(id) == 6:
|
|
for item in successful_items:
|
|
if item.url_validated is not None:
|
|
print("id:{}, status:{} ".format(id, str(item.url_validated)))
|
|
if item.id == id:
|
|
if item.url_validated is not None:
|
|
return not item.url_validated
|
|
else:
|
|
# if url_validated is None
|
|
return True
|
|
# return True by default
|
|
return False
|
|
else:
|
|
print("id not valid:{}".format(id))
|
|
return False
|
|
|
|
|
|
def read_mails():
|
|
mail_address1 = MailAddress(mail="appointment2022@aol.com", password="gyilpmvyyvlcaviq")
|
|
mail_address2 = MailAddress(mail="chenpeijun@aol.com", password="ytifuwguknzifqyb")
|
|
mail_address3 = MailAddress(mail="ciyuexie@aol.com", password="czezlmmyypokdfce")
|
|
mail_address4 = MailAddress(mail="hongjiang176@aol.com", password="ftzpscgzvwneelmn")
|
|
mail_list = [mail_address3, mail_address2, mail_address1, mail_address4]
|
|
# mail_list = [mail_address3]
|
|
for mail in mail_list:
|
|
mail_reader = MailReader(mail.mail, mail.password)
|
|
successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
|
|
mail_info_list = mail_reader.read_emails()
|
|
with ThreadPoolExecutor(max_workers=10) as executor:
|
|
for mail in mail_info_list:
|
|
match = re.search(VALIDATION_URL_REGEX, mail.body)
|
|
if match:
|
|
url = match.group(0)
|
|
if need_to_valid_url(url, successful_items):
|
|
url_validator = LinkValidator(url)
|
|
print("need to validate url: " + url)
|
|
# .start_page(params.get_proxy(ProxyType.BRIGHT_DATA))
|
|
executor.submit(url_validator.start_page, params.get_proxy(ProxyType.BRIGHT_DATA), True)
|
|
else:
|
|
print("do not need to click url --> {}".format(mail))
|
|
|
|
|
|
# check whether the url has already been clicked
|
|
if __name__ == '__main__':
|
|
read_mails()
|