129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
封装 Cloudreve 客户端
|
|
"""
|
|
|
|
from typing import Generator
|
|
from urllib.parse import quote
|
|
|
|
from authenticator import Authenticator
|
|
from request import Request
|
|
|
|
|
|
class Cloudreve:
|
|
"""Cloudreve 客户端"""
|
|
|
|
def __init__(self):
|
|
# 实例化认证器
|
|
self.authenticator = Authenticator()
|
|
# 实例化请求客户端
|
|
self.http_client = Request()
|
|
|
|
def _get_upload_session(self, uri: str, size: int) -> str:
|
|
"""
|
|
获取上传会话标识
|
|
:param uri: 统一资源标识符
|
|
:param size: 文件大小
|
|
:return: 上传会话标识
|
|
"""
|
|
response = self.http_client.put(
|
|
url=f"https://cloudreve.liubiren.cloud/api/v4/file/upload",
|
|
headers={
|
|
"Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
},
|
|
json={
|
|
"uri": quote(string=uri, safe=":/?&="), # 编码统一资源标识
|
|
"size": size,
|
|
"policy_id": "zpHb", # https://cloudreve.liubiren.cloud/api/v4/file?uri=cloudreve://my/转直链 可查看该文件夹储存策略
|
|
},
|
|
)
|
|
# 若非响应成功则抛出异常
|
|
if not response["code"] == 0:
|
|
raise RuntimeError("创建上传会话发生异常")
|
|
return response["data"]["session_id"]
|
|
|
|
def _upload_file_chunk(self, session_id: str, index: int, chunk: bytes) -> None:
|
|
"""
|
|
上传文件块
|
|
:param session_id: 上传会话标识
|
|
:param index: 文件块索引
|
|
:param chunk: 文件块数据
|
|
:return: None
|
|
"""
|
|
response = self.http_client.post(
|
|
url=f"https://cloudreve.liubiren.cloud/api/v4/file/upload/{session_id}/{index}",
|
|
headers={
|
|
"Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
|
|
"Content-Type": "application/octet-stream",
|
|
"Content-Length": str(len(chunk)),
|
|
},
|
|
data=chunk,
|
|
)
|
|
if not response["code"] == 0:
|
|
raise RuntimeError("上传文件块发生异常")
|
|
|
|
def _upload_file(self, uri: str, size: int, generator: Generator) -> None:
|
|
"""
|
|
上传文件
|
|
:param uri: 统一资源标识符
|
|
:param size: 文件大小
|
|
:param generator: 文件块生成器
|
|
:return: None
|
|
"""
|
|
# 获取上传会话标识
|
|
session_id = self._get_upload_session(uri=uri, size=size)
|
|
for index, chunk in enumerate(generator):
|
|
# 上传文件块
|
|
self._upload_file_chunk(session_id=session_id, index=index, chunk=chunk)
|
|
|
|
def _create_direct_link(self, uri: str) -> str:
|
|
"""
|
|
创建直链
|
|
:param uri: 统一资源标识符
|
|
:return: 直链
|
|
"""
|
|
response = self.http_client.put(
|
|
url="https://cloudreve.liubiren.cloud/api/v4/file/source",
|
|
headers={
|
|
"Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
},
|
|
json={
|
|
"uris": [quote(string=uri, safe=":/?&=")], # 编码统一资源标识
|
|
},
|
|
)
|
|
if not response["code"] == 0:
|
|
raise RuntimeError("获取直链发生异常")
|
|
return response["data"][0]["link"]
|
|
|
|
def get_direct_link(self, uri: str, size: int, generator: Generator) -> str:
|
|
"""
|
|
获取直链
|
|
:param uri: 统一资源标识符
|
|
:return: 直链
|
|
"""
|
|
response = self.http_client.get(
|
|
url=f"https://cloudreve.liubiren.cloud/api/v4/file/info",
|
|
headers={
|
|
"Authorization": f"Bearer {self.authenticator.get_token(servicer="cloudreve")}",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
},
|
|
params={
|
|
"uri": quote(string=uri, safe=":/?&="), # 编码统一资源标识
|
|
"extended": "true", # 获取文件扩展信息
|
|
},
|
|
)
|
|
# 若文件已存在则返回直链，否则上传文件并创建直连
|
|
if response["code"] == 0:
|
|
return response["data"]["extended_info"]["direct_links"][0]["url"]
|
|
# 上传文件
|
|
self._upload_file(
|
|
generator=generator,
|
|
uri=uri,
|
|
size=size,
|
|
)
|
|
# 创建直连
|
|
direct_link = self._create_direct_link(uri=uri)
|
|
return direct_link
|