Tubearchivist RESTful API使用示例：Python与JavaScript调用教程

引言：解决自托管YouTube媒体库的API集成痛点

你是否正在寻找一种高效方式来管理自托管的YouTube媒体库？Tubearchivist作为开源的媒体服务器解决方案，提供了强大的RESTful API接口，让开发者可以轻松实现视频管理、频道订阅、下载队列等功能的自动化。本文将通过Python和JavaScript两种语言的实战示例，带你全面掌握Tubearchivist API的使用方法，从认证流程到高级功能调用，一次解决所有集成难题。

读完本文你将获得：

• 完整的API认证与授权实现方案
• 10+常用API端点的调用示例代码
• Python/JavaScript两种语言的最佳实践
• 错误处理与性能优化的实用技巧
• 可直接复用的项目集成模板

API架构概览

系统架构图

核心API端点概览

/api/user/login//api/user/logout//api/video//api/video/{video_id}//api/video/{video_id}/progress//api/channel//api/channel/{channel_id}//api/download//api/download/{video_id}//api/playlist/

认证机制详解

Tubearchivist采用基于Session Cookie的认证机制，配合CSRF令牌保护，确保API调用的安全性。认证流程如下：

Python认证实现

import requests

BASE_URL = "http://localhost:8000/api"
LOGIN_ENDPOINT = f"{BASE_URL}/user/login/"
VIDEO_LIST_ENDPOINT = f"{BASE_URL}/video/"

# 1. 登录获取会话
session = requests.Session()
login_data = {
    "username": "your_username",
    "password": "your_password",
    "remember_me": "on"
}
response = session.post(LOGIN_ENDPOINT, data=login_data)
response.raise_for_status()  # 检查登录是否成功

# 2. 获取CSRF令牌（从cookie中提取）
csrf_token = session.cookies.get("csrftoken")
headers = {"X-CSRFToken": csrf_token}

# 3. 调用需要认证的API
response = session.get(VIDEO_LIST_ENDPOINT, headers=headers)
videos = response.json()
print(f"获取到 {len(videos['data'])} 个视频")

JavaScript认证实现

const BASE_URL = "http://localhost:8000/api";
const LOGIN_ENDPOINT = `${BASE_URL}/user/login/`;
const VIDEO_LIST_ENDPOINT = `${BASE_URL}/video/`;

// 1. 获取CSRF令牌（从cookie中提取）
function getCookie(name) {
    const value = `; ${document.cookie}`;
    const parts = value.split(`; ${name}=`);
    if (parts.length === 2) return parts.pop().split(';').shift();
}

// 2. 登录获取会话
async function login(username, password) {
    const csrfToken = getCookie('csrftoken');
    const response = await fetch(LOGIN_ENDPOINT, {
        method: 'POST',
        headers: {
            'Content-Type': 'application/x-www-form-urlencoded',
            'X-CSRFToken': csrfToken
        },
        body: new URLSearchParams({
            'username': username,
            'password': password,
            'remember_me': 'on'
        }),
        credentials: 'include' // 保存会话cookie
    });
    
    if (!response.ok) throw new Error('登录失败');
    return true;
}

// 3. 调用需要认证的API
async function getVideoList() {
    const csrfToken = getCookie('csrftoken');
    const response = await fetch(VIDEO_LIST_ENDPOINT, {
        method: 'GET',
        headers: {
            'X-CSRFToken': csrfToken
        },
        credentials: 'include' // 发送会话cookie
    });
    
    if (!response.ok) throw new Error('获取视频列表失败');
    return response.json();
}

// 使用示例
login('your_username', 'your_password')
    .then(() => getVideoList())
    .then(videos => console.log('视频列表:', videos))
    .catch(error => console.error('错误:', error));

核心API调用示例

1. 视频管理API

获取视频列表（带过滤和分页）

# Python示例
params = {
    "sort": "published",
    "order": "desc",
    "type": "videos",
    "page": 1
}
response = session.get(VIDEO_LIST_ENDPOINT, headers=headers, params=params)
video_list = response.json()

# 响应结构解析
print(f"总视频数: {video_list['paginate']['total']}")
print(f"当前页: {video_list['paginate']['current_page']}")
print(f"每页数量: {video_list['paginate']['per_page']}")

# 打印第一个视频信息
if video_list['data']:
    first_video = video_list['data'][0]
    print(f"标题: {first_video['title']}")
    print(f"观看次数: {first_video['stats']['view_count']}")
    print(f"发布日期: {first_video['published']}")

// JavaScript示例
async function getFilteredVideos() {
    const params = new URLSearchParams({
        sort: 'published',
        order: 'desc',
        type: 'videos',
        page: 1
    });
    
    const url = `${VIDEO_LIST_ENDPOINT}?${params}`;
    const response = await fetch(url, {
        headers: { 'X-CSRFToken': getCookie('csrftoken') },
        credentials: 'include'
    });
    
    const data = await response.json();
    console.log(`总视频数: ${data.paginate.total}`);
    return data;
}

获取视频详情

# Python示例
VIDEO_DETAIL_ENDPOINT = f"{BASE_URL}/video/{video_id}/"

response = session.get(VIDEO_DETAIL_ENDPOINT, headers=headers)
video_detail = response.json()

# 打印视频详细信息
print(f"标题: {video_detail['title']}")
print(f"描述: {video_detail['description']}")
print(f"时长: {video_detail['player']['duration_str']}")
print(f"分辨率: {video_detail['streams'][0]['width']}x{video_detail['streams'][0]['height']}")
print(f"文件大小: {video_detail['media_size']} bytes")

更新视频观看进度

# Python示例
PROGRESS_ENDPOINT = f"{BASE_URL}/video/{video_id}/progress/"

progress_data = {
    "position": 360.5  # 秒
}
response = session.post(PROGRESS_ENDPOINT, headers=headers, json=progress_data)
response.raise_for_status()
print("进度更新成功")

2. 频道管理API

获取频道列表

# Python示例
CHANNEL_LIST_ENDPOINT = f"{BASE_URL}/channel/"

response = session.get(CHANNEL_LIST_ENDPOINT, headers=headers)
channels = response.json()

# 打印所有订阅频道
for channel in channels['data']:
    print(f"频道名称: {channel['channel_name']}")
    print(f"订阅状态: {'已订阅' if channel['channel_subscribed'] else '未订阅'}")
    print(f"视频数量: {channel['video_count']}")

3. 下载队列管理

添加视频到下载队列

# Python示例
DOWNLOAD_ENDPOINT = f"{BASE_URL}/download/"

download_data = {
    "data": [
        {"youtube_id": "dQw4w9WgXcQ", "status": "pending"}
    ]
}
response = session.post(DOWNLOAD_ENDPOINT, headers=headers, json=download_data)
response.raise_for_status()
print("视频已添加到下载队列")

获取下载队列状态

# Python示例
response = session.get(DOWNLOAD_ENDPOINT, headers=headers)
download_queue = response.json()

# 打印队列状态
print(f"等待下载: {download_queue['paginate']['total']}个视频")
for item in download_queue['data']:
    print(f"标题: {item['title']}")
    print(f"状态: {item['status']}")
    print(f"频道: {item['channel_name']}")

高级应用：构建视频管理工具

下面是一个综合示例，展示如何使用Tubearchivist API构建一个简单的视频管理工具，实现视频搜索、观看进度跟踪和批量下载功能。

Python工具示例

class TubeManager:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.session = requests.Session()
        self.headers = {}
        self.login(username, password)
    
    def login(self, username, password):
        """登录并初始化会话"""
        login_url = f"{self.base_url}/user/login/"
        data = {
            "username": username,
            "password": password,
            "remember_me": "on"
        }
        response = self.session.post(login_url, data=data)
        response.raise_for_status()
        self.headers["X-CSRFToken"] = self.session.cookies.get("csrftoken")
    
    def search_videos(self, query, sort="published", order="desc"):
        """搜索视频"""
        search_url = f"{self.base_url}/video/"
        params = {"q": query, "sort": sort, "order": order}
        response = self.session.get(search_url, headers=self.headers, params=params)
        return response.json()
    
    def track_progress(self, video_id, position):
        """更新视频观看进度"""
        progress_url = f"{self.base_url}/video/{video_id}/progress/"
        data = {"position": position}
        response = self.session.post(progress_url, headers=self.headers, json=data)
        response.raise_for_status()
        return True
    
    def batch_download(self, video_ids):
        """批量添加视频到下载队列"""
        download_url = f"{self.base_url}/download/"
        data = {
            "data": [{"youtube_id": vid, "status": "pending"} for vid in video_ids]
        }
        response = self.session.post(download_url, headers=self.headers, json=data)
        response.raise_for_status()
        return True

# 使用示例
if __name__ == "__main__":
    manager = TubeManager(
        base_url="http://localhost:8000/api",
        username="your_username",
        password="your_password"
    )
    
    # 搜索视频
    results = manager.search_videos("python tutorial")
    print(f"搜索结果: {results['paginate']['total']}个视频")
    
    # 跟踪第一个视频的进度
    if results['data']:
        first_video = results['data'][0]
        manager.track_progress(first_video['youtube_id'], 120)
        print(f"已更新视频进度: {first_video['title']}")
    
    # 批量下载前3个视频
    video_ids = [item['youtube_id'] for item in results['data'][:3]]
    manager.batch_download(video_ids)
    print(f"已添加{len(video_ids)}个视频到下载队列")

错误处理与最佳实践

常见错误及解决方案

性能优化建议

1. 批量操作：尽量使用批量API减少请求次数，如批量添加下载任务
2. 合理分页：使用page参数分页获取大量数据，避免一次性加载过多内容
3. 缓存结果：对不常变化的数据进行本地缓存，如视频元信息
4. 异步处理：长时间运行的操作（如下载）采用异步方式，通过轮询获取状态

总结与展望

Tubearchivist的RESTful API为开发者提供了强大的媒体库管理能力，通过本文介绍的认证机制和API调用示例，你可以轻松实现自定义的媒体管理工具。无论是构建自动化脚本、移动应用还是第三方集成，Tubearchivist API都能满足你的需求。

未来，随着Tubearchivist的不断发展，API可能会增加更多高级功能，如视频转码、自定义元数据管理等。建议开发者关注项目的更新日志，及时了解新功能和API变化。

最后，如果你在使用API过程中遇到问题，可以查阅项目的官方文档或提交issue寻求帮助。祝你开发愉快！

附录：API参考速查表

视频API参数速查

认证头信息速查