Lei PAN
124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
import datetime
|
|
import logging
|
|
|
|
from pymongo import MongoClient
|
|
|
|
from src import params
|
|
from src.pojo.ReserveResultPojo import ReserveResultPojo
|
|
from src.pojo.ResultEnum import ResultEnum
|
|
from src.pojo.black_contact import BlackContactPojo
|
|
from src.pojo.contact_pojo import ContactPojo
|
|
|
|
MONGO_DB_URL = "mongo.lpaconsulting.fr"
|
|
CAPTCHA_ERROR_COLLECTION_PREFIX = "CAPTCHA_ERROR_"
|
|
BLACK_LIST = "BLACK_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 insert_captcha_error_contact(self, contact: ContactPojo):
|
|
day = str(datetime.date.today())
|
|
collection_name = CAPTCHA_ERROR_COLLECTION_PREFIX + day
|
|
try:
|
|
collection_to_use = self.db[collection_name]
|
|
collection_to_use.replace_one(filter={'_id': contact.mail, }, replacement=contact.to_firestore_dict(),
|
|
upsert=True)
|
|
except Exception as error:
|
|
self.logger.info(error)
|
|
|
|
def insert_blacklist_contact(self, contact: ContactPojo):
|
|
collection_name = BLACK_LIST
|
|
black_contact = BlackContactPojo(contact.phone, contact.passport, contact.last_name, contact.first_name,
|
|
contact.mail)
|
|
try:
|
|
collection_to_use = self.db[collection_name]
|
|
collection_to_use.replace_one(filter={'_id': black_contact.mail, },
|
|
replacement=black_contact.to_firestore_dict(),
|
|
upsert=True)
|
|
except Exception as error:
|
|
self.logger.info(error)
|
|
|
|
def get_blacklist_contacts(self) -> list:
|
|
collection_name = BLACK_LIST
|
|
black_list_contacts = []
|
|
try:
|
|
collection_to_use = self.db[collection_name]
|
|
for document in collection_to_use.find():
|
|
black_list_contacts.append(BlackContactPojo.from_firestore_dict(document))
|
|
except Exception as error:
|
|
self.logger.info(error)
|
|
return black_list_contacts
|
|
|
|
def get_all_successful_items_for_day(self) -> list:
|
|
params.oracle_log_sender.send_read_db_event("get_all_successful_items_for_day")
|
|
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 get_all_successful_items_for_one_day(self, day_in_str: str) -> list:
|
|
params.oracle_log_sender.send_read_db_event("get_all_successful_items_for_one_day")
|
|
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 update_reserve_result(self, reserve_id: str, result: ResultEnum):
|
|
collection_name = str(datetime.date.today())
|
|
collection = self.db[collection_name]
|
|
collection.find_one_and_update({'_id': reserve_id}, {"$set": {"accepted": result.name}}, upsert=False)
|
|
|
|
def get_captcha_error_contacts_for_current_day(self) -> list:
|
|
day = str(datetime.date.today())
|
|
collection_name = CAPTCHA_ERROR_COLLECTION_PREFIX + day
|
|
cursor = self.db[collection_name]
|
|
contact_list = []
|
|
for document in cursor.find():
|
|
contact_list.append(ContactPojo.from_firestore_dict(document))
|
|
return contact_list
|
|
|
|
def delete_captcha_error_contact_for_current_day(self, contact: ContactPojo):
|
|
day = str(datetime.date.today())
|
|
collection_name = CAPTCHA_ERROR_COLLECTION_PREFIX + day
|
|
collection = self.db[collection_name]
|
|
to_delete = {'_id': contact.mail}
|
|
try:
|
|
collection.delete_one(to_delete)
|
|
except Exception as error:
|
|
self.logger.info(error)
|
|
|
|
def remove_contact_from_black_list(self, contact: ContactPojo):
|
|
collection_name = BLACK_LIST
|
|
collection = self.db[collection_name]
|
|
to_delete = {'_id': contact.mail}
|
|
try:
|
|
collection.delete_one(to_delete)
|
|
except Exception as error:
|
|
self.logger.info(error)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
db_manager = MongoDbManager()
|
|
black_list = db_manager.get_blacklist_contacts()
|
|
for contact in black_list:
|
|
print(contact) |