API 接入指引

package vm

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"
    "strconv"
    "strings"
    "time"
)

type createmod struct {
    BizType string `json:"BizType"`
    Type    string `json:"Type"`
    Tasks   []task
}
type task struct {
    DataId string    `json:"DataId"`
    Input  taskInput `json:"Input"`
}

type taskInput struct {
    Type string `json:"Type"`
    Url  string `json:"Url"`
}

func sha256heCreate(s string) string {
    b := sha256.Sum256([]byte(s))
    return hex.EncodeToString(b[:])
}

func hmacsha256Create(s, key string) string {
    hashed := hmac.New(sha256.New, []byte(key))
    hashed.Write([]byte(s))
    return string(hashed.Sum(nil))
}

func DetectCreateSingle() {
    secretId := ""
    secretKey := ""
    host := "vm.tencentcloudapi.com"
    algorithm := "TC3-HMAC-SHA256"
    service := "vm"
    version := "2021-09-22"
    action := "CreateVideoModerationTask"
    region := "ap-guangzhou"
    var timestamp int64 = time.Now().Unix()
    // loc, _ := time.LoadLocation("Asia/Shanghai")
    // var timestamp int64 = time.Now().In(loc).Unix()

    // step 1: build canonical request string
    httpRequestMethod := "POST"
    canonicalURI := "/"
    canonicalQueryString := ""
    canonicalHeaders := "content-type:application/json; charset=utf-8\\n" + "host:" + host + "\\n"
    signedHeaders := "content-type;host"
    var a = createmod{
        BizType: "default",
        Type:    "VIDEO",
        Tasks: []task{
            {
                DataId: "123",
                Input: taskInput{
                    Type: "URL",
                    Url:  "", //接口入参
                },
            },
        },
    }
    b, err := json.Marshal(a)
    if err != nil {
        fmt.Print(err.Error())
        return
    }
    payload := string(b)
    hashedRequestPayload := sha256heCreate(payload)
    canonicalRequest := fmt.Sprintf("%s\\n%s\\n%s\\n%s\\n%s\\n%s",
        httpRequestMethod,
        canonicalURI,
        canonicalQueryString,
        canonicalHeaders,
        signedHeaders,
        hashedRequestPayload)
    // fmt.Println(canonicalRequest)

    // step 2: build string to sign
    date := time.Unix(timestamp, 0).UTC().Format("2006-01-02")
    credentialScope := fmt.Sprintf("%s/%s/tc3_request", date, service)
    hashedCanonicalRequest := sha256heCreate(canonicalRequest)
    string2sign := fmt.Sprintf("%s\\n%d\\n%s\\n%s",
        algorithm,
        timestamp,
        credentialScope,
        hashedCanonicalRequest)
    // fmt.Println(string2sign)

    // step 3: sign string
    secretDate := hmacsha256Create(date, "TC3"+secretKey)
    secretService := hmacsha256Create(service, secretDate)
    secretSigning := hmacsha256Create("tc3_request", secretService)
    signature := hex.EncodeToString([]byte(hmacsha256Create(string2sign, secretSigning)))
    // fmt.Println(signature)

    // step 4: build authorization
    authorization := fmt.Sprintf("%s Credential=%s/%s, SignedHeaders=%s, Signature=%s",
        algorithm,
        secretId,
        credentialScope,
        signedHeaders,
        signature)

    req, _ := http.NewRequest("POST", "https://"+host, strings.NewReader(payload))
    req.Header.Set("Authorization", authorization)
    req.Header.Set("Content-Type", "application/json; charset=utf-8")
    req.Header.Set("Host", host)
    req.Header.Set("X-TC-Action", action)
    req.Header.Set("X-TC-Timestamp", strconv.FormatInt(timestamp, 10))
    req.Header.Set("X-TC-Version", version)
    req.Header.Set("X-TC-Region", region)

    resp, err := (&http.Client{}).Do(req)
    if err != nil {
        fmt.Print(err.Error())
        return
    }
    defer resp.Body.Close()

    respByte, _ := ioutil.ReadAll(resp.Body)
    fmt.Println(string(respByte))
}

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import hashlib
import hmac
import json
import time
import datetime
import requests

from concurrent.futures import ThreadPoolExecutor

secret_id = ""
secret_key = ""

service = "vm"
host = "vm.tencentcloudapi.com"
endpoint = "https://" + host
region = "ap-guangzhou"
version = "2021-09-22"
algorithm = "TC3-HMAC-SHA256"


def do_action(action, params):

    timestamp = int(time.time())
    # datetime.datetime.utcfromtimestamp
    day = datetime.datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")

    # ************* 步骤 1：拼接规范请求串 *************
    http_request_method = "POST"
    canonical_uri = "/"
    canonical_querystring = ""
    ct = "application/json; charset=utf-8"
    payload = json.dumps(params)
    # print(payload)
    canonical_headers = "content-type:%s\\nhost:%s\\n" % (ct, host)
    signed_headers = "content-type;host"
    hashed_request_payload = hashlib.sha256(
        payload.encode("utf-8")).hexdigest()
    # print(hashed_request_payload)
    canonical_request = (http_request_method + "\\n" + canonical_uri + "\\n" +
                         canonical_querystring + "\\n" + canonical_headers +
                         "\\n" + signed_headers + "\\n" + hashed_request_payload)
    # print(canonical_request)

    # ************* 步骤 2：拼接待签名字符串 *************
    credential_scope = day + "/" + service + "/" + "tc3_request"
    hashed_canonical_request = hashlib.sha256(
        canonical_request.encode("utf-8")).hexdigest()
    string_to_sign = (algorithm + "\\n" + str(timestamp) + "\\n" +
                      credential_scope + "\\n" + hashed_canonical_request)

    # print(string_to_sign)

    secret_date = sign(("TC3" + secret_key).encode("utf-8"), day)
    secret_service = sign(secret_date, service)
    secret_signing = sign(secret_service, "tc3_request")
    signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"),
                         hashlib.sha256).hexdigest()
    # print(signature)

    # ************* 步骤 4：拼接 Authorization *************
    authorization = (algorithm + " " + "Credential=" + secret_id + "/" +
                     credential_scope + ", " + "SignedHeaders=" +
                     signed_headers + ", " + "Signature=" + signature)
    # print(authorization)

    headers = {
        "Authorization": authorization,
        "Content-Type": "application/json; charset=utf-8",
        "Host": host,
        "X-TC-Action": action,
        "X-TC-Timestamp": str(timestamp),
        "X-TC-Version": version,
        "X-TC-Region": region
    }
    # time.sleep(15)
    # 发送请求
    resp = requests.post(url=endpoint, data=payload, headers=headers)

    return resp


# 计算签名摘要函数
def sign(key, msg):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def audio_detect(params):
    action = "CreateVideoModerationTask"
    resp = do_action(action, params)

    return resp.text


if __name__ == "__main__":
    params = []
    video_url = ""
    p = {
        "Type":
        "VIDEO",
        "Tasks": [{
            "DataId": "default",
            "Name": "测试视频",
            "Input": {
                "Type": "URL",
                "Url": video_url
            }
        }],
        "BizType":
        "default"
    }
    params.append(p)
    executor = ThreadPoolExecutor(max_workers=5)
    resps = executor.map(audio_detect, params)
    for resp in resps:
        print(resp)