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

can filter mail by subject

Browse Source
This commit is contained in:
PAN Lei
2022-08-06 20:22:24 +02:00
parent d707bd56e0
commit fc331aa39b
5 changed files with 124 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/db/mongo_manager.py
+20
View File
@@ -16,6 +16,7 @@ CAPTCHA_ERROR_COLLECTION_PREFIX = "CAPTCHA_ERROR_"
BLACK_LIST = "BLACK_LIST"
ACCEPTED_APPOINTMENT_LIST = "ACCEPTED_APPOINTMENT_LIST"
EMAIL_LIST = "EMAIL_LIST"
DESTINATION_EMAIL_LIST = "DESTINATION_EMAIL_LIST"
class MongoDbManager:
@@ -36,6 +37,14 @@ class MongoDbManager:
except Exception as Error:
self.logger.info(Error)
def save_destinary_emails(self, email: MailAddress):
try:
collection_to_use = self.db[DESTINATION_EMAIL_LIST]
collection_to_use.replace_one(filter={'_id': email.mail, }, replacement=email.to_firestore_dict(),
upsert=True)
except Exception as Error:
self.logger.info(Error)
def insert_email(self, reserve: MailAddress):
try:
collection_to_use = self.db[EMAIL_LIST]
@@ -62,6 +71,17 @@ class MongoDbManager:
self.logger.info(error)
return appointment_list_contacts
def get_destination_emails(self) -> list:
collection_name = DESTINATION_EMAIL_LIST
email_list = []
try:
collection_to_use = self.db[collection_name]
for document in collection_to_use.find():
email_list.append(MailAddress.from_firestore_dict(document))
except Exception as error:
self.logger.info(error)
return email_list
def insert_captcha_error_contact(self, contact: ContactPojo):
day = str(datetime.date.today())
collection_name = CAPTCHA_ERROR_COLLECTION_PREFIX + day
src/mail/mail_reader.py
+40 -24
View File
@@ -16,6 +16,7 @@ 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]+"""
HERMES_EMAIL = "no-reply@hermes.com"
class MailReader():
@@ -23,19 +24,41 @@ class MailReader():
self.login = login
self.password = password
@staticmethod
def show_folders(imap):
for i in imap.list()[1]:
l = i.decode().split(' "/" ')
print(l[0] + " = " + l[1])
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")
mail_list =[]
self.show_folders(imap)
# total number of emails
messages = int(messages[0])
for i in range(messages, 0, -1):
# get mails from inbox
# (\Archive \HasNoChildren) = "Archive"
# (\Junk \HasNoChildren) = "Bulk"
# (\Drafts \HasNoChildren) = "Draft"
# (\HasNoChildren) = "Inbox"
# (\Sent \HasNoChildren) = "Sent"
# (\Trash \HasNoChildren) = "Trash"
mail_list.extend(self._get_messages_from_folder(imap))
# mail_list.extend(self._get_messages_from_folder(imap, folder="Bulk"))
# close the connection and logout
imap.close()
imap.logout()
return mail_list
def _get_messages_from_folder(self, imap, folder="INBOX") -> list:
imap.select(folder)
mail_messages = []
typ, data = imap.search(None, '(SUBJECT "{}")'.format(VALIDATION_URL_SUBJECT))
for i in data[0].split():
# fetch the email message by ID
res, msg = imap.fetch(str(i), "(RFC822)")
# res, msg = imap.fetch(str(i))
res, msg = imap.fetch(i.decode("utf-8"), "(RFC822)")
body = ''
for response in msg:
if isinstance(response, tuple):
@@ -72,17 +95,8 @@ class MailReader():
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"
mail_messages.append(mail)
return mail_messages
def clean(text):
@@ -112,12 +126,15 @@ def need_to_valid_url(url: str, successful_items) -> bool:
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]
# get email address
mail_list = MONGO_STORE_MANAGER.get_destination_emails()
# mail_address1 = MailAddress(mail="appointment2022@aol.com", password="gyilpmvyyvlcaviq")
# mail_address2 = MailAddress(mail="chenpeijun@aol.com", password="ytifuwguknzifqyb")
# mail_address2 = MailAddress(mail="sdfgfhgf1986@aol.com", password="fjwcgvhxxlywqfwm")
# 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_address1]
for mail in mail_list:
mail_reader = MailReader(mail.mail, mail.password)
successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
@@ -130,7 +147,6 @@ def read_mails():
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))
src/pojo/mail/mail_pojo.py
+6
View File
@@ -13,6 +13,12 @@ class MailAddress:
}
return dest
@staticmethod
def from_firestore_dict(source):
password = source['password']
mail = source['mail']
return MailAddress(mail=mail, password=password)
class MailPojo:
from_address: str
src/utils/excel_reader.py
+56 -7
View File
@@ -8,7 +8,7 @@ import xlsxwriter
from src.config import CONTACT_LIST_FILE
from src.db.mongo_manager import MONGO_STORE_MANAGER
from src.pojo.contact_pojo import ContactPojo
from src.pojo.mail.mail_pojo import MailAddress
from src.pojo.mail.mail_pojo import MailAddress, MailPojo
from src.utils.generate_random_passport_id import get_random_passport_id_number
phone_number_prefix = ['6']
@@ -94,6 +94,21 @@ class ExcelHelper:
return contact_list
def read_email_pojo(self, file_name=CONTACT_LIST_FILE) -> list:
email_info_in_json = pandas.read_excel(file_name).to_json(orient='records')
contact_dict_list = json.loads(email_info_in_json)
contact_list = []
count = 0
for contact_dict in contact_dict_list:
if contact_dict['email']:
email = contact_dict['email'].strip()
password = contact_dict['code']
email_destinaire = MailAddress(email, password)
count = count + 1
contact_list.append(email_destinaire)
return contact_list
def get_random_phone_numbers():
length = 8 # number of characters in the string.
@@ -124,7 +139,7 @@ def get_random_id_number() -> str:
return ran
def write_new_contacts_to_excel(valid_contacts: list, generate_passport = True):
def write_new_contacts_to_excel(valid_contacts: list, generate_passport=True):
row = 0
col = 0
# Create a workbook and add a worksheet.
@@ -149,11 +164,45 @@ def write_new_contacts_to_excel(valid_contacts: list, generate_passport = True):
workbook.close()
if __name__ == '__main__':
def write_destinaire_email(valid_contacts: list, generate_passport=True):
row = 0
col = 0
# Create a workbook and add a worksheet.
workbook = xlsxwriter.Workbook('real_contacts_{}.xlsx'.format(len(valid_contacts)))
header_data = ['name', 'phone', 'passport', 'email']
worksheet = workbook.add_worksheet()
header_format = workbook.add_format({'bold': True})
for col_num, data in enumerate(header_data):
worksheet.write(row, col_num, data, header_format)
row = row + 1
for info in valid_contacts:
info.phone = get_random_phone_numbers()
info.passport = get_random_passport_id_number()
info.mail = generate_email_from_name(info.first_name, info.last_name)
# Iterate over the data and write it out row by row.
worksheet.write(row, col, "{} {}".format(info.last_name, info.first_name))
worksheet.write(row, col + 1, info.phone)
worksheet.write(row, col + 2, info.passport)
worksheet.write(row, col + 3, info.mail)
row += 1
workbook.close()
def save_mails_to_db():
excel_reader = ExcelHelper()
contacts = excel_reader.read_names("/Users/lpan/Documents/rdv/real_contacts_318.xlsx")
print(contacts)
write_new_contacts_to_excel(valid_contacts=contacts)
emails = excel_reader.read_email_pojo("/Users/lpan/Downloads/aol_mails_21.xlsx")
print(emails)
for mail in emails:
MONGO_STORE_MANAGER.save_destinary_emails(mail)
if __name__ == '__main__':
# excel_reader = ExcelHelper()
# contacts = excel_reader.read_names("/Users/lpan/Documents/rdv/real_contacts_318.xlsx")
# print(contacts)
# write_new_contacts_to_excel(valid_contacts=contacts)
save_mails_to_db()
# for mail in excel_reader.read_mails_and_pwd():
# MONGO_STORE_MANAGER.insert_email(mail)
# for i in range(1,20):
# print(get_random_phone_numbers())