data-sync-service/clients/client-example.py

"""
通用数据同步服务 - Python 客户端示例
演示如何在 Python 脚本中使用同步服务
"""

import sys
sys.path.append('/home/hermes/projects/data-sync-service/scripts')

from utils import (
    upload_data, 
    get_data, 
    delete_data, 
    list_all_tags,
    load_cookies_from_sync,
    load_token_from_sync
)


def example_upload_cookies():
    """示例：上传 Cookie 数据"""
    
    # 模拟 Cookie 数据
    cookies_data = {
        "cookies": [
            {
                "name": "session_id",
                "value": "abc123xyz789",
                "domain": ".example.com",
                "path": "/"
            },
            {
                "name": "auth_token",
                "value": "token_value_here",
                "domain": ".example.com",
                "path": "/"
            }
        ],
        "raw_cookie": "session_id=abc123xyz789; auth_token=token_value_here",
        "user_agent": "Mozilla/5.0 ...",
        "url": "https://example.com"
    }
    
    success = upload_data(
        tag="example:user@example.com",
        data_type="cookie",
        data=cookies_data,
        expires_in=604800,  # 7天
        note="示例账号",
        login_url="https://example.com/login"
    )
    
    if success:
        print("✅ Cookie 上传成功")
    else:
        print("❌ Cookie 上传失败")


def example_upload_token():
    """示例：上传 Token 数据"""
    
    token_data = {
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
        "token_type": "Bearer",
        "api_endpoint": "https://api.example.com/v1"
    }
    
    success = upload_data(
        tag="api:service_account",
        data_type="token",
        data=token_data,
        expires_in=2592000,  # 30天
        note="API 服务账号",
        login_url="https://console.example.com"
    )
    
    if success:
        print("✅ Token 上传成功")
    else:
        print("❌ Token 上传失败")


def example_get_data():
    """示例：获取数据"""
    
    data = get_data("example:user@example.com")
    
    if data:
        print("✅ 数据获取成功:")
        print(f"  Tag: {data.get('tag')}")
        print(f"  Type: {data.get('type')}")
        print(f"  Created: {data.get('metadata', {}).get('created_at')}")
        print(f"  Expires: {data.get('metadata', {}).get('expires_at')}")
    else:
        print("❌ 数据不存在或获取失败")


def example_load_cookies():
    """示例：加载 Cookie 数据（便捷方法）"""
    
    cookies = load_cookies_from_sync("tingwu:ykaayk@qq.com")
    
    if cookies:
        print("✅ Cookie 加载成功:")
        for cookie in cookies:
            print(f"  - {cookie['name']}: {cookie['value'][:20]}...")
    else:
        print("❌ Cookie 加载失败")


def example_list_all():
    """示例：列出所有 tag"""
    
    tags = list_all_tags()
    
    print(f"📋 共有 {len(tags)} 个数据:")
    for item in tags:
        status_icon = {
            "valid": "✅",
            "expiring_soon": "⚠️",
            "expired": "❌",
            "no_expiry": "♾️"
        }.get(item.get("status", "unknown"), "❓")
        
        print(f"  {status_icon} {item['tag']} ({item['type']})")


def example_delete():
    """示例：删除数据"""
    
    success = delete_data("example:user@example.com")
    
    if success:
        print("✅ 数据删除成功")
    else:
        print("❌ 数据删除失败")


def example_tingwu_script():
    """
    示例：在通义听悟脚本中使用同步服务
    替代原有的本地文件读取方式
    """
    
    # 尝试从同步服务加载 Cookie
    cookies = load_cookies_from_sync("tingwu:ykaayk@qq.com")
    
    if cookies:
        print("✅ 从同步服务加载 Cookie 成功")
        
        # 转换为 requests 可用的格式
        cookie_dict = {c['name']: c['value'] for c in cookies}
        
        # 使用 Cookie 发起请求
        import requests
        try:
            response = requests.get(
                "https://tingwu.aliyun.com/api/some-endpoint",
                cookies=cookie_dict,
                timeout=10
            )
            print(f"  API 调用成功: {response.status_code}")
        except Exception as e:
            print(f"  API 调用失败: {e}")
    else:
        print("❌ Cookie 加载失败，请检查同步服务")


def example_with_fallback():
    """
    示例：带降级方案的使用方式
    优先从同步服务加载，失败时使用本地文件
    """
    
    # 尝试从同步服务加载
    data = get_data("tingwu:ykaayk@qq.com")
    
    if data:
        print("✅ 从同步服务加载成功")
        cookies = data.get("data", {}).get("cookies", [])
    else:
        print("⚠️ 同步服务不可用，使用本地文件")
        # 降级到本地文件
        import json
        try:
            with open("/path/to/local/cookies.json", 'r') as f:
                local_data = json.load(f)
                cookies = local_data.get("cookies", [])
        except:
            print("❌ 本地文件也加载失败")
            cookies = []
    
    return cookies


if __name__ == "__main__":
    print("=" * 50)
    print("通用数据同步服务 - Python 客户端示例")
    print("=" * 50)
    
    # 运行示例
    print("\n1️⃣ 上传 Cookie 数据")
    example_upload_cookies()
    
    print("\n2️⃣ 上传 Token 数据")
    example_upload_token()
    
    print("\n3️⃣ 获取数据")
    example_get_data()
    
    print("\n4️⃣ 列出所有 tag")
    example_list_all()
    
    print("\n5️⃣ 加载 Cookie（便捷方法）")
    example_load_cookies()
    
    print("\n6️⃣ 通义听悟脚本示例")
    example_tingwu_script()
    
    # 删除示例数据（取消注释以执行）
    # print("\n7️⃣ 删除数据")
    # example_delete()
    
    print("\n" + "=" * 50)