request_sender.py optimization

.gitignore

@@ -5,3 +5,4 @@ __pycache__/
 db/__pycache__/*
 *.iml
 venv
+.env

db/mongo_manager.py

@@ -1,7 +1,6 @@
 import datetime
 import logging
 import time
-import os
 
 from pymongo import MongoClient
 
@@ -26,22 +25,7 @@ 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",
-        )
+        client = MongoClient(MONGO_DB_URL, username='appointment', password='Rdv@20222021', authSource='appointment')
         self.db = client.appointment
         self.logger = logging.getLogger("mongoDb")
 
@@ -52,13 +36,8 @@ 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)
 
@@ -104,14 +83,8 @@ 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"
@@ -122,42 +95,32 @@ class MongoDbManager:
                 _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,
-            )
+            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
+            },
+                                          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,
-            )
+            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
+            },
+                                          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:
@@ -171,9 +134,7 @@ 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:
@@ -206,18 +167,8 @@ 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)
@@ -230,10 +181,7 @@ 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())
@@ -242,27 +190,18 @@ 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,
-                    "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,
-        )
+        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})
+        collection_to_use.delete_one({'_id': linkPojo.email})
 
 
 MONGO_STORE_MANAGER = MongoDbManager()

request_sender.py

@@ -1,54 +1,40 @@
-import datetime
 import logging
+import random
 import sys
 from threading import Thread
 
 from db.mongo_manager import MONGO_STORE_MANAGER
 from excel_reader import read_contacts
 from models.contact_pojo import ContactPojo
-from queue_message.CookiesPublisher import CookiesPublisher, SHARED_OBJECT, TEST_QUEUE, MORNING_DATA_CACHE_2, \
-    MORNING_DATA_CACHE
+from queue_message.CookiesPublisher import CookiesPublisher, SHARED_OBJECT, TEST_QUEUE, MORNING_DATA_CACHE, \
+    MORNING_DATA_CACHE_2, MORNING_DATA_CACHE_BAK
 from queue_message.appointmentrequestsender import AppointmentRequestSender
-from utiles import is_time_between
 from utils.AppLogging import init_logger
-from workers.proxies_constants import MOBILE_PROXY_LIST_FR
-
-IPFIY = 'http://api.ipify.org'
-NGROK_TEST = "https://bcc6-193-164-156-53.ngrok-free.app"
-
-
-def is_already_sent(contact: ContactPojo) -> bool:
-    already_sent_contacts = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
-    for required_contact in already_sent_contacts:
-        if contact.mail == required_contact.email:
-            return True
-    return False
 
 
 def filter_contacts(_contact_list: list) -> list:
     already_sent_contacts = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
     _link_to_validate_list = MONGO_STORE_MANAGER.get_links_to_validate()
+
+    # Optimization: Use sets for O(1) lookup complexity
+    sent_emails = {booked.email for booked in already_sent_contacts}
+    validate_emails = {link.email for link in _link_to_validate_list}
+
     _contact_list_to_book = []
     for contact in _contact_list:
-        _to_add = True
-        for booked in already_sent_contacts:
-            if contact.mail == booked.email:
-                _to_add = False
+        if contact.mail in sent_emails:
+            continue
+
         # 如果已经收到链接了，就不要再请求
-        for link_to_validate in _link_to_validate_list:
-            if contact.mail == link_to_validate.email:
-                logger.info("{}: link already received".format(contact.mail))
-                _to_add = False
-        if _to_add:
-            _contact_list_to_book.append(contact)
+        if contact.mail in validate_emails:
+            logger.info("{}: link already received".format(contact.mail))
+            continue
+
+        _contact_list_to_book.append(contact)
 
     return _contact_list_to_book
 
 
-def is_open():
-    return is_time_between(datetime.time(10, 30), datetime.time(19, 00))
-
-
 count = 0
 init_logger()
 logger = logging.getLogger()
@@ -56,36 +42,79 @@ logger = logging.getLogger()
 logger.addHandler(logging.StreamHandler(stream=sys.stdout))
 
 
-def send_appointment_request(message_queue_name, _contact_list):
+def send_appointment_request(message_queue_name, _contact_list, stop_at_hour=11, stop_at_mins=30):
     global count
     count = count + 1
     for _contact in _contact_list:
         logger.info(_contact)
     _cookiesPublisher = CookiesPublisher(queue_name=message_queue_name)
     _cookiesPublisher.set_up_connection()
-    receiver = AppointmentRequestSender(sub_contact_list=_contact_list, queue_name=message_queue_name,
-                                        cookiesPublisher=_cookiesPublisher, logger=logger)
+    _backUp_cookiesPublisher = CookiesPublisher(queue_name=MORNING_DATA_CACHE_BAK)
+    _backUp_cookiesPublisher.set_up_connection()
+    receiver = AppointmentRequestSender(sub_contact_list=_contact_list,
+                                        queue_name=message_queue_name,
+                                        cookiesPublisher=_cookiesPublisher,
+                                        bakeUpCookiesPublisher=_backUp_cookiesPublisher, logger=logger,
+                                        stop_at_hour=stop_at_hour, stop_at_mins=stop_at_mins)
     print("count is " + str(count))
     receiver.run()
 
 
-def start_send_requests():
+def start_send_requests(thread_number, contact_list, data_queue_name=MORNING_DATA_CACHE, stop_at_hour=14,
+                        stop_at_mins=56):
     print("start send requests")
-    contacts_file_path = '~/Desktop/06_05_to_test.xlsx'
-    _contact_list = read_contacts(contacts_file_path)[:1]
-    _contact_list_to_book = filter_contacts(_contact_list)
-    _segment_number = 1
-    logger.info("{} contacts to book".format(len(_contact_list_to_book)))
-    last_thread = None
-    for i in range(0, _segment_number):
+    _contact_list_to_book = filter_contacts(contact_list)
+    _segment_number = thread_number
+    total_contacts = len(_contact_list_to_book)
+    logger.info("{} contacts to book".format(total_contacts))
+
+    if total_contacts == 0:
+        return
+
+    # Optimization: Better distribution of contacts among threads
+    thread_list = []
+    chunk_size = total_contacts // _segment_number
+    remainder = total_contacts % _segment_number
+
+    start_index = 0
+    for i in range(_segment_number):
+        # If we have more threads than contacts, some threads will get empty lists, which is fine
+        if start_index >= total_contacts:
+            break
+
         logger.info("segment is {}".format(i))
-        _step = int(len(_contact_list_to_book) / _segment_number)
-        _sublist = _contact_list_to_book[i * _step:_step * (i + 1)]
-        _thread1 = Thread(target=send_appointment_request, args=(MORNING_DATA_CACHE, _sublist))
-        last_thread = _thread1
-        _thread1.start()
-    last_thread.join()
+
+        # Distribute remainder to the first few threads
+        current_chunk_size = chunk_size + (1 if i < remainder else 0)
+        end_index = start_index + current_chunk_size
+
+        _sublist = _contact_list_to_book[start_index:end_index]
+        start_index = end_index
+
+        if _sublist:
+            _thread1 = Thread(target=send_appointment_request, args=(data_queue_name, _sublist, stop_at_hour, stop_at_mins))
+            thread_list.append(_thread1)
+            _thread1.start()
+
+    for _thread in thread_list:
+        _thread.join()
+
+
+def send_request_for_file_list(file_list: list, thread_number: int = 20, data_queue_name=MORNING_DATA_CACHE,
+                               stop_at_hour=11, stop_at_mins=30):
+    logger.info("stop_at_hour is " + str(stop_at_hour) + " stop_at_mins is " + str(stop_at_mins))
+    for _file_path in file_list:
+        logger.info("send request for file: " + _file_path)
+        _contact_list = read_contacts(_file_path)
+        random.shuffle(_contact_list)
+        start_send_requests(thread_number=thread_number, contact_list=_contact_list,
+                            data_queue_name=data_queue_name, stop_at_hour=stop_at_hour, stop_at_mins=stop_at_mins)
 
 
 if __name__ == '__main__':
-    start_send_requests()
+    # file_list = ['~/Desktop/contact_list_2024-05-23.xlsx',
+    #              '~/Desktop/contact_list_2024-05-21.xlsx',
+    # file_list = ['~/Desktop/contact_list_2025-10-30.xlsx']
+    file_list = ['~/Desktop/contact_list_2025-11-28.xlsx']
+    send_request_for_file_list(file_list=file_list, thread_number=10,
+                               data_queue_name=MORNING_DATA_CACHE_2, stop_at_hour=19, stop_at_mins=50)

request_sender_test.py

@@ -1,107 +0,0 @@
-import datetime
-import logging
-import random
-import sys
-from threading import Thread
-
-from db.mongo_manager import MONGO_STORE_MANAGER
-from excel_reader import read_contacts
-from models.contact_pojo import ContactPojo
-from queue_message.CookiesPublisher import CookiesPublisher, SHARED_OBJECT, TEST_QUEUE, MORNING_DATA_CACHE, \
-    MORNING_DATA_CACHE_2, MORNING_DATA_CACHE_BAK
-from queue_message.appointmentrequestsender import AppointmentRequestSender
-from utils.AppLogging import init_logger
-
-
-def is_already_sent(contact: ContactPojo) -> bool:
-    already_sent_contacts = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
-    for required_contact in already_sent_contacts:
-        if contact.mail == required_contact.email:
-            return True
-    return False
-
-
-def filter_contacts(_contact_list: list) -> list:
-    already_sent_contacts = MONGO_STORE_MANAGER.get_all_successful_items_for_day()
-    _link_to_validate_list = MONGO_STORE_MANAGER.get_links_to_validate()
-    _contact_list_to_book = []
-    for contact in _contact_list:
-        _to_add = True
-        for booked in already_sent_contacts:
-            if contact.mail == booked.email:
-                _to_add = False
-        # 如果已经收到链接了，就不要再请求
-        for link_to_validate in _link_to_validate_list:
-            if contact.mail == link_to_validate.email:
-                logger.info("{}: link already received".format(contact.mail))
-                _to_add = False
-        if _to_add:
-            _contact_list_to_book.append(contact)
-
-    return _contact_list_to_book
-
-
-count = 0
-init_logger()
-logger = logging.getLogger()
-
-logger.addHandler(logging.StreamHandler(stream=sys.stdout))
-
-
-def send_appointment_request(message_queue_name, _contact_list, stop_at_hour=11, stop_at_mins=30):
-    global count
-    count = count + 1
-    for _contact in _contact_list:
-        logger.info(_contact)
-    _cookiesPublisher = CookiesPublisher(queue_name=message_queue_name)
-    _cookiesPublisher.set_up_connection()
-    _backUp_cookiesPublisher = CookiesPublisher(queue_name=MORNING_DATA_CACHE_BAK)
-    _backUp_cookiesPublisher.set_up_connection()
-    receiver = AppointmentRequestSender(sub_contact_list=_contact_list,
-                                        queue_name=message_queue_name,
-                                        cookiesPublisher=_cookiesPublisher,
-                                        bakeUpCookiesPublisher=_backUp_cookiesPublisher, logger=logger,
-                                        stop_at_hour=stop_at_hour, stop_at_mins=stop_at_mins)
-    print("count is " + str(count))
-    receiver.run()
-
-
-def start_send_requests(thread_number, contact_list, data_queue_name=MORNING_DATA_CACHE, stop_at_hour=14,
-                        stop_at_mins=56):
-    print("start send requests")
-    _contact_list_to_book = filter_contacts(contact_list)
-    _segment_number = thread_number
-    logger.info("{} contacts to book".format(len(_contact_list_to_book)))
-    # last_thread = None
-    thread_list = []
-    for i in range(0, _segment_number):
-        logger.info("segment is {}".format(i))
-        _step = int(len(_contact_list_to_book) / _segment_number)
-        _sublist = _contact_list_to_book[i * _step:_step * (i + 1)]
-        _thread1 = Thread(target=send_appointment_request, args=(data_queue_name, _sublist, stop_at_hour, stop_at_mins))
-        thread_list.append(_thread1)
-        _thread1.start()
-    for _thread in thread_list:
-        _thread.join()
-
-
-def send_request_for_file_list(file_list: list, thread_number: int = 20, data_queue_name=MORNING_DATA_CACHE,
-                               stop_at_hour=11, stop_at_mins=30):
-    logger.info("stop_at_hour is " + str(stop_at_hour) + " stop_at_mins is " + str(stop_at_mins))
-    for _file_path in file_list:
-        logger.info("send request for file: " + _file_path)
-        _contact_list = read_contacts(_file_path)
-        random.shuffle(_contact_list)
-        start_send_requests(thread_number=thread_number, contact_list=_contact_list,
-                            data_queue_name=data_queue_name, stop_at_hour=stop_at_hour, stop_at_mins=stop_at_mins)
-
-
-if __name__ == '__main__':
-    # file_list = ['~/Desktop/contact_list_2024-05-23.xlsx',
-    #              '~/Desktop/contact_list_2024-05-21.xlsx',
-    # file_list = ['~/Desktop/contact_list_2025-10-30.xlsx']
-    file_list = ['~/Desktop/contact_list_2025-11-28.xlsx']
-    # file_list = ['~/Desktop/contact_list_2025-11-06.xlsx']
-    # file_list = ['~/Desktop/contact_list_all.xlsx']
-    send_request_for_file_list(file_list=file_list, thread_number=20,
-                               data_queue_name=MORNING_DATA_CACHE_2, stop_at_hour=19, stop_at_mins=50)

scripts/db_sync.py

@@ -10,36 +10,21 @@ from db.mongo_manager import MONGO_DB_URL
 MONGO_HOST = "mongo.lpaconsulting.fr"
 MONGO_PORT = "27017"
 MONGO_DB_NAME = "appointment"  # 你要备份/恢复的数据库名
-
-# 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
+MONGO_USER = "appointment"       # 如果没有密码，保持为空字符串 ""
+MONGO_PASS = "Rdv@2022"       # 如果没有密码，保持为空字符串 ""
 
 # 备份存放的根目录
 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. 创建带有时间戳的备份文件夹
@@ -55,14 +40,10 @@ 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
     ]
 
     # 添加认证参数
@@ -73,14 +54,13 @@ def backup_mongo():
         result = subprocess.run(cmd, check=True, text=True, capture_output=True)
         print(f"[+] 备份成功！")
         print(f"    存储路径: {backup_path}")
-        print(f"    日志: {result.stderr}")  # mongodump 通常把进度输出到 stderr
+        print(f"    日志: {result.stderr}") # mongodump 通常把进度输出到 stderr
         return backup_path
     except subprocess.CalledProcessError as e:
         print(f"[-] 备份失败: {e}")
         print(f"    错误信息: {e.stderr}")
         return None
 
-
 def restore_mongo(backup_source_path):
     """
     执行恢复操作
@@ -92,9 +72,7 @@ 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} ...")
@@ -103,14 +81,11 @@ 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())
@@ -123,7 +98,6 @@ def restore_mongo(backup_source_path):
         print(f"[-] 恢复失败: {e}")
         print(f"    错误信息: {e.stderr}")
 
-
 # ================= 主程序入口 =================
 if __name__ == "__main__":
     print("请选择操作:")
@@ -150,9 +124,7 @@ 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("[-] 无效的选择。")