获取真人 IP 版权视频授权书 API 教程

1
2
3
4{
  "IPCategory": "REAL_PERSON",
  "ProjectName": "default"
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18{
  "Result": {
    "Items": [
      {
        "IpId": "ip-c-example-a",
        "IPCategory": "REAL_PERSON",
        "IPName": "示例演员甲",
        "SupportsAuthLetterApplication": true
      },
      {
        "IpId": "ip-c-example-b",
        "IPCategory": "REAL_PERSON",
        "IPName": "示例演员乙",
        "SupportsAuthLetterApplication": false
      }
    ]
  }
}

1
2
3
4
5
6
7{
  "IpId": "ip-c-example",
  "AuthorizedSubject": {
    "SubjectName": "示例文化传媒有限公司"
  },
  "ProjectName": "default"
}

1
2
3{
  "SessionId": "example-session-id"
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15{
  "ResponseMetadata": {
    "RequestId": "example-request-id"
  },
  "Result": {
    "Session": {
      "SessionId": "example-session-id",
      "Status": "SUCCEEDED",
      "File": {
        "FileName": "授权书.pdf",
        "URL": "https://<FILE_SERVICE_HOST>/<SIGNED_PATH>"
      }
    }
  }
}

1
2
3
4
5
6{
  "Callback": {
    "URL": "https://callback.example.com/auth-letter",
    "Secret": "<43_CHARACTER_BASE64URL_SECRET>"
  }
}

1python -c 'import secrets; print(secrets.token_urlsafe(32))'

1
2
3
4
5{
  "SessionId": "example-session-id",
  "Status": "SUCCEEDED",
  "ApplicationId": "example-application-id"
}

1python -m pip install --upgrade volcengine requests

1
2export VOLC_ACCESSKEY="<ACCESS_KEY>"
export VOLC_SECRETKEY="<SECRET_KEY>"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169import json
import math
import os
import time
from pathlib import Path
from urllib.parse import urlsplit

import requests
from volcengine.Credentials import Credentials
from volcengine.auth.SignerV4 import SignerV4
from volcengine.base.Request import Request

HOST = "open.volcengineapi.com"
VERSION = "2024-01-01"


def call_api(action, payload):
    region = os.getenv("VOLC_REGION", "cn-beijing")
    credentials = Credentials(
        os.environ["VOLC_ACCESSKEY"],
        os.environ["VOLC_SECRETKEY"],
        "ark",
        region,
        os.getenv("VOLC_SESSION_TOKEN", ""),
    )
    request = Request()
    request.schema = "https"
    request.host = HOST
    request.path = "/"
    request.method = "POST"
    request.query = {"Action": action, "Version": VERSION}
    request.headers = {"Host": HOST, "Content-Type": "application/json"}
    request.body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
    SignerV4.sign(request, credentials)

    try:
        # Send the same UTF-8 body that was signed without serializing it again.
        response = requests.post(
            request.build(),
            headers=request.headers,
            data=request.body.encode("utf-8"),
            timeout=(5, 30),
            allow_redirects=False,
        )
    except requests.RequestException:
        message = f"{action} network error"
        if action == "CreateRightAuthLetterApplicationSession":
            message += "; do not automatically retry this request"
        raise RuntimeError(message) from None

    try:
        data = response.json()
    except ValueError:
        raise RuntimeError(
            f"API returned non-JSON content: HTTP {response.status_code}"
        ) from None

    metadata = data.get("ResponseMetadata", {})
    error = metadata.get("Error")
    if not 200 <= response.status_code < 300 or error:
        code = (error or {}).get("Code", "UnknownError")
        request_id = metadata.get("RequestId", "")
        raise RuntimeError(f"API error {code}; RequestId={request_id}")
    result = data.get("Result")
    if not isinstance(result, dict):
        raise RuntimeError(f"{action} response does not contain Result")
    return result


def list_applicable_ips(project_name="default"):
    result = call_api(
        "ListAuthorizedCommercialIPIds",
        {
            "IPCategory": "REAL_PERSON",
            "ProjectName": project_name,
        },
    )
    items = result.get("Items")
    if not isinstance(items, list):
        raise RuntimeError(
            "ListAuthorizedCommercialIPIds response does not contain Items"
        )
    return [item for item in items if item.get("SupportsAuthLetterApplication") is True]


def create_entry(ip_id, subject_name, callback=None):
    payload = {
        "IpId": ip_id,
        "AuthorizedSubject": {"SubjectName": subject_name},
        "ProjectName": "default",
    }
    if callback is not None:
        payload["Callback"] = callback
    return call_api("CreateRightAuthLetterApplicationSession", payload)


def get_session(session_id):
    return call_api(
        "GetRightAuthLetterApplicationSession",
        {"SessionId": session_id},
    )["Session"]


def wait_for_file(session_id, wait_seconds=120, interval_seconds=5):
    # Reaching the polling deadline does not mean the application failed.
    if wait_seconds < 0:
        raise ValueError("wait_seconds must be non-negative")
    if interval_seconds <= 0:
        raise ValueError("interval_seconds must be positive")

    deadline = time.monotonic() + wait_seconds
    max_attempts = max(1, math.ceil(wait_seconds / interval_seconds) + 1)
    for _ in range(max_attempts):
        session = get_session(session_id)
        status = session["Status"]
        if status == "SUCCEEDED":
            file_info = session.get("File")
            if not file_info or not file_info.get("URL"):
                raise RuntimeError("Successful response does not contain a file URL")
            return file_info
        if status in {"FAILED", "EXPIRED", "VOIDED"}:
            raise RuntimeError(f"Application status: {status}")
        if status not in {"CREATED", "PROCESSING"}:
            raise RuntimeError("Unknown status; keep the SessionId and contact support")

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return None
        time.sleep(min(interval_seconds, remaining))
    return None


def download_file(file_info, output_path):
    # Choose the local path in your application; do not trust a remote filename.
    url = file_info["URL"]
    parsed_url = urlsplit(url)
    if parsed_url.scheme != "https" or not parsed_url.hostname:
        raise RuntimeError("File URL is not a valid HTTPS URL")

    try:
        # Do not attach OpenAPI credentials or follow redirects.
        with requests.get(
            url,
            timeout=(5, 60),
            allow_redirects=False,
        ) as response:
            if response.status_code != 200:
                raise RuntimeError(
                    f"File download returned HTTP {response.status_code}; "
                    "query the session again for a new URL"
                )
            if not response.content:
                raise RuntimeError("Downloaded file is empty")
            with Path(output_path).open("xb") as output:
                output.write(response.content)
    except requests.RequestException:
        # Do not expose an exception that may contain the signed URL.
        raise RuntimeError(
            "Download network error; query the session again before retrying"
        ) from None


if __name__ == "__main__":
    eligible_ips = list_applicable_ips()
    safe_items = [
        {"IpId": item.get("IpId"), "IPName": item.get("IPName")}
        for item in eligible_ips
    ]
    print(json.dumps({"Items": safe_items}, ensure_ascii=False, indent=2))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19import secrets

ips = list_applicable_ips()
# Present ips to the terminal user and obtain the selected IP ID.

callback = None
# To enable callbacks, replace CALLBACK_URL and securely store the secret:
# callback = {
#     "URL": "<CALLBACK_URL>",
#     "Secret": secrets.token_urlsafe(32),
# }

created = create_entry(
    "<COMMERCIAL_IP_ID>",
    "<AUTHORIZED_SUBJECT_NAME>",
    callback=callback,
)
session_id = created["SessionId"]
application_url = created["ApplicationURL"]

1
2
3file_info = wait_for_file("<SESSION_ID>", wait_seconds=120)
if file_info is not None:
    download_file(file_info, "authorization-letter.pdf")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import base64
import hashlib
import hmac
import re
import time


def verify_callback(raw_body, timestamp, signature, secret):
    if not isinstance(raw_body, bytes):
        return False
    if not re.fullmatch(r"[A-Za-z0-9_-]{43}", secret):
        return False
    if not re.fullmatch(r"[0-9]{1,12}", timestamp):
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False
    if not re.fullmatch(r"v1=[0-9a-f]{64}", signature):
        return False

    key = base64.urlsafe_b64decode(secret + "=")
    if len(key) != 32:
        return False
    message = timestamp.encode("ascii") + b"." + raw_body
    expected = "v1=" + hmac.new(key, message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)