初始提交: 通用数据同步服务

- 基于OpenResty的RESTful API服务
- 支持Cookie/Token等数据类型的存储和管理
- 登录认证保护
- Web管理界面
- 数据过期管理

scripts/utils.py

@@ -0,0 +1,185 @@
+"""
+通用数据同步服务 - 工具函数库
+"""
+
+import json
+import requests
+from pathlib import Path
+from typing import Optional, Dict, Any, List
+
+
+SERVER_URL = "http://47.122.126.244:5001"
+PASSWORD = "admin123123123"
+
+
+def upload_data(tag: str, data_type: str, data: Dict[Any, Any], 
+                expires_in: Optional[int] = None, 
+                note: Optional[str] = None,
+                login_url: Optional[str] = None) -> bool:
+    """
+    上传数据到同步服务
+    
+    Args:
+        tag: 数据标签（格式: 项目名:账号名）
+        data_type: 数据类型 (cookie/token/credential/session/custom)
+        data: 数据内容
+        expires_in: 过期时间（秒）
+        note: 备注
+        login_url: 登录链接
+    
+    Returns:
+        bool: 上传成功返回 True
+    """
+    payload = {
+        "password": PASSWORD,
+        "tag": tag,
+        "type": data_type,
+        "data": data,
+        "metadata": {}
+    }
+    
+    if expires_in:
+        payload["metadata"]["expires_in"] = expires_in
+    if note:
+        payload["metadata"]["note"] = note
+    if login_url:
+        payload["metadata"]["login_url"] = login_url
+    
+    try:
+        response = requests.post(
+            f"{SERVER_URL}/api/data",
+            json=payload,
+            timeout=10
+        )
+        return response.status_code == 200
+    except Exception as e:
+        print(f"⚠️ 上传失败: {e}")
+        return False
+
+
+def get_data(tag: str) -> Optional[Dict[Any, Any]]:
+    """
+    获取数据
+    
+    Args:
+        tag: 数据标签
+    
+    Returns:
+        Dict: 数据内容，失败返回 None
+    """
+    try:
+        response = requests.get(
+            f"{SERVER_URL}/api/data/{tag}",
+            headers={"X-Password": PASSWORD},
+            timeout=10
+        )
+        
+        if response.status_code == 200:
+            return response.json()
+        return None
+    except Exception as e:
+        print(f"⚠️ 获取数据失败: {e}")
+        return None
+
+
+def delete_data(tag: str) -> bool:
+    """
+    删除数据
+    
+    Args:
+        tag: 数据标签
+    
+    Returns:
+        bool: 删除成功返回 True
+    """
+    try:
+        response = requests.delete(
+            f"{SERVER_URL}/api/data/{tag}",
+            headers={"X-Password": PASSWORD},
+            timeout=10
+        )
+        return response.status_code == 200
+    except Exception as e:
+        print(f"⚠️ 删除失败: {e}")
+        return False
+
+
+def list_all_tags() -> List[Dict[str, Any]]:
+    """
+    列出所有 tag
+    
+    Returns:
+        List: tag 列表
+    """
+    try:
+        response = requests.get(
+            f"{SERVER_URL}/api/data",
+            timeout=10
+        )
+        
+        if response.status_code == 200:
+            data = response.json()
+            return data.get("tags", [])
+        return []
+    except Exception as e:
+        print(f"⚠️ 获取列表失败: {e}")
+        return []
+
+
+def get_copy_config(tag: str) -> Optional[str]:
+    """
+    获取一键复制配置
+    
+    Args:
+        tag: 数据标签
+    
+    Returns:
+        str: 配置文本
+    """
+    try:
+        response = requests.get(
+            f"{SERVER_URL}/api/data/copy/{tag}",
+            timeout=10
+        )
+        
+        if response.status_code == 200:
+            data = response.json()
+            return data.get("copy_text")
+        return None
+    except Exception as e:
+        print(f"⚠️ 获取配置失败: {e}")
+        return None
+
+
+# 便捷函数：从同步服务加载 Cookie 数据
+def load_cookies_from_sync(tag: str) -> Optional[List[Dict[str, Any]]]:
+    """
+    从同步服务加载 Cookie 数据
+    
+    Args:
+        tag: 数据标签
+    
+    Returns:
+        List: Cookie 列表，失败返回 None
+    """
+    data = get_data(tag)
+    if data and data.get("type") == "cookie":
+        return data.get("data", {}).get("cookies")
+    return None
+
+
+# 便捷函数：从同步服务加载 Token 数据
+def load_token_from_sync(tag: str) -> Optional[str]:
+    """
+    从同步服务加载 Token 数据
+    
+    Args:
+        tag: 数据标签
+    
+    Returns:
+        str: Token 字符串，失败返回 None
+    """
+    data = get_data(tag)
+    if data and data.get("type") == "token":
+        return data.get("data", {}).get("token")
+    return None