appointment_request/db/mongo_manager.py

import datetime
import logging
import time
import os

from pymongo import MongoClient

from models.AcceptedResultPojo import AcceptedResultPojo
from models.LinkPojo import LinkPojo
from models.ReserveResultPojo import ReserveResultPojo
from models.contact_pojo import ContactPojo
from models.mail_pojo import MailAddress
from models.regisered_user_pojo import RegisteredUserPojo

MONGO_DB_URL = "mongodb://mongo2.lpaconsulting.fr/?timeoutMS=100000"
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"
CONTACT_LIST_SERIAL_MAP = "CONTACT_LIST_SERIAL_MAP"


class MongoDbManager:
    def __init__(self):
        # Get username and password from environment variables
        mongo_username = os.getenv("MONGO_USERNAME")
        mongo_password = os.getenv("MONGO_PASSWORD")

        # Validate that environment variables exist
        if not mongo_username or not mongo_password:
            raise ValueError(
                "MONGO_USERNAME and MONGO_PASSWORD environment variables must be set"
            )

        client = MongoClient(
            MONGO_DB_URL,
            username=mongo_username,
            password=mongo_password,
            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 get_all_contact_serial_list(self) -> list:
        result_list = []
        cursor = self.db[CONTACT_LIST_SERIAL_MAP]
        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,
        model: str,
        _all_contact_list: list,
        _used_ip: str = "",
    ):
        collection_to_use = self.db[LINKS_TO_VALIDATE]
        updated_at = time.strftime("%H:%M:%S", time.localtime())
        _ip_country = "FR"
        serial = "requests"
        # 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={
                    "url": link,
                    "email": mail_address,
                    "serial": serial,
                    "model": model,
                    "ip_country": _ip_country,
                    "_used_ip": _used_ip,
                    "updated_at": updated_at,
                },
                upsert=True,
            )
        else:
            collection_to_use.replace_one(
                filter={
                    "_id": link,
                },
                replacement={
                    "url": link,
                    "serial": serial,
                    "model": model,
                    "ip_country": _ip_country,
                    "_used_ip": _used_ip,
                    "updated_at": updated_at,
                },
                upsert=True,
            )

    def get_code_for_email(self, email: str):
        collection_name = DESTINATION_EMAIL_LIST
        try:
            collection_to_use = self.db[collection_name]
            mailDocument = collection_to_use.find_one(filter={"_id": email})
            if mailDocument is not None:
                return MailAddress.from_firestore_dict(mailDocument).password
            else:
                return ""
        except Exception as error:
            self.logger.info(error)
            return ""

    def get_all_registered_users(self) -> list:
        _collection_name = "Registered_users"
        _cursor = self.db[_collection_name]
        registered_user_list = []
        for document in _cursor.find():
            registered_user_list.append(
                RegisteredUserPojo.from_firestore_dict(document)
            )
        return registered_user_list

    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 insert_accepted_reserve(self, accepted_pojo: AcceptedResultPojo):
        try:
            collection_to_use = self.db[ACCEPTED_APPOINTMENT_LIST]
            collection_to_use.insert_one(accepted_pojo.to_firestore_dict())
        except Exception as Error:
            self.logger.info(Error)

    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,
        is_invalid=False,
        segement_position=1,
        ua="",
        model="",
        timestamp_in_s: list = None,
    ):
        if timestamp_in_s is None:
            timestamp_in_s = []
        print("link_validated_for_result() called with url = " + link)
        if is_duplicated:
            _id = link.split("/")[-2]
        elif is_invalid:
            _id = linkPojo.url.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:" + str(segement_position)
        if is_invalid:
            validated_by = "Invalid"
        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,
                    "url": link,
                    "validated_by_model": model,
                    "serial": linkPojo.serial,
                    "validated_by_ua": ua,
                    "timestamp_in_s": "-".join(str(x) for x in timestamp_in_s),
                    "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()