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

use imapClient

Browse Source
This commit is contained in:
PAN Lei
2024-12-05 23:31:14 +01:00
parent f1cf40198d
commit d0d3a30377
3 changed files with 60 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mail/mail_confirmation.py
+21 -12
View File
@@ -1,5 +1,7 @@
import datetime
import email
import logging
import sys
from builtins import list
from concurrent.futures import ThreadPoolExecutor
from email.header import decode_header
@@ -8,6 +10,7 @@ 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.logs.AppLogging import init_logger
from src.mail.mail_constants import create_imap, show_folders
from src.notification.AcceptedResultPojo import get_accepted_result_from
from src.notification.SignalSender import SignalSender
@@ -78,12 +81,12 @@ class MailConfirmationReader():
body = body + str(part.get_payload(decode=True).decode("utf-8"))
elif part.get_content_type() == "text/plain":
body = body + part.get_payload()
print("mail is {} and subject is {}".format(self.login, subject))
logger.info("mail is {} and subject is {}".format(self.login, subject))
if CONFIRMATION_SUBJECT_FR in subject or CONFIRMATION_SUBJECT_EN in subject or "Votre_rendez-vous_est_confirm" in subject:
mail = MailPojo(subject=subject, body=body, from_address=from_address)
mail.isImapClient = True
print("subject is {}".format(subject))
print("body is {}".format(body))
logger.info("subject is {}".format(subject))
logger.info("body is {}".format(body))
mail_messages.append(mail)
except Exception as error:
print(error)
@@ -93,9 +96,12 @@ class MailConfirmationReader():
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)))
# typ, data = imap.search(None, '(SUBJECT "{}" SINCE "{}")'.format(subject,
# datetime.datetime.today().strftime(
# date_format)))
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)")
@@ -114,8 +120,8 @@ class MailConfirmationReader():
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)
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
@@ -132,7 +138,7 @@ class MailConfirmationReader():
print(Error)
else:
body = msg.get_payload(decode=True).decode()
print(body)
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
@@ -156,7 +162,7 @@ def accept_appointment_found(accepted_result_list: list):
for user in _all_register_account:
if user.mail == result.email:
result.account_password = user.password
mailer.send_email(result, to_all=False)
mailer.send_email(result, to_all=True)
MONGO_STORE_MANAGER.update_reserve_result(reserve.id, ResultEnum.ACCEPTED, reserve.message)
# sginal.send_result(result)
@@ -173,8 +179,8 @@ def find_confirmation_contacts_for_today():
for _item in _all_appointments_today:
for _mail in _all_mail_list:
if _mail.mail == _item.mail:
if _item.url_validated is True:
_mail_list_to_scan.append(_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
@@ -234,6 +240,9 @@ def read_mails_and_find_confirmation_contacts(all_mails=False):
# mail_list = [mail_address3]
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()
src/mail/mail_constants.py
+36 -18
View File
@@ -86,47 +86,65 @@ def create_imap(login: str):
if DOMAIN_163 in login:
imap = IMAPClient(IMAP_SERVER_163, use_uid=True)
elif DOMAIN_YAHOO in login:
imap = imaplib.IMAP4_SSL(YAHOO_IMAP_SERVER)
# imap = imaplib.IMAP4_SSL(YAHOO_IMAP_SERVER)
imap = IMAPClient(YAHOO_IMAP_SERVER, use_uid=True)
elif DOMAIN_FIREMAIL_DE in login:
# imap = imaplib.IMAP4_SSL(SERVER_FIREMAIL_DE)
imap = IMAPClient(SERVER_FIREMAIL_DE, use_uid=True)
elif DOMAIN_GMX in login or DOMAIN_GMX_FR in login or DOMAIN_GMX_US in login or DOMAIN_GMX_CH in login or DOMAIN_GMX_PT in login or DOMAIN_GMX_SG in login:
imap = imaplib.IMAP4_SSL(SERVER_GMX)
# imap = imaplib.IMAP4_SSL(SERVER_GMX)
imap = IMAPClient(SERVER_GMX, use_uid=True)
elif DOMAIN_SINA in login:
# imap = imaplib.IMAP4_SSL(IMAP_SERVER_SINA)
imap = IMAPClient(IMAP_SERVER_SINA, use_uid=True)
elif DOMAIN_HOTMAIL in login or DOMAIN_OUTLOOK_COM in login:
imap = imaplib.IMAP4_SSL(HOTMAIL_IMAP_SERVER)
# imap = imaplib.IMAP4_SSL(HOTMAIL_IMAP_SERVER)
imap = IMAPClient(HOTMAIL_IMAP_SERVER, use_uid=True)
elif DOMAIN_RAMBLER_RU in login:
imap = imaplib.IMAP4_SSL(RAMBLER_IMAP_SERVER)
# imap = imaplib.IMAP4_SSL(RAMBLER_IMAP_SERVER)
imap = IMAPClient(RAMBLER_IMAP_SERVER, use_uid=True)
elif DOMAN_BTVM_NE_JP in login:
imap = imaplib.IMAP4_SSL(BTVM_NE_JP)
# imap = imaplib.IMAP4_SSL(BTVM_NE_JP)
imap = IMAPClient(BTVM_NE_JP, use_uid=True)
elif DOMAN_GMAIL in login:
imap = imaplib.IMAP4_SSL(SEREVER_GMAIL, port=993)
# imap = imaplib.IMAP4_SSL(SEREVER_GMAIL, port=993)
imap = IMAPClient(SEREVER_GMAIL, use_uid=True)
elif DOMAIN_ONET in login:
imap = IMAPClient(SERVER_IMAGE_ONET, use_uid=True)
elif DOMAIN_TIM_IT in login:
imap = imaplib.IMAP4(TIME_IT_SERVER)
# imap = imaplib.IMAP4(TIME_IT_SERVER)
imap = IMAPClient(TIME_IT_SERVER, use_uid=True)
elif DOMAIN_ALICE_IT in login:
imap = imaplib.IMAP4(ALICE_IMAP_SERVER, port=143)
# imap = imaplib.IMAP4(ALICE_IMAP_SERVER, port=143)
imap = IMAPClient(ALICE_IMAP_SERVER, use_uid=True)
elif DOMAIN_MARS_DTI_NE_JP in login:
imap = imaplib.IMAP4(MARS_DTI_NE_JP_SERVER, port=143)
# imap = imaplib.IMAP4(MARS_DTI_NE_JP_SERVER, port=143)
imap = IMAPClient(MARS_DTI_NE_JP_SERVER, use_uid=True)
elif DOMAN_AURORA_DTI_NE_JP in login:
imap = imaplib.IMAP4(MARS_DTI_NE_JP_SERVER, port=143)
# imap = imaplib.IMAP4(MARS_DTI_NE_JP_SERVER, port=143)
imap = IMAPClient(MARS_DTI_NE_JP_SERVER, use_uid=True)
elif DOMAIN_NAVER in login:
imap = imaplib.IMAP4_SSL(NAVER_SERVER, port=993)
# imap = imaplib.IMAP4_SSL(NAVER_SERVER, port=993)
imap = IMAPClient(NAVER_SERVER, use_uid=True)
elif DOMAIN_GMX_DE in login or DOMAIN_GMX_NET in login:
imap = imaplib.IMAP4_SSL(SERVER_GMX_NET, port=993)
# imap = imaplib.IMAP4_SSL(SERVER_GMX_NET, port=993)
imap = IMAPClient(SERVER_GMX_NET, use_uid=True)
elif DOMAIN_GMX_AT in login:
imap = imaplib.IMAP4_SSL(SERVER_GMX_AT, port=993)
# imap = imaplib.IMAP4_SSL(SERVER_GMX_AT, port=993)
imap = IMAPClient(SERVER_GMX_AT, use_uid=True)
elif DOMAIN_GAZETA_PL in login:
imap = imaplib.IMAP4_SSL(IMAP_SERVER_DOMAIN_GAZETA_PL, port=993)
# imap = imaplib.IMAP4_SSL(IMAP_SERVER_DOMAIN_GAZETA_PL, port=993)
imap = IMAPClient(IMAP_SERVER_DOMAIN_GAZETA_PL, use_uid=True)
elif DOMAIN_INBOX_LV in login:
imap = imaplib.IMAP4_SSL(INBOX_LV, port=993)
# imap = imaplib.IMAP4_SSL(INBOX_LV, port=993)
imap = IMAPClient(INBOX_LV, use_uid=True)
elif DOMAIN_WEB_DE in login:
imap = imaplib.IMAP4_SSL(SERVER_WEB_DE, port=993)
# imap = imaplib.IMAP4_SSL(SERVER_WEB_DE, port=993)
imap = IMAPClient(SERVER_WEB_DE, use_uid=True)
elif DOMAIN_PISS_MAIL in login or DOMAIN_CHILD_PIZZA in login or DOMAIN_DMC_CHAT in login or DOMAIN_GENOCIDE_FUN in login or DOMAIN_HATESJE_WS in login or DOMAIN_INCEL_EMAIL in login or DOMAIN_SHITPOSTING_EXPERT in login:
imap = imaplib.IMAP4_SSL(SERVER_PISS_MAIL, port=993)
# imap = imaplib.IMAP4_SSL(SERVER_PISS_MAIL, port=993)
imap = IMAPClient(SERVER_PISS_MAIL, use_uid=True)
else:
imap = imaplib.IMAP4_SSL(AOL_IMAP_SERVER)
# imap = imaplib.IMAP4_SSL(AOL_IMAP_SERVER)
imap = IMAPClient(AOL_IMAP_SERVER, use_uid=True)
return imap
src/mail/mail_reader_all_contacts.py
+3 -9
View File
@@ -13,10 +13,9 @@ from src.db.mongo_manager import MONGO_STORE_MANAGER
from src.logs.AppLogging import init_logger
from src.mail.mail_constants import DOMAIN_HOTMAIL, create_imap
from src.pojo.mail.mail_pojo import MailPojo
from src.utils import excel_reader
from src.utils.timeutiles import is_time_between
VALIDATION_URL_SUBJECT_fr = 'Validation de votre demande de rendez-vous'
VALIDATION_URL_SUBJECT_fr = '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]+"""
@@ -186,7 +185,7 @@ class MailReader():
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 "Validation=20de=20votre=20demande=20de=20rendez-vous" in subject:
if VALIDATION_URL_SUBJECT_fr in subject or VALIDATION_URL_SUBJECT_EN in subject or "Votre=20demande=20de=20rendez-vous" in subject:
mail = MailPojo(subject=subject, body=body, from_address=from_address)
mail.isImapClient = True
print("email is {}".format(self.login))
@@ -256,7 +255,7 @@ def need_to_check_email(mail: str, successful_items) -> bool:
def read_all_mails(contact_to_book_list=None):
# check time before start checking emails
if is_time_between(time(7, 30), time(19, 30)):
if is_time_between(time(7, 30), time(23, 50)):
# get email address
if contact_to_book_list is None:
contact_to_book_list = MONGO_STORE_MANAGER.get_all_contact_to_book_list()
@@ -284,9 +283,6 @@ def read_all_mails(contact_to_book_list=None):
_refreshed_successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
with ThreadPoolExecutor(max_workers=10) as executor:
for mail in mails_messages:
# if mail.isImapClient:
# match = re.search(PART_VALIDATION_URL_REGEX, mail.body.replace("\n", ""))
# else:
match = re.search(VALIDATION_URL_REGEX, mail.body)
if match:
# url_to_validate = match.group(0)
@@ -298,8 +294,6 @@ def read_all_mails(contact_to_book_list=None):
print("need to validate url: " + url)
MONGO_STORE_MANAGER.save_links_to_validate(url, mail.to_address,
_all_contact_list=contact_to_book_list)
# url_validator = LinkValidator(url)
# executor.submit(url_validator.start_page, params.get_proxy(ProxyType.OXYLABS), False)
else:
print("do not need to click url --> {}".format(mail.mail_address))