import datetime import email import logging import sys from builtins import list from concurrent.futures import ThreadPoolExecutor from email.header import decode_header from email.message import Message from imapclient import IMAPClient from src.db.mirgration.migration_tools import migre_accepted_appointment from src.db.mongo_manager import MONGO_STORE_MANAGER from src.mail.mail_constants import create_imap, show_folders, is_gmx_address from src.mail.mail_reader import get_gmx_proxy_config from src.mail.imap_proxy_reader import ProxyMailReader, MailAccount, ProxyConfig from src.notification.AcceptedResultPojo import get_accepted_result_from from src.notification.mailer import Mailer from src.pojo.ResultEnum import ResultEnum from src.pojo.mail.mail_pojo import MailPojo, MailAddress CONFIRMATION_SUBJECT_FR = 'Votre=20rendez-vous=20est=20confirm=C3' CONFIRMATION_SUBJECT_EN = 'confirmed' HERMES_EMAIL = "no-reply@hermes.com" date_format = "%d-%b-%Y" # DD-Mon-YYYY e.g., 3-Mar-2014 FRENCH_CONFIRMED_MESSAGE = "Nous aurons le plaisir de vous accueillir" def read_gmx_proxy_confirmation_emails(mail, mails_messages: list, proxy_config: ProxyConfig) -> None: account = MailAccount(login=mail.mail, password=mail.password) results = ProxyMailReader(account, proxy_config).read(since=datetime.datetime.today()) for result in results: mail_pojo = MailPojo(subject=result.subject, body=result.body, from_address=result.from_address) mail_pojo.mail_address = mail.mail mail_pojo.to_address = result.to_address or mail.mail mail_pojo.isImapClient = True mails_messages.append(mail_pojo) class MailConfirmationReader(): def __init__(self, login, password): self.login = login self.password = password def read_emails(self, mails_messages: list) -> list: imap = create_imap(self.login) isImapClient = isinstance(imap, IMAPClient) # authenticate 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 isImapClient: mail_list.extend(self._get_messages_from_folder_for_imapclient(imap)) else: folderList = show_folders(imap) for folder in folderList: mail_list.extend(self._get_messages_from_folder(imap, CONFIRMATION_SUBJECT_FR, folder=folder)) mail_list.extend(self._get_messages_from_folder(imap, CONFIRMATION_SUBJECT_EN, folder=folder)) if not isImapClient: imap.close() imap.logout() mails_messages.extend(mail_list) return mail_list def _get_messages_from_folder_for_imapclient(self, imap, folder="INBOX") -> list: mail_messages = [] search_terms = 'SINCE "{}"'.format( datetime.datetime.today().strftime( date_format)) imap.select_folder(folder) messages = imap.search(['SINCE', datetime.datetime.today()]) 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 = "" # if "no-reply@hermes.com" in from_address: for part in email_message.walk(): print(part.get_content_type()) if part.get_content_type() == "text/html": body = body + str(part.get_payload(decode=True).decode("utf-8")) elif part.get_content_type() == "text/plain": body = body + part.get_payload() logger.info("mail is {} and subject is {}, body is {}".format(self.login, subject, body)) if CONFIRMATION_SUBJECT_FR in subject or CONFIRMATION_SUBJECT_EN in subject or "Votre_rendez-vous_est_confirm" in subject or "Votre rendez-vous est confirm" in body: mail = MailPojo(subject=subject, body=body, from_address=from_address) mail.isImapClient = True mail.mail_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 def _get_messages_from_folder(self, imap, subject, folder="INBOX") -> list: imap.select(folder) mail_messages = [] typ, data = imap.search(None, '(SINCE "{}")'.format( 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, subject_encoded = decode_header(msg.get("From"))[0] if isinstance(from_address, bytes): from_address = from_address.decode(subject_encoded) logger.info("From:{}".format(from_address)) logger.info("Subject: {}".format(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() logger.info(body) if CONFIRMATION_SUBJECT_FR in subject or CONFIRMATION_SUBJECT_EN in subject: mail = MailPojo(subject=subject, body=body, from_address=from_address) mail.mail_address = self.login mail_messages.append(mail) return mail_messages def clean(text): # clean text for creating a folder return "".join(c if c.isalnum() else "_" for c in text) def accept_appointment_found(accepted_result_list: list): _all_contact_list = MONGO_STORE_MANAGER.get_all_contact_to_book_list() _all_register_account = MONGO_STORE_MANAGER.get_all_registered_users() mailer = Mailer() # sginal = SignalSender() print(accepted_result_list) for reserve in accepted_result_list: result = get_accepted_result_from(reserve, MONGO_STORE_MANAGER, _all_contact_list) for user in _all_register_account: if user.mail == result.email: result.account_password = user.password mailer.send_email(result, to_all=False) MONGO_STORE_MANAGER.update_reserve_result(reserve.id, ResultEnum.ACCEPTED, reserve.message) # sginal.send_result(result) if len(accepted_result_list) > 0: migre_accepted_appointment(str(datetime.date.today())) def find_confirmation_contacts_for_today(): _all_mail_list = MONGO_STORE_MANAGER.get_destination_emails() _all_appointments_today = MONGO_STORE_MANAGER.get_all_successful_items_for_day() if len(_all_appointments_today) == 0: _all_appointments_today = MONGO_STORE_MANAGER.get_all_successful_items_for_yesterday() _mail_list_to_scan = [] for _item in _all_appointments_today: for _mail in _all_mail_list: if _mail.mail == _item.mail: # do not need to scan outlook if "outlook.com" not in _mail.mail: # if _item.url_validated is True: _mail_list_to_scan.append(_mail) break print("Found {} emails to scan".format(len(_mail_list_to_scan))) return _mail_list_to_scan def find_confirmation_contacts_mail_list(mail_list): mail_list.append(MailAddress("saigecong1990@pissmail.com", "cvExXKOP8oY1D@")) mails_messages = [] gmx_proxy_config = get_gmx_proxy_config() # read all the emails with ThreadPoolExecutor(max_workers=200) as executor: for mail in mail_list: if is_gmx_address(mail.mail) and gmx_proxy_config is not None: executor.submit(read_gmx_proxy_confirmation_emails, mail, mails_messages, gmx_proxy_config) else: mail_reader = MailConfirmationReader(mail.mail, mail.password) executor.submit(mail_reader.read_emails, mails_messages) accepted_appointment_list = [] if len(mails_messages) > 0: successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day() # check the hours current_hour = datetime.datetime.now().hour if current_hour < 15: # add yesterday's appointment only for morning successful_items.extend(MONGO_STORE_MANAGER.get_all_successful_items_for_yesterday()) for mail in mails_messages: message_body = mail.body for item in successful_items: if item.id in message_body and item.id != "welcome" and len(item.id) > 0: # 如果邮件中包含有ID item.message = message_body accepted_appointment_list.append(item) elif "10:30" in message_body and ( item.mail == mail.mail_address or item.mail in message_body) and len(item.id) > 0: item.message = message_body accepted_appointment_list.append(item) elif FRENCH_CONFIRMED_MESSAGE in message_body and ( item.mail == mail.mail_address or item.mail in message_body) and len(item.id) > 0: item.message = message_body accepted_appointment_list.append(item) elif "11:30" in message_body and ( item.mail == mail.mail_address or item.mail in message_body) and len(item.id) > 0: item.message = message_body accepted_appointment_list.append(item) print(mail.mail_address) print(mail.subject) print(mail.body) accept_appointment_found(accepted_appointment_list) return accepted_appointment_list def read_mails_and_find_confirmation_contacts(all_mails=False): if all_mails: mail_list = MONGO_STORE_MANAGER.get_destination_emails() else: mail_list = find_confirmation_contacts_for_today() return find_confirmation_contacts_mail_list(mail_list) # init_logger() logger = logging.getLogger() logger.addHandler(logging.StreamHandler(stream=sys.stdout)) # check whether the url has already been clicked if __name__ == '__main__': # read_mails_and_find_confirmation_contacts() _mail_list_today = find_confirmation_contacts_for_today() # print("size is {}".format(len(_mail_list_today))) find_confirmation_contacts_mail_list(_mail_list_today) # _items = MONGO_STORE_MANAGER.get_all_successful_items_for_day() # accept_appointment_found([random.choice(_items)])