data-sync-service/scripts/utils.py

"""
通用数据同步服务 - 工具函数库
"""

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