fix: git init

util/json_util.py

@@ -0,0 +1,42 @@
+"""
+自定义的一些 json功能
+"""
+import re
+
+
+def underscore_to_camelcase(snake_str):
+    """ 将下划线命名转换为驼峰命名 """
+    components = snake_str.split('_')
+    return components[0] + ''.join(x.title() for x in components[1:])
+
+def camelcase_to_underscore(camel_str):
+    """将驼峰命名转换为下划线命名"""
+    return re.sub(r'(?<!^)(?=[A-Z])', '_', camel_str).lower()
+
+
+def convert_keys_to_camelcase(obj):
+    """ 递归处理 JSON 对象 """
+    if isinstance(obj, dict):
+        new_obj = {}
+        for k, v in obj.items():
+            new_key = underscore_to_camelcase(k)
+            new_obj[new_key] = convert_keys_to_camelcase(v)
+        return new_obj
+    elif isinstance(obj, list):
+        return [convert_keys_to_camelcase(item) for item in obj]
+    else:
+        return obj
+
+
+def convert_keys_to_underscore(obj):
+    if isinstance(obj, dict):
+        new_obj = {}
+        for k, v in obj.items():
+            new_key = camelcase_to_underscore(k)
+            new_obj[new_key] = convert_keys_to_underscore(v)
+        return new_obj
+    elif isinstance(obj, list):
+        return [convert_keys_to_underscore(item) for item in obj]
+    else:
+        return obj
+

util/lock_util.py

@@ -0,0 +1,35 @@
+"""
+
+"""
+import logging
+import threading
+
+
+class LockManager:
+    """
+    全局锁管理，每个账号只能打开一个上下文相同的浏览器
+    """
+
+    def __init__(self):
+        self.locks = {}
+
+    def acquire_lock(self, key):
+        if key not in self.locks:
+            self.locks[key] = threading.Lock()
+        acquire = self.locks[key].acquire(timeout=300)
+        if acquire:
+            logging.info(f"{key} 获取锁成功")
+
+    def release_lock(self, key):
+        if key in self.locks:
+            self.locks[key].release()
+            logging.info(f"{key} 释放锁成功")
+
+    def is_locked(self, key):
+        """
+        检查给定的键是否处于锁定状态
+        """
+        if key in self.locks:
+            return self.locks[key].locked()
+        else:
+            return False