素材库鉴权与通用约定

import hashlib
import hmac
import json
import os
from datetime import datetime, timezone
from urllib.error import HTTPError
from urllib.request import Request, urlopen

HOST = "api-vip.apigo.ai"
REGION, SERVICE, VERSION = "ap-southeast-1", "ark", "2024-01-01"


def call(key, action, body):
    payload = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    x_date = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    day = x_date[:8]
    query = f"Action={action}&Version={VERSION}"
    body_hash = hashlib.sha256(payload).hexdigest()
    signed_headers = "content-type;host;x-content-sha256;x-date"
    canonical_request = "\n".join([
        "POST", "/", query,
        f"content-type:application/json\nhost:{HOST}\n"
        f"x-content-sha256:{body_hash}\nx-date:{x_date}\n",
        signed_headers, body_hash,
    ])
    scope = f"{day}/{REGION}/{SERVICE}/request"
    string_to_sign = "\n".join([
        "HMAC-SHA256", x_date, scope,
        hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
    ])
    signing_key = key.encode("utf-8")
    for part in (day, REGION, SERVICE, "request"):
        signing_key = hmac.new(signing_key, part.encode("utf-8"), hashlib.sha256).digest()
    signature = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    request = Request(f"https://{HOST}/?{query}", data=payload, method="POST", headers={
        "Content-Type": "application/json",
        "X-Date": x_date,
        "X-Content-Sha256": body_hash,
        "Authorization": f"HMAC-SHA256 Credential={key}/{scope}, "
                         f"SignedHeaders={signed_headers}, Signature={signature}",
    })
    try:
        with urlopen(request, timeout=60) as resp:
            return resp.status, json.loads(resp.read())
    except HTTPError as e:
        return e.code, json.loads(e.read())


if __name__ == "__main__":
    status, data = call(os.environ["APIGO_API_KEY"], "CreateAssetGroup", {
        "Name": "product-assets",
        "GroupType": "AIGC",
    })
    print(status, json.dumps(data, ensure_ascii=False, indent=2))