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 imapclient import IMAPClient from db.mongo_manager import MONGO_STORE_MANAGER from excel_reader import read_contacts 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_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]+""" 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 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 find_from_mail(param): 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: 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(): def __init__(self, login, password): self.login = login self.password = password @staticmethod def show_folders(imap) -> list: folders = [] isImapClient = isinstance(imap, IMAPClient) if not isImapClient: 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: name = i[-1] folders.append(name) return folders def read_emails(self, mails_messages: list) -> list: imap = create_imap(self.login) isImapClient = isinstance(imap, IMAPClient) print("isImapClient is " + str(isImapClient)) if isImapClient: # authenticate dat = imap.login(self.login, str(self.password)) print("type is {} for {}".format(dat, self.login)) else: responseType, dat = imap.login(self.login, str(self.password)) print("type is {} for {}".format(responseType, self.login)) 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: imap.close() imap.logout() mails_messages.extend(mail_list) return mail_list def _get_messages_from_folder(self, imap, subject, folder="INBOX") -> list: imap.select(folder) mail_messages = [] typ, data = imap.search(None, '(SUBJECT "{}" SINCE "{}")'.format(subject, datetime.datetime.today().strftime( date_format))) 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) return mail_messages def _get_messages_from_folder_for_imapclient(self, imap, folder="INBOX") -> list: mail_messages = [] search_terms = 'SINCE "{}"'.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) 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)) return mail_messages # # Find the ReserveResultPojo object from persisted items of DB # def find_item_by_url(url: str, successful_items) -> Union[None, ReserveResultPojo]: 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: print("url is :" + url) parts = url.split('/') 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 return True return True else: 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 def find_links_to_validate_from_mail_list(mail_list: list, logger): 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: 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) _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 = "" 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) 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 for _link in _all_links: if _link.email == mail_pojo.mail: _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@")] find_links_to_validate_from_mail_list(filter_mail, logger)