更改任务为队列+并发的格式, 登录,获取账号配置, 更新账号配置功能为单独的锁能并发两次. 更改账号被检测到自动化时自动进行解锁账号

main.py

@@ -1,5 +1,10 @@
 from spider.task import *
 from loguru import logger
+from concurrent.futures import ThreadPoolExecutor
+from queue import Queue
+import threading
+import time
+import requests
 
 logger.add("./log/logging.log", rotation="50 MB")
 
@@ -18,8 +23,6 @@ HOST = "http://118.193.40.152:8002"
 
 
 def get_task():
-    if lock._block.locked():
-        return
     url = f'{HOST}/queue/get-data'
     header = {
         'Content-Type': 'application/json'
@@ -28,10 +31,16 @@ def get_task():
         "include_task_type": [],
         "exclude_task_type": []
     }
-    response = requests.post(url, headers=header, json=data, proxies=None)
-    if response.status_code == 200:
-        result = response.json()
-        return result
+    try:
+        response = requests.post(url, headers=header, json=data, proxies=None)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            # logger.error(f"Failed to get task. Status code: {response.status_code}, Response: {response.text}")
+            return None
+    except Exception as e:
+        logger.error(f"Error occurred while getting task: {str(e)}")
+        return None
 
 
 def task_callback(tid, data, status='success', msg='success'):
@@ -65,20 +74,39 @@ def execute_task(tid, task_type, **kwargs):
 
 
 def main():
-    while True:
-        try:
-            task = get_task()
-            if task is None:
-                logger.info("无任务")
+    # 创建任务队列，最大容量为3
+    task_queue = Queue(maxsize=2)
+    # 存储正在运行的任务
+    running_tasks = set()
+
+    # 创建线程池执行任务（3个工作线程）
+    with ThreadPoolExecutor(max_workers=3) as executor:
+        while True:
+            try:
+                # 清理已完成的任务
+                running_tasks = {task for task in running_tasks if not task.done()}
+
+                # 如果队列未满，尝试获取新任务
+                if len(running_tasks) < 3:
+                    task = get_task()
+                    if task:
+                        logger.info(f"收到任务: {task['id']} - {task['task_type']}")
+                        task_data = task['data']
+                        task_data['tid'] = task['id']
+                        task_data['task_type'] = task['task_type']
+                        # 提交任务到线程池并保存future对象
+                        future = executor.submit(execute_task, **task_data)
+                        running_tasks.add(future)
+                    else:
+                        logger.info("无更多任务")
+                        time.sleep(10)
+                else:
+                    # 达到最大并发数时等待
+                    time.sleep(1)
+
+            except Exception as e:
+                logger.error(f'Main Error: {e}')
                 time.sleep(10)
-                continue
-            logger.info(f"收到任务{task}")
-            task['data']['tid'] = task['id']
-            task['data']['task_type'] = task['task_type']
-            execute_task(**task['data'])
-        except Exception as e:
-            logger.error(f'Main Error: {e}')
-            time.sleep(10)
 
 
 if __name__ == '__main__':