86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
from pathlib import Path
|
|
|
|
from src.db.mongo_manager import MONGO_STORE_MANAGER
|
|
from src.pojo.contact_pojo import ContactPojo
|
|
from src.utils.excel_reader import read_contacts
|
|
|
|
DOMAIN_TO_INCLUDE = ["aol.com", "yahoo.com", "gmx.de", "gmx.net", "inbox.lv"]
|
|
|
|
|
|
def get_all_collections():
|
|
return MONGO_STORE_MANAGER.list_collection_names()
|
|
|
|
|
|
def need_to_check_mail(mail_address):
|
|
for domain in DOMAIN_TO_INCLUDE:
|
|
if domain in mail_address:
|
|
return True
|
|
return False
|
|
|
|
|
|
def get_all_not_used_mails():
|
|
all_emails = MONGO_STORE_MANAGER.get_destination_emails()
|
|
all_valid_contact_list = MONGO_STORE_MANAGER.get_all_contacts_to_book()
|
|
not_used_emails = []
|
|
for mail_item in all_emails:
|
|
if need_to_check_mail(mail_item.mail):
|
|
_need_to_add = True
|
|
for valid_contact in all_valid_contact_list:
|
|
if valid_contact.mail == mail_item.mail:
|
|
_need_to_add = False
|
|
break
|
|
if _need_to_add:
|
|
not_used_emails.append(mail_item)
|
|
return not_used_emails
|
|
|
|
|
|
def get_all_old_contact_set(start_with="2024"):
|
|
all_collection_list = get_all_collections()
|
|
collection_to_check = []
|
|
for col in all_collection_list:
|
|
if col.startswith(start_with):
|
|
collection_to_check.append(col)
|
|
items_set = set()
|
|
for collection in collection_to_check:
|
|
items_for_day = MONGO_STORE_MANAGER.get_all_successful_items_for_one_day(collection)
|
|
items_set.update(items_for_day)
|
|
return items_set
|
|
|
|
|
|
def get_all_contact_with_source_from():
|
|
_list_to_check = get_all_old_contact_set()
|
|
_to_return = []
|
|
for contact in _list_to_check:
|
|
if contact.serial is not None and len(contact.serial) > 0:
|
|
_to_return.append(contact)
|
|
return _to_return
|
|
|
|
|
|
def generate_not_used_contact_list():
|
|
not_used_mail_list = get_all_not_used_mails()
|
|
all_old_contact_set = get_all_old_contact_set()
|
|
contact_list = []
|
|
for mail_item in not_used_mail_list:
|
|
print("mail is " + mail_item.mail)
|
|
for contact in all_old_contact_set:
|
|
if contact.mail == mail_item.mail:
|
|
if contact.phone is not None and len(contact.phone) > 0:
|
|
new_contact = ContactPojo(mail=mail_item.mail, phone_number=contact.phone,
|
|
passport_number=contact.passport,
|
|
last_name=contact.last_name, first_name=contact.first_name,
|
|
store="random")
|
|
contact_list.append(new_contact)
|
|
break
|
|
return contact_list
|
|
|
|
|
|
def upload_to_collection():
|
|
_contacts = read_contacts(str(Path.home()) + "/Desktop/contact_list_2024-11-05.xlsx")
|
|
MONGO_STORE_MANAGER.upload_contact_list_to_collection(_contacts,"CONTACT_LIST_SERIAL_MAP")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
upload_to_collection()
|
|
# contact_to_save = get_all_contact_with_source_from()
|
|
# write_new_contacts_to_excel(valid_contacts=contact_to_save, file_name="contacts_with_source")
|