appointment_request/db/mongo_manager.py

import datetime
import logging
import time

from pymongo import MongoClient

from models.LinkPojo import LinkPojo
from models.ReserveResultPojo import ReserveResultPojo
from models.contact_pojo import ContactPojo
from models.mail_pojo import MailAddress

MONGO_DB_URL = "mongo.lpaconsulting.fr"
CAPTCHA_ERROR_COLLECTION_PREFIX = "CAPTCHA_ERROR_"
BLACK_LIST = "BLACK_LIST"
ACCEPTED_APPOINTMENT_LIST = "ACCEPTED_APPOINTMENT_LIST"
CONTACT_LIST_TO_BOOK = "CONTACT_LIST_TO_BOOK"
EMAIL_LIST = "EMAIL_LIST"
DESTINATION_EMAIL_LIST = "DESTINATION_EMAIL_LIST"
LINKS_TO_VALIDATE = "LINKS_TO_VALIDATE"
INVALID_EMAIL_LIST = "INVALID_EMAIL_LIST"


class MongoDbManager:
    def __init__(self):
        client = MongoClient(MONGO_DB_URL, username='appointment', password='Rdv@2022', authSource='appointment')
        self.db = client.appointment
        self.logger = logging.getLogger("mongoDb")

    def insert_one(self, collection_name: str, dict: dict):
        collection_to_use = self.db[collection_name]
        collection_to_use.insert_one(dict)

    def insert_reserve_result(self, collection_name, reserve: ReserveResultPojo):
        try:
            collection_to_use = self.db[collection_name]
            collection_to_use.replace_one(filter={'_id': reserve.id, }, replacement=reserve.to_firestore_dict(),
                                          upsert=True)
        except Exception as Error:
            self.logger.info(Error)

    def get_all_successful_items_for_day(self) -> list:
        collection_name = str(datetime.date.today())
        result_list = []
        cursor = self.db[collection_name]
        for document in cursor.find():
            result_list.append(ReserveResultPojo.from_firestore_dict(document))
        return result_list

    def count_all_successful_items_for_day(self, day: str) -> list:
        return self.db[day].count_documents({})

    def get_all_successful_items_for_yesterday(self) -> list:
        yesterday = datetime.date.today() - datetime.timedelta(days=1)
        collection_name = str(yesterday)
        result_list = []
        cursor = self.db[collection_name]
        for document in cursor.find():
            result_list.append(ReserveResultPojo.from_firestore_dict(document))
        return result_list

    def get_all_successful_items_for_one_day(self, day_in_str: str) -> list:
        collection_name = day_in_str
        result_list = []
        cursor = self.db[collection_name]
        for document in cursor.find():
            result_list.append(ReserveResultPojo.from_firestore_dict(document))
        return result_list

    def get_all_contact_to_book_list(self) -> list:
        result_list = []
        cursor = self.db[CONTACT_LIST_TO_BOOK]
        for document in cursor.find():
            result_list.append(ContactPojo.from_firestore_dict(document))
        return result_list

    def save_links_to_validate(self, link: str, mail_address: str, _all_contact_list: list):
        collection_to_use = self.db[LINKS_TO_VALIDATE]
        updated_at = time.strftime("%H:%M:%S", time.localtime())
        _ip_country = "FR"
        # find ip_country info
        for _contact in _all_contact_list:
            if _contact.mail == mail_address:
                _ip_country = _contact.ip_country

        if len(mail_address) > 0:
            collection_to_use.replace_one(filter={'_id': mail_address, }, replacement={
                u'url': link,
                u'email': mail_address,
                u'ip_country': _ip_country,
                "updated_at": updated_at
            },
                                          upsert=True)
        else:
            collection_to_use.replace_one(filter={'_id': link, }, replacement={
                u'url': link,
                u'ip_country': _ip_country,
                "updated_at": updated_at
            },
                                          upsert=True)

    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)
            print(error)
        return email_list

    def get_links_to_validate(self) -> list:
        collection_name = LINKS_TO_VALIDATE
        link_list = []
        try:
            collection_to_use = self.db[collection_name]
            for document in collection_to_use.find():
                link_list.append(LinkPojo.from_firestore_dict(document))
        except Exception as error:
            self.logger.info(error)
        return link_list

    def link_validated_for_result(self, link: str, linkPojo: LinkPojo, state=True, is_duplicated=False):
        print("link_validated_for_result() called with url = " + link)
        if is_duplicated:
            _id = link.split("/")[-2]
        else:
            _id = link.split("/")[-1]
        print("link_validated_for_result() called with id = " + _id)

        collection_name = str(datetime.date.today())
        print("link_validated_for_result() called with collection_name = " + collection_name)

        collection = self.db[collection_name]
        validated_at = time.strftime("%H:%M:%S", time.localtime())
        validated_by = "requests"
        if is_duplicated:
            validated_by = "Double"
        collection.find_one_and_update({'_id': _id}, {
            "$set": {"url_validated": state, "validated_at": validated_at, "id": _id, "email": linkPojo.email,
                     "validated_by": validated_by}},
                                       upsert=True)
        #     remove the link from db
        collection_to_use = self.db[LINKS_TO_VALIDATE]
        collection_to_use.delete_one({'_id': linkPojo.email})


MONGO_STORE_MANAGER = MongoDbManager()