# clouds-storage-sdk

**Repository Path**: snail_tech/clouds-storage-sdk

## Basic Information

- **Project Name**: clouds-storage-sdk
- **Description**: 云存储sdk （类似openlist ），只提供基础的sdk功能
- **Primary Language**: Python
- **License**: GPL-3.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-10-17
- **Last Updated**: 2026-01-05

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 云存储SDK

一个统一的云存储SDK，提供简洁易用的API接口，支持多种云存储平台。

## 🚀 功能特性

- **统一API接口**: 基于抽象基类的统一接口设计，所有云存储平台使用相同的API
- **多平台支持**: 支持天翼云盘TV版，预留百度网盘、阿里云盘、OneDrive、Google Drive等平台接口
- **灵活登录流程**: 支持各平台特有的登录方式，包括二维码、OAuth等
- **完整文件操作**: 上传、下载、删除、移动、重命名等基本操作
- **高级功能**: 文件搜索、快速上传（秒传）、回收站管理等
- **会话管理**: 支持登录状态序列化与恢复

## 📦 安装

```bash
# 从源码安装
git clone https://gitee.com/snail_tech/clouds-storage-sdk.git
cd clouds-storage-sdk
pip install -e .
```

## 🎯 快速开始

### 基本使用示例

```python
from clouds_storage_sdk import CloudStorageClientFactory, PROVIDER_TIANYI_TV

# 1. 创建客户端
client = CloudStorageClientFactory.create_client(PROVIDER_TIANYI_TV)

# 2. 登录流程
# 触发登录
login_url = client.trigger_login()
print(f"请访问链接登录: {login_url}")

# 检测登录状态
import time
while True:
    persistent_info = client.check_login_status()
    if persistent_info:
        print("登录成功！")
        break
    time.sleep(5)


# 检测登录状态
import time
while True:
    persistent_info = client.check_login_status()
    if persistent_info:
        print("登录成功！")
        break
    time.sleep(5)


# 缓存登录
获取登录的缓存 登录凭证字符串
login_token = client.serialize_login_info()
# 适用方可以将登录缓存保存到文件或数据库，以便有序进行登录恢复，避免重复登录
# 使用登录缓存的登录凭证字符串恢复登录状态
client.restore_login_from_string(login_token)
print(f"请访问链接登录: {login_url}")


# 3. 基本文件操作
# 获取账户信息
account_info = client.get_account_info()
print(f"总空间: {account_info.total_space:,} bytes")

# 创建文件夹
folder = client.create_folder('我的测试文件夹')
print(f"创建文件夹: {folder.name} (ID: {folder.file_id})")

# 上传文件
with open('test.txt', 'rb') as f:
    file_data = f.read()
uploaded_file = client.upload_file('test.txt', file_data, folder.file_id)
print(f"上传成功: {uploaded_file.name} (ID: {uploaded_file.file_id})")

# 下载文件
client.download_file(uploaded_file.file_id, 'downloaded_test.txt')
print("下载完成")
```

---

# 🔐 登录指南

本SDK支持多种云存储平台的登录方式，每种平台可能有不同的登录流程。

## 天翼云盘TV版登录

天翼云盘TV版采用二维码扫码登录方式。

### 登录流程

1. **触发登录**
   ```python
   from clouds_storage_sdk import CloudStorageClientFactory, PROVIDER_TIANYI_TV
   
   # 创建客户端
   client = CloudStorageClientFactory.create_client(PROVIDER_TIANYI_TV)
   
   # 触发登录，返回登录授权链接
   login_url = client.trigger_login()
   print(f"请访问链接或扫描二维码登录: {login_url}")
   ```

2. **检测登录状态**
   ```python
   import time
   
   # 循环检测登录状态
   while True:
       persistent_info = client.check_login_status()
       if persistent_info:
           print("登录成功！")
           break
       time.sleep(5)
   ```

3. **保存登录状态**
   ```python
   # 序列化登录信息，保存到文件或数据库
   serialized_info = client.serialize_login_info()
   with open('login_info.json', 'w') as f:
       f.write(serialized_info)
   ```

4. **恢复登录状态**
   ```python
   # 从保存的信息中恢复登录状态
   with open('login_info.json', 'r') as f:
       serialized_info = f.read()
   
   new_client = CloudStorageClientFactory.create_client(PROVIDER_TIANYI_TV)
   if new_client.restore_login_from_string(serialized_info):
       print("登录状态恢复成功")
   ```

5. **验证登录状态**
   ```python
   # 获取账户信息，验证登录状态
   account_info = client.get_account_info()
   print(f"用户名: {account_info.username}")
   print(f"总空间: {account_info.total_space:,} bytes")
   print(f"已用空间: {account_info.used_space:,} bytes")
   ```

### 登录相关API

| 方法 | 描述 | 参数 | 返回值 |
|------|------|------|--------|
| `trigger_login` | 触发登录流程 | `credentials: LoginCredentials = None` | `str` - 登录授权链接 |
| `check_login_status` | 检测登录状态 | 无 | `Optional[PersistentLoginInfo]` |
| `serialize_login_info` | 序列化登录信息 | 无 | `str` |
| `restore_login_from_string` | 恢复登录状态 | `serialized_info: str` | `bool` |
| `get_account_info` | 获取账户信息 | 无 | `AccountInfo` |

## 其他平台登录（预留接口）

SDK已预留其他云存储平台的登录接口，未来版本将支持：

- **百度网盘**: OAuth2.0授权登录
- **阿里云盘**: OAuth2.0授权登录
- **OneDrive**: Microsoft Graph API授权
- **Google Drive**: Google OAuth2.0授权

---

# 📁 文件操作指南

所有云存储平台提供统一的文件操作API，确保代码在不同平台间的一致性。

## 基本文件操作

### 1. 列出文件

```python
# 列出根目录文件
files = client.list_files()
for file in files:
    print(f"{file.name} ({'文件夹' if file.is_folder else '文件'})")

# 列出指定文件夹内容
folder_id = "文件夹ID"
files = client.list_files(folder_id)
```

### 2. 创建文件夹

```python
# 在根目录创建文件夹
folder = client.create_folder('新文件夹')
print(f"创建成功: {folder.name} (ID: {folder.file_id})")

# 在指定文件夹内创建文件夹
parent_folder_id = "父文件夹ID"
subfolder = client.create_folder('子文件夹', parent_folder_id)
```

### 3. 上传文件

```python
# 普通上传
with open('local_file.txt', 'rb') as f:
    file_data = f.read()

# 上传到根目录
uploaded_file = client.upload_file('remote_file.txt', file_data)

# 上传到指定文件夹
folder_id = "目标文件夹ID"
uploaded_file = client.upload_file('remote_file.txt', file_data, folder_id)

# 覆盖上传
uploaded_file = client.upload_file('remote_file.txt', file_data, folder_id, overwrite=True)
```

### 4. 下载文件

```python
# 下载文件到本地
file_id = "要下载的文件ID"
local_path = "保存路径/文件名.txt"
client.download_file(file_id, local_path)
```

### 5. 获取文件信息

```python
# 获取文件详细信息
file_id = "文件ID"
file_info = client.get_file_info(file_id)
print(f"文件名: {file_info.name}")
print(f"大小: {file_info.size} bytes")
print(f"创建时间: {file_info.created_at}")
print(f"修改时间: {file_info.modified_at}")
print(f"MD5: {file_info.md5}")
```

### 6. 重命名文件

```python
# 重命名文件或文件夹
file_id = "文件ID"
new_name = "新名称.txt"
renamed_file = client.rename_file(file_id, new_name)
print(f"重命名成功: {renamed_file.name}")
```

### 7. 移动文件

```python
# 移动文件到另一个文件夹
file_id = "文件ID"
target_folder_id = "目标文件夹ID"
moved_file = client.move_file(file_id, target_folder_id)
print(f"移动成功: {moved_file.name}")
```

### 8. 删除文件

```python
# 删除文件或文件夹
file_id = "文件ID"
client.delete_file(file_id)
print("删除成功")
```

## 高级文件操作

### 1. 文件搜索

```python
# 搜索文件
query = "搜索关键词"
search_results = client.search_files(query)
print(f"找到 {len(search_results)} 个文件")

for file in search_results:
    print(f"{file.name} - {file.path}")
```

### 2. 快速上传（秒传）

```python
# 检查是否可以秒传
file_name = "large_file.zip"
with open(file_name, 'rb') as f:
    file_data = f.read()

if client.check_fast_upload(file_name, file_data):
    # 使用秒传
    fast_uploaded = client.fast_upload_file(file_name, file_data)
    print("秒传成功")
else:
    # 使用普通上传
    uploaded = client.upload_file(file_name, file_data)
    print("普通上传成功")
```

### 3. 获取下载链接

```python
# 获取文件的直接下载链接
file_id = "文件ID"
download_url = client.get_download_url(file_id)
print(f"下载链接: {download_url}")
```

### 4. 回收站操作

```python
# 列出回收站文件
trash_files = client.list_trash()
print(f"回收站文件: {len(trash_files)} 个")

# 清空回收站
client.empty_trash()
print("回收站已清空")
```

## 文件操作API参考

### 基本操作

| 方法 | 描述 | 参数 | 返回值 |
|------|------|------|--------|
| `list_files` | 列出文件 | `folder_id: str = None` | `List[CloudFile]` |
| `create_folder` | 创建文件夹 | `name: str, parent_id: str = None` | `CloudFile` |
| `upload_file` | 上传文件 | `file_name, file_data, parent_id=None, overwrite=False` | `CloudFile` |
| `download_file` | 下载文件 | `file_id: str, local_path: str` | `None` |
| `get_file_info` | 获取文件信息 | `file_id: str, folder_id: str = None` | `CloudFile` |
| `rename_file` | 重命名文件 | `file_id: str, new_name: str` | `CloudFile` |
| `move_file` | 移动文件 | `file_id: str, target_folder_id: str` | `CloudFile` |
| `delete_file` | 删除文件 | `file_id: str` | `None` |

### 高级操作

| 方法 | 描述 | 参数 | 返回值 |
|------|------|------|--------|
| `search_files` | 搜索文件 | `query: str` | `List[CloudFile]` |
| `get_download_url` | 获取下载链接 | `file_id: str` | `str` |
| `check_fast_upload` | 检查快速上传 | `file_name: str, file_data: bytes` | `bool` |
| `fast_upload_file` | 快速上传文件 | `file_name, file_data, parent_id=None` | `CloudFile` |
| `list_trash` | 回收站列表 | 无 | `List[TrashItem]` |
| `empty_trash` | 清空回收站 | 无 | `None` |

---

## 📁 数据模型

### CloudFile
```python
@dataclass
class CloudFile:
    file_id: str          # 文件ID
    name: str            # 文件名
    path: str            # 完整路径
    size: int            # 文件大小（字节）
    is_folder: bool      # 是否为文件夹
    parent_id: Optional[str] = None      # 父文件夹ID
    created_at: Optional[datetime] = None  # 创建时间
    modified_at: Optional[datetime] = None # 修改时间
    md5: Optional[str] = None            # MD5哈希
```

### AccountInfo
```python
@dataclass
class AccountInfo:
    user_id: str         # 用户ID
    username: str        # 用户名
    email: Optional[str] = None         # 邮箱
    phone: Optional[str] = None         # 手机号
    total_space: int     # 总空间（字节）
    used_space: int      # 已用空间（字节）
```

### TrashItem
```python
@dataclass
class TrashItem:
    file_id: str         # 文件ID
    name: str            # 文件名
    path: str            # 原始路径
    size: int            # 文件大小（字节）
    is_folder: bool      # 是否为文件夹
    deleted_at: datetime # 删除时间
```

## 🧪 测试

运行完整测试用例：

```bash
python test_comprehensive_real.py
```

测试用例包含以下功能测试：
- 登录流程测试
- 账户信息获取
- 文件夹创建
- 文件上传
- 文件信息获取
- 文件搜索
- 文件重命名
- 文件移动
- 快速上传（秒传）
- 文件下载
- 文件删除

## 📁 项目结构

```
clouds-storage-sdk/
├── clouds_storage_sdk/           # 核心SDK代码
│   ├── base/                    # 基础抽象类
│   ├── providers/               # 提供商实现
│   │   └── tianyi_tv/          # 天翼云盘TV版实现
│   ├── factory.py               # 工厂模式
│   └── constants.py             # 常量定义
├── test_comprehensive_real.py   # 综合测试用例
└── README.md                   # 使用说明
```

## 📋 版本信息

- **当前版本**: v0.4.0
- **支持平台**: 天翼云盘TV版
- **Python版本**: 3.8+

## 📄 许可证

MIT License