p123client

https://p123client.readthedocs.io/
Admin
p123client is a Python client module for 123 Cloud Drive that provides direct interface wrappers...
Tokens:4,039
Snippets:35
Trust Score:9.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install p123client

Source: https://p123client.readthedocs.io

Installation commands for the library via pip or GitHub.

```bash
pip install -U p123client
```

```bash
pip install -U git+https://github.com/ChenyangGao/p123client@main
```

--------------------------------

### Create Client Instance with Credentials

Source: https://p123client.readthedocs.io

Initialize the client using account passport and password.

```python
# TODO: 手机号或者邮箱
passport = "..." 
# TODO: 密码
password = "..."

client = P123Client(passport=passport, password=passport)
```

```python
client = P123Client(passport, passport)
```

--------------------------------

### Create Client Instance with Token

Source: https://p123client.readthedocs.io

Initialize the client using an access token, either directly or from a file path.

```python
# TODO: 访问令牌
token = "..."

client = P123Client(token=token)
```

```python
client = P123Client(token)
```

```python
from pathlib import Path

client = P123Client(token=Path("~/123-token.txt").expanduser())
```

```python
client = P123Client(Path("~/123-token.txt").expanduser())
```

--------------------------------

### Create Client Instance with Developer Credentials

Source: https://p123client.readthedocs.io

Initialize the client using client_id and client_secret for developer applications.

```python
# TODO: 应用标识，创建应用时分配的 appId
client_id = "..." 
# TODO: 应用密钥，创建应用时分配的 secretId
client_secret = "..."

client = P123Client(client_id=client_id, client_secret=client_secret)
```

```python
client = P123Client(client_id, client_secret)
```

--------------------------------

### Create Client Instance with Refresh Token

Source: https://p123client.readthedocs.io

Initialize the client using a refresh token. Note that this is restricted to open interfaces.

```python
# TODO: 刷新令牌
refresh_token = "..."

client = P123Client(refresh_token=refresh_token)
```

```python
client = P123Client(refresh_token)
```

--------------------------------

### QR Code Login

Source: https://p123client.readthedocs.io

Initialize the client without arguments to trigger QR code login.

```python
client = P123Client()
```

--------------------------------

### Import Module

Source: https://p123client.readthedocs.io

Basic import statement for the library.

```python
from p123client import P123Client
```

--------------------------------

### Import P123Client Utility Tools

Source: https://p123client.readthedocs.io

Access the collection of helper functions provided by the library.

```python
from p123client import tool
```

--------------------------------

### Perform Synchronous and Asynchronous API Calls

Source: https://p123client.readthedocs.io

Demonstrates the basic syntax for invoking API methods using either synchronous or asynchronous patterns.

```python
# 同步调用
client.method(payload)
client.method(payload, async_=False)

# 异步调用
await client.method(payload, async_=True)
```

--------------------------------

### 获取下载直链

Source: https://p123client.readthedocs.io

根据自定义 URI 获取文件的下载直链，支持带或不带 s3_key_flag 的情况。

```python
from p123client import P123Client
from p123client.tool import get_downurl

# TODO: 改成你自己的账户和密码
client = P123Client(passport="手机号或邮箱", password="登录密码")

# 带 s3_key_flag
print(get_downurl(client, "123://torrentgalaxy.db|1976025090|582aa8bfb0ad8e6f512d9661f6243bdd?1812602326-0"))
# 不带 s3_key_flag（会转存）
print(get_downurl(client, "123://torrentgalaxy.db|1976025090|582aa8bfb0ad8e6f512d9661f6243bdd"))
```

--------------------------------

### Login Another Client

Source: https://p123client.readthedocs.io

Use an existing token to authorize a new client instance or perform third-party OAuth mounting.

```python
client_new = client.login_another()
```

```python
import requests
from urllib.parse import parse_qsl, urlsplit

resp = client.login_oauth_authorize({
    "client_id": "7c278c60da9e43848c45ff8e6fa9da0a", 
    "redirect_uri": "https://redirect123pan.zhenyunpan.com/redirect_pan123", 
    "accessToken": client.token, 
    "state": "http://localhost:19798", 
})
with requests.get(resp["url"], allow_redirects=False, stream=True) as response:
    resp = dict(parse_qsl(urlsplit(response.headers["location"]).query))
client_new = P123Client(token=resp["access_token"], refresh_token=resp["refresh_token"])
```

--------------------------------

### 搭建直链服务

Source: https://p123client.readthedocs.io

使用 blacksheep 框架搭建一个简单的 HTTP 服务来获取文件直链。

```bash
pip install 'blacksheep[uvicorn]'
```

```python
from blacksheep import json, redirect, Application, Request
from p123client import P123Client
from p123client.tool import get_downurl

# TODO: 改成你自己的账户和密码
client = P123Client(passport="", password="")

app = Application(show_error_details=__debug__)

@app.router.route("/{path:uri}", methods=["GET", "HEAD"])
async def index(request: Request, uri: str):
    try:
        payload = int(uri)
    except ValueError:
        if uri.count("|") < 2:
            return json({"state": False, "message": f"bad uri: {uri!r}"}, 500)
        payload = uri
        if s3_key_flag := request.url.query:
            payload += "?" + s3_key_flag.decode("ascii")
    url = await get_downurl(client, payload, quoted=False, async_=True)
    return redirect(url)

if __name__ == "__main__":
    from uvicorn import run

    run(app, host="0.0.0.0", port=8123)
```

--------------------------------

### 查询操作记录

Source: https://p123client.readthedocs.io

查询网盘最新上传文件、同步空间文件列表及同步日志。

```python
client.fs_list_new({
    "orderBy": "update_time", 
    "orderDirection": "desc", 
    "SearchData": ".", 
})
```

```python
client.fs_list_new({
    "operateType": "SyncSpacePage", 
    "event": "syncFileList", 
    "RequestSource": 1, 
})
```

```python
client.fs_sync_log()
```

--------------------------------

### 由自定义 URI 转存文件

Source: https://p123client.readthedocs.io

将自定义 URI 指向的文件转存到个人网盘中。

```python
from p123client import P123Client
from p123client.tool import upload_uri

# TODO: 改成你自己的账户和密码
client = P123Client(passport="手机号或邮箱", password="登录密码")

uri = "123://torrentgalaxy.db|1976025090|582aa8bfb0ad8e6f512d9661f6243bdd"
print(upload_uri(client, uri, duplicate=1))
```

--------------------------------

### 遍历文件列表

Source: https://p123client.readthedocs.io

演示如何遍历网盘文件、分享链接文件以及将分享列表导出为 JSON。

```python
from p123client import P123Client
from p123client.tool import iterdir

# TODO: 改成你自己的账户和密码
client = P123Client(passport="手机号或邮箱", password="登录密码")

for info in iterdir(client, parent_id=0, max_depth=-1):
    print(info)
```

```python
from p123client.tool import share_iterdir

# NOTE: 无需登录
for info in share_iterdir(
    "https://www.123684.com/s/oec7Vv-CIYWh?提取码:ZY4K", 
    max_depth=-1, 
    predicate=lambda a: not a["is_dir"], 
):
    print(info)
```

```python
from orjson import dumps
from p123client.tool import share_iterdir

def export_share_files_json(
    link: str, 
    path: str = "", 
    cooldown: float = 0, 
):
    """把分享链接中的文件列表导出到 json 文件

    :param link: 分享链接，可以包含提取码
    :param path: 保存的路径，如果不提供则自动确定
    :param cooldown: 两次调用之间，冷却的时间（用两次调用开始时的时间差，而不是一次完成到下一次开始的时间差）
    """
    print("< 将拉取:", link)
    ls: list[dict] = []
    for i, a in enumerate(share_iterdir(link, max_depth=-1, cooldown=cooldown), 1):
        ls.append(a)
        print(i, a)
    if ls:
        info = ls[0]
        if not path:
            suffix = f"@{info['ShareKey']},{info['SharePwd']}.json"
            path = f"{info['name'][:255-len(suffix)]}{suffix}"
        open(path, "wb").write(dumps(ls))
        print()
        print("> 已拉取:", link)
        print("> 已保存:", path)

export_share_files_json("https://www.123684.com/s/oec7Vv-CIYWh?提取码:ZY4K")
```

--------------------------------

### OAuth Login

Source: https://p123client.readthedocs.io

Perform OAuth login to obtain access and refresh tokens.

```python
# TODO: 应用标识，创建应用时分配的 appId
client_id = "..." 
# TODO: 应用密钥，创建应用时分配的 secretId
client_secret = "..."
# TODO: 回调链接
redirect_uri = "..."
# TODO: 访问令牌
token = "..."

resp = P123Client.login_with_oauth(
    client_id=client_id, 
    client_secret=client_secret, 
    redirect_uri=redirect_uri, 
    token=token, 
)
client = P123Client(
    client_id=client_id, 
    client_secret=client_secret, 
    token=resp["access_token"], 
    refresh_token=resp["refresh_token"], 
)
```

--------------------------------

### 创建自定义 URI

Source: https://p123client.readthedocs.io

使用 P123Client 实例和文件 ID 生成自定义 URI。

```python
from p123client import P123Client
from p123client.tool import make_uri

# TODO: 改成你自己的账户和密码
client = P123Client(passport="手机号或邮箱", password="登录密码")

# TODO: 请改成你要处理的文件 id
file_id = 15688945
print(make_uri(client, file_id))
```

--------------------------------

### Extend P123Client for Custom Interfaces

Source: https://p123client.readthedocs.io

Implement custom API methods by subclassing P123Client and using the request method with appropriate parameters.

```python
from collections.abc import Coroutine
from typing import overload, Any, Literal

from p123client import P123Client

class MyCustom123Client(P123Client):

    @overload
    def foo(
        self, 
        payload: dict, 
        /, 
        async_: Literal[False] = False, 
        **request_kwargs, 
    ) -> dict:
        ...
    @overload
    def foo(
        self, 
        payload: dict, 
        /, 
        async_: Literal[True],
        **request_kwargs, 
    ) -> Coroutine[Any, Any, dict]:
        ...
    def foo(
        self, 
        payload: dict, 
        /, 
        async_: bool = False, 
        **request_kwargs, 
    ) -> dict | Coroutine[Any, Any, dict]:
        api = "https://www.123pan.com/api/foo"
        return self.request(
            api, 
            method="GET", 
            params=payload, 
            async_=async_, 
            **request_kwargs, 
        )

    @overload
    def bar(
        self, 
        payload: dict, 
        /, 
        async_: Literal[False] = False, 
        **request_kwargs, 
    ) -> dict:
        ...
    @overload
    def bar(
        self, 
        payload: dict, 
        /, 
        async_: Literal[True],
        **request_kwargs, 
    ) -> Coroutine[Any, Any, dict]:
        ...
    def bar(
        self, 
        payload: dict, 
        /, 
        async_: bool = False, 
        **request_kwargs, 
    ) -> dict | Coroutine[Any, Any, dict]:
        api = "https://www.123pan.com/api/bar"
        return self.request(
            api, 
            method="POST", 
            json=payload, 
            async_=async_, 
            **request_kwargs, 
        )
```

--------------------------------

### Parse Token Information

Source: https://p123client.readthedocs.io

Decode and inspect the JWT token structure.

```python
from base64 import urlsafe_b64decode
from json import loads

token = client.token

method, user_info, sign = token.split(".", 2)
print("JWT 算法:", loads(urlsafe_b64decode(method)))
print("用户信息:", loads(urlsafe_b64decode(user_info+"==")))
print("签名:", sign)
```

--------------------------------

### Execute Raw Requests via P123Client

Source: https://p123client.readthedocs.io

Directly invoke the underlying request method to perform HTTP operations.

```python
url = "https://www.123pan.com/api/someapi"
response = client.request(url=url, json={...})
```

--------------------------------

### P123Client.request

Source: https://p123client.readthedocs.io

The core method used to perform HTTP requests. All API interface calls are ultimately routed through this method.

```APIDOC
## P123Client.request

### Description
Performs an HTTP request to the specified URL. This is the base method for all API interactions within the client.

### Parameters
- **url** (str) - Required - The target API endpoint URL.
- **json** (dict) - Optional - The JSON payload to send in the request body.
- **method** (str) - Optional - The HTTP method (e.g., GET, POST).
- **params** (dict) - Optional - Query parameters for the request.
- **async_** (bool) - Optional - Whether to perform the request asynchronously.

### Request Example
```python
url = "https://www.123pan.com/api/someapi"
response = client.request(url=url, json={...})
```
```

--------------------------------

### Validate API Responses

Source: https://p123client.readthedocs.io

Use check_response to verify the 'code' field in JSON responses and raise P123OSError if validation fails.

```python
from p123client import check_response

# 检查同步调用
data = check_response(client.method(payload))
# 检查异步调用
data = check_response(await client.method(payload, async_=True))
```

--------------------------------

### p123client.check_response

Source: https://p123client.readthedocs.io

Validates the response from an API call. It checks the 'code' field in the returned dictionary.

```APIDOC
## p123client.check_response

### Description
Checks the response dictionary for a 'code' field. If the code is 0, 200, or missing, the data is returned. Otherwise, a P123OSError is raised.

### Parameters
- **data** (dict) - Required - The response data returned from an API call.

### Usage Example
```python
from p123client import check_response

# Check synchronous response
data = check_response(client.method(payload))

# Check asynchronous response
data = check_response(await client.method(payload, async_=True))
```
```
Tokens:
Raw