Welcome to py115’s documentation!
API Reference
Module contents
- py115.connect(credential=None, protocol_kwargs=None)
Connect to 115 cloud.
- Parameters:
credential (py115.types.Credential) – Credential data to identity user.
protocol_kwargs (dict) – Keyword arguments for underlying protocol client.
- Returns:
Cloud instance.
- Return type:
Submodules
py115.cloud module
- class py115.cloud.Cloud(credential=None, protocol_kwargs=None)
Bases:
_Handler115 cloud service.
- Parameters:
credential (py115.types.Credential) – Credential object.
protocol_kwargs (dict) – Settings for underlying protocol client.
- import_credential(credential)
Import credential to cloud instance.
- Parameters:
credential (py115.types.Credential) – Credential object to identity user.
- Returns:
Is credential valid.
- Return type:
bool
- export_credentail()
Export current credentail from cloud instance.
- Returns:
Credential object, or None when credential is invalid.
- Return type:
- qrcode_login(app_type)
Start QRcode login session.
- Parameters:
app_type (py115.types.AppType) – App to login.
- Returns:
QRcode login session.
- Return type:
- offline()
Get offline service.
- Returns:
Offline service instance.
- Return type:
- storage()
Get storage service.
- Returns:
Storage service instance.
- Return type:
py115.login module
py115.services module
- class py115.services.OfflineService
Bases:
objectOffline task manager.
- list()
Get all tasks.
- Yields:
py115.types.Task – Task object
- Return type:
Generator[Task, None, None]
- add_url(*urls)
Create task(s) from download URL.
- Parameters:
*urls (str) – Download URL, can be a http/ftp/ed2k/magnet link.
- Returns:
Task list for the download URLs.
- Return type:
Iterable[py115.types.Task]
- delete(*task_ids)
Delete task(s).
- Parameters:
*task_ids (str) – The ID of tasks you wants to delete.
- clear(status=TaskStatus.Complete)
Clear tasks.
- Parameters:
status (py115.types.TaskStatus) – Tasks in given status will be cleared. Set status to None to clear all tasks.
- class py115.services.StorageService
Bases:
objectCloud file/directory manager.
- space()
Get total size and used size of the storage.
- Returns:
Total size and used size in byte.
- Return type:
Tuple[int, int]
- list(dir_id='0')
Get files under a directory.
- Parameters:
dir_id (str) – Directory ID to list, default is “0” which means root directory.
- Yields:
py115.types.File – File object under the directory.
- Return type:
Generator[File, None, None]
- search(keyword, dir_id='0')
Recursively search files under a directory.
- Parameters:
keyword (str) – Keyword to search files.
dir_id (str) – Directory ID to search.
- Yields:
py115.types.File – File object whose name contains the keyword.
- move(target_dir_id, *file_ids)
Move files to a directory.
- Parameters:
target_dir_id (str) – ID of target directory where to move files.
*file_ids (str) – ID of files to be moved.
- rename(file_id, new_name)
Rename file.
- Parameters:
file_id (str) – ID of file to be renamed.
new_name (str) – New name for the file.
- delete(*file_ids)
Delete files.
- Parameters:
*file_ids (str) – ID of files to be deleted.
- make_dir(parent_id, name)
Make new directory under a directory.
- Parameters:
parent_id (str) – ID of parent directory where to make new directory.
name (str) – Name for the new directory.
- Returns:
File object of the created directory.
- Return type:
- request_download(pickcode)
Download file from cloud storage.
- Parameters:
pickcode (str) – Pick code of file.
- Returns:
A ticket contains all required fields to download file from cloud.
- Return type:
- request_play(pickcode)
Play a video file on cloud, returns required parameters as a ticket.
- Parameters:
pickcode (str) – Pick code of file.
- Returns:
A ticket contains all required fields to play the media file on cloud.
- Return type:
- request_upload(dir_id, file_path)
Upload local file to cloud storage.
- Parameters:
dir_id (str) – ID of directory where to store the file.
file_path (str) – Path of the local file.
- Returns:
A ticket contains all required fields to upload file to cloud, should be used with aliyun-oss-python-sdk.
- Return type:
- request_upload_data(dir_id, save_name, data_io)
Upload data as a file to cloud storage.
- Parameters:
dir_id (str) – ID of directory where to store the file.
save_name (str) – File name to be saved.
data_io (BinaryIO) – IO stream of data.
- Returns:
A ticket contains all required fields to upload file to cloud, should be used with aliyun-oss-python-sdk.
- Return type:
py115.types module
- class py115.types.AppType(value)
Bases:
IntEnumApp to login.
- Web = 0
Login as web
- Mac = 1
Login as MAC app
- Linux = 2
Login as Linux app
- Windows = 3
Login as Windows app
- class py115.types.Credential(uid=None, cid=None, seid=None)
Bases:
_BaseCredential contains required information to identify user.
- Parameters:
uid (str) – The “UID” value in cookies.
cid (str) – The “CID” value in cookies.
seid (str) – The “SEID” value in cookies.
- classmethod from_dict(d)
Create Credential from a dict object.
- Parameters:
d (dict) – Dict object.
- Returns:
Credential object.
- Return type:
- to_dict()
Convert credential object to dict object.
- Returns:
Dict object.
- Return type:
dict
- class py115.types.File
Bases:
_BaseFile represents a cloud file or directory.
- file_id: str
Unique ID of the file on cloud.
- parent_id: str
File ID of the parent directory.
- name: str
Base name.
- size: int
Size in bytes.
- modified_time: datetime
Last modified datetime.
- sha1: str
SHA-1 hash of the file in HEX encoding.
- pickcode: str
Pickcode to download the file.
- is_dir: bool
Is file a directory.
- class py115.types.DownloadTicket
Bases:
_BaseDownloadTicket contains required parameters to download a file from cloud storage.
Please check examples for detail usage.
- file_name: str
Base name of the file.
- file_size: int
Size of the file.
- url: str
Download URL.
- headers: dict
Required headers that should be used with download URL.
- class py115.types.PlayTicket
Bases:
_Base- url: str
Download URL.
- headers: dict
Required headers that should be used with download URL.
- class py115.types.UploadTicket
Bases:
_BaseUploadTicket contains required parameters to upload a file to cloud storage.
Please check examples for detial usage.
- is_done: bool
Is file already uploaded.
- oss_endpoint: str
OSS endpoint address.
- oss_token: dict
OSS token
- bucket_name: str
OSS bucket name.
- object_key: str
OSS object key.
- headers: dict
Required headers that should be used in upload.
- expiration: datetime
Expiration time of this ticket.
- is_valid()
Check whether the ticket is valid.
- Returns:
Valid flag.
- Return type:
bool
- class py115.types.TaskStatus(value)
Bases:
IntEnumAn enumeration.
- Running = 1
Task is running.
- Complete = 2
Task is complete.
- Failed = -1
Task is failed.
- Unknown = 0
Unknown status?
- class py115.types.Task
Bases:
_BaseTask represents an offline task.
- task_id: str
Unique ID of the task.
- name: str
Task name.
- size: int
Total size to be downloaded.
- created_time: datetime
Task created time.
- percent: float
Precentage of the download, 0~100.
- file_id: str
Downloaded file ID of the task, may be None if the task does not finish.
- file_is_dir: bool
Is the downloaded file a directory.
- status: TaskStatus
Task status.
- is_complete()
Check is task complete.
- Return type:
bool
- is_failed()
Check is task failed.
- Return type:
bool
- is_running()
Check is task still running.
- Return type:
bool
Examples
QRcode Login
import io
import typing
from PIL import Image
import py115
from py115.types import LoginPlatform
class QRcode:
_char_dict = ['\u2588', '\u2584', '\u2580', ' ']
_matrix: typing.List[int]
_width: int
_height: int
@staticmethod
def _is_black(c: int) -> bool:
return c < 128
def __init__(self, img_io: typing.BinaryIO) -> None:
with Image.open(img_io).convert('L') as img:
self._parse(img)
def _parse(self, img: Image.Image):
width, height = img.size
# Locate QRcode area
left, right, upper, lower = -1, -1, -1, -1
for y in range(height):
for x in range(width):
if QRcode._is_black(img.getpixel((x, y))):
left, upper = x, y
break
if left > 0: break
for x in range(width - 1, left, -1):
if QRcode._is_black(img.getpixel((x, upper))):
right = x
break
for y in range(height - 1, upper, -1):
if QRcode._is_black(img.getpixel((left, y))):
lower = y
break
# Measure dot size
dot_size = 0
for step in range(0, right - left + 1):
pt = img.getpixel((left + step, upper + step))
if not QRcode._is_black(pt):
dot_size = step
break
# Parse QRcode as matrix
self._matrix = []
self._width = int((right - left + 1) / dot_size)
self._height = int((lower - upper + 1) / dot_size)
for y in range(self._height):
for x in range(self._width):
pt = img.getpixel((
left + x * dot_size,
upper + y * dot_size
))
self._matrix.append(
1 if QRcode._is_black(pt) else 0
)
def _get_point(self, x: int, y: int) -> int:
if x < 0 or x >= self._width or y < 0 or y >= self._height:
return 0
return self._matrix[y * self._width + x]
def to_ascii(self, margin: int = 3) -> str:
buf = []
for y in range(0, self._height + margin * 2, 2):
for x in range(0, self._width + margin * 2):
dot_1 = self._get_point(x - margin, y - margin)
dot_2 = self._get_point(x - margin, y + 1 - margin)
code = dot_2 * 2 + dot_1
buf.append(QRcode._char_dict[code])
buf.append('\n')
return ''.join(buf)
if __name__ == '__main__':
cloud = py115.connect()
print('Start QRcode login ...')
session = cloud.qrcode_login(LoginPlatform.Linux)
qr = QRcode(io.BytesIO(session.image_data))
print(f'Please scan QRcode: \n{qr.to_ascii()}')
if session.poll():
print('Login succeeded!')
else:
print('Login failed!')
Common Operations
Connect to Cloud
import py115
import py115.types
cloud = py115.connect(
credential=py115.types.Credential(
uid='', cid='', seid=''
)
)
# Get storage service
storage = cloud.storage()
# Get offline service
offline = cloud.offline()
List Files
# List files under root directory
for file in storage.list(dir_id='0'):
print(f'File: ID={file.file_id}, Name={file.name}')
Download File
import subprocess
# Request download ticket
ticket = storage.request_download(pickcode='pickcode-of-file')
# Download via curl command
args = [
'curl', ticket.url,
'-o', ticket.file_name,
]
for name, value in ticket.headers.items():
args.extend([
'-H', f'{name}: {value}'
])
subprocess.call(args)
Upload File
# File to upload
file_path = '/path/to/local-file'
# Request upload ticket
ticket = storage.request_upload(
dir_id='0',
file_path=file_path
)
if ticket is None:
print('Request upload failed!')
elif ticket.is_done:
print('File has been imported to your cloud!')
else:
# Upload via "aliyun-oss-python-sdk"
import oss2
# Create OSS auth
auth = oss2.StsAuth(**ticket.oss_token)
# Get bucket object
bucket = oss2.Bucket(
auth=auth,
endpoint=ticket.oss_endpoint,
bucket_name=ticket.bucket_name,
)
# Upload file, that may take a looooong time
por = bucket.put_object_from_file(
key=ticket.object_key,
filename=file_path,
headers=ticket.headers, # DO NOT forget this!!!
)
# Parse result
result = por.resp.response.json()
print(f'Upload result: {result!r}')
List Tasks
# List all offline tasks
for task in offline.list():
print(f'Task: ID={task.task_id}, Name={task.name}')
Add Task
# Add offline task from download URL
# Support HTTP/HTTPS/FTP/magnet/ed2k link
offline.add_url(
'ed2k://ed2k-file-link',
'magnet:?xt=urn:btih:magnet-file-link'
)