password in env

db/mongo_manager.py

@@ -1,6 +1,7 @@
 import datetime
 import logging
 import time
+import os
 
 from pymongo import MongoClient
 
@@ -25,7 +26,22 @@ CONTACT_LIST_SERIAL_MAP = "CONTACT_LIST_SERIAL_MAP"
 
 class MongoDbManager:
     def __init__(self):
-        client = MongoClient(MONGO_DB_URL, username='appointment', password='Rdv@20222021', authSource='appointment')
+        # 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")
 
@@ -36,8 +52,13 @@ class MongoDbManager:
     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)
+            collection_to_use.replace_one(
+                filter={
+                    "_id": reserve.id,
+                },
+                replacement=reserve.to_firestore_dict(),
+                upsert=True,
+            )
         except Exception as Error:
             self.logger.info(Error)
 
@@ -83,8 +104,14 @@ class MongoDbManager:
             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 = ""):
+    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"
@@ -95,32 +122,42 @@ class MongoDbManager:
                 _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'serial': serial,
-                u'model': model,
-                u'ip_country': _ip_country,
-                u'_used_ip': _used_ip,
-                "updated_at": updated_at
+            collection_to_use.replace_one(
+                filter={
+                    "_id": mail_address,
                 },
-                                          upsert=True)
+                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={
-                u'url': link,
-                u'serial': serial,
-                u'model': model,
-                u'ip_country': _ip_country,
-                u'_used_ip': _used_ip,
-                "updated_at": updated_at
+            collection_to_use.replace_one(
+                filter={
+                    "_id": link,
                 },
-                                          upsert=True)
+                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})
+            mailDocument = collection_to_use.find_one(filter={"_id": email})
             if mailDocument is not None:
                 return MailAddress.from_firestore_dict(mailDocument).password
             else:
@@ -134,7 +171,9 @@ class MongoDbManager:
         _cursor = self.db[_collection_name]
         registered_user_list = []
         for document in _cursor.find():
-            registered_user_list.append(RegisteredUserPojo.from_firestore_dict(document))
+            registered_user_list.append(
+                RegisteredUserPojo.from_firestore_dict(document)
+            )
         return registered_user_list
 
     def get_destination_emails(self) -> list:
@@ -167,8 +206,18 @@ class MongoDbManager:
             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):
+    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)
@@ -181,7 +230,10 @@ class MongoDbManager:
         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)
+        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())
@@ -190,18 +242,27 @@ class MongoDbManager:
             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,
+        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)
+                    "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})
+        collection_to_use.delete_one({"_id": linkPojo.email})
 
 
 MONGO_STORE_MANAGER = MongoDbManager()

scripts/db_sync.py

@@ -10,21 +10,36 @@ from db.mongo_manager import MONGO_DB_URL
 MONGO_HOST = "mongo.lpaconsulting.fr"
 MONGO_PORT = "27017"
 MONGO_DB_NAME = "appointment"  # 你要备份/恢复的数据库名
-MONGO_USER = "appointment"       # 如果没有密码，保持为空字符串 ""
-MONGO_PASS = "Rdv@2022"       # 如果没有密码，保持为空字符串 ""
+
+# Get MongoDB credentials from environment variables
+MONGO_USER = os.getenv(
+    "MONGO_USER", "appointment"
+)  # Default to 'appointment' if not set
+MONGO_PASS = os.getenv("MONGO_PASS", "Rdv@2022")  # Default to 'Rdv@2022' if not set
 
 # 备份存放的根目录
 BACKUP_DIR_ROOT = "./mongo_backups"
 
 # ===========================================
 
+
 def get_auth_args():
     """构建认证参数列表"""
     args = []
     if MONGO_USER and MONGO_PASS:
-        args.extend(["--username", MONGO_USER, "--password", MONGO_PASS, "--authenticationDatabase", "appointment"])
+        args.extend(
+            [
+                "--username",
+                MONGO_USER,
+                "--password",
+                MONGO_PASS,
+                "--authenticationDatabase",
+                "appointment",
+            ]
+        )
     return args
 
+
 def backup_mongo():
     """执行备份操作"""
     # 1. 创建带有时间戳的备份文件夹
@@ -40,10 +55,14 @@ def backup_mongo():
     # 命令格式: mongodump --host <host> --port <port> --db <db> --out <path> [auth]
     cmd = [
         "mongodump",
-        "--host", MONGO_HOST,
-        "--port", MONGO_PORT,
-        "--db", MONGO_DB_NAME,
-        "--out", backup_path
+        "--host",
+        MONGO_HOST,
+        "--port",
+        MONGO_PORT,
+        "--db",
+        MONGO_DB_NAME,
+        "--out",
+        backup_path,
     ]
 
     # 添加认证参数
@@ -61,6 +80,7 @@ def backup_mongo():
         print(f"    错误信息: {e.stderr}")
         return None
 
+
 def restore_mongo(backup_source_path):
     """
     执行恢复操作
@@ -72,7 +92,9 @@ def restore_mongo(backup_source_path):
     target_dir = os.path.join(backup_source_path, MONGO_DB_NAME)
 
     if not os.path.exists(target_dir):
-        print(f"[-] 错误: 在路径 {backup_source_path} 下找不到数据库 {MONGO_DB_NAME} 的备份文件。")
+        print(
+            f"[-] 错误: 在路径 {backup_source_path} 下找不到数据库 {MONGO_DB_NAME} 的备份文件。"
+        )
         return
 
     print(f"[*] 开始恢复数据库: {MONGO_DB_NAME} 从 {target_dir} ...")
@@ -81,11 +103,14 @@ def restore_mongo(backup_source_path):
     # 命令格式: mongorestore --host <host> --port <port> --db <db> <path_to_bson_files> [auth]
     cmd = [
         "mongorestore",
-        "--host", MONGO_HOST,
-        "--port", MONGO_PORT,
-        "--db", MONGO_DB_NAME,
+        "--host",
+        MONGO_HOST,
+        "--port",
+        MONGO_PORT,
+        "--db",
+        MONGO_DB_NAME,
         "--drop",  # 警告：这会在恢复前删除现有集合，确保数据干净。根据需要移除此项。
-        target_dir
+        target_dir,
     ]
 
     cmd.extend(get_auth_args())
@@ -98,6 +123,7 @@ def restore_mongo(backup_source_path):
         print(f"[-] 恢复失败: {e}")
         print(f"    错误信息: {e.stderr}")
 
+
 # ================= 主程序入口 =================
 if __name__ == "__main__":
     print("请选择操作:")
@@ -124,7 +150,9 @@ if __name__ == "__main__":
                 try:
                     idx_choice = int(input("\n请选择要恢复的备份编号: ")) - 1
                     if 0 <= idx_choice < len(backups):
-                        selected_backup = os.path.join(BACKUP_DIR_ROOT, backups[idx_choice])
+                        selected_backup = os.path.join(
+                            BACKUP_DIR_ROOT, backups[idx_choice]
+                        )
                         restore_mongo(selected_backup)
                     else:
                         print("[-] 无效的选择。")