|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +import hashlib |
| 3 | +import hmac |
| 4 | +import json |
| 5 | +import sys |
| 6 | +import time |
| 7 | +import base64 |
| 8 | +from datetime import datetime |
| 9 | +import os |
| 10 | +import requests |
| 11 | +from src.config import TENCENT_SECRET_ID, TENCENT_SECRET_KEY |
| 12 | + |
| 13 | +if sys.version_info[0] <= 2: |
| 14 | + from httplib import HTTPSConnection |
| 15 | +else: |
| 16 | + from http.client import HTTPSConnection |
| 17 | + |
| 18 | + |
| 19 | +def sign(key, msg): |
| 20 | + return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest() |
| 21 | + |
| 22 | + |
| 23 | +def post_request(action, payload): |
| 24 | + secret_id = TENCENT_SECRET_ID |
| 25 | + secret_key = TENCENT_SECRET_KEY |
| 26 | + token = "" |
| 27 | + service = "hunyuan" |
| 28 | + host = "hunyuan.tencentcloudapi.com" |
| 29 | + region = "ap-guangzhou" |
| 30 | + version = "2023-09-01" |
| 31 | + params = json.loads(payload) |
| 32 | + endpoint = "https://hunyuan.tencentcloudapi.com" |
| 33 | + algorithm = "TC3-HMAC-SHA256" |
| 34 | + timestamp = int(time.time()) |
| 35 | + date = datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d") |
| 36 | + # ************* step 1 canonical request ************* |
| 37 | + http_request_method = "POST" |
| 38 | + canonical_uri = "/" |
| 39 | + canonical_querystring = "" |
| 40 | + ct = "application/json; charset=utf-8" |
| 41 | + canonical_headers = "content-type:%s\nhost:%s\nx-tc-action:%s\n" % ( |
| 42 | + ct, |
| 43 | + host, |
| 44 | + action.lower(), |
| 45 | + ) |
| 46 | + signed_headers = "content-type;host;x-tc-action" |
| 47 | + hashed_request_payload = hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 48 | + canonical_request = ( |
| 49 | + http_request_method |
| 50 | + + "\n" |
| 51 | + + canonical_uri |
| 52 | + + "\n" |
| 53 | + + canonical_querystring |
| 54 | + + "\n" |
| 55 | + + canonical_headers |
| 56 | + + "\n" |
| 57 | + + signed_headers |
| 58 | + + "\n" |
| 59 | + + hashed_request_payload |
| 60 | + ) |
| 61 | + |
| 62 | + # ************* step 2 string to sign ************* |
| 63 | + credential_scope = date + "/" + service + "/" + "tc3_request" |
| 64 | + hashed_canonical_request = hashlib.sha256( |
| 65 | + canonical_request.encode("utf-8") |
| 66 | + ).hexdigest() |
| 67 | + string_to_sign = ( |
| 68 | + algorithm |
| 69 | + + "\n" |
| 70 | + + str(timestamp) |
| 71 | + + "\n" |
| 72 | + + credential_scope |
| 73 | + + "\n" |
| 74 | + + hashed_canonical_request |
| 75 | + ) |
| 76 | + |
| 77 | + # ************* step 3 calculate signature ************* |
| 78 | + secret_date = sign(("TC3" + secret_key).encode("utf-8"), date) |
| 79 | + secret_service = sign(secret_date, service) |
| 80 | + secret_signing = sign(secret_service, "tc3_request") |
| 81 | + signature = hmac.new( |
| 82 | + secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256 |
| 83 | + ).hexdigest() |
| 84 | + |
| 85 | + # ************* step 4 build authorization ************* |
| 86 | + authorization = ( |
| 87 | + algorithm |
| 88 | + + " " |
| 89 | + + "Credential=" |
| 90 | + + secret_id |
| 91 | + + "/" |
| 92 | + + credential_scope |
| 93 | + + ", " |
| 94 | + + "SignedHeaders=" |
| 95 | + + signed_headers |
| 96 | + + ", " |
| 97 | + + "Signature=" |
| 98 | + + signature |
| 99 | + ) |
| 100 | + |
| 101 | + # ************* step 5 build and send request ************* |
| 102 | + headers = { |
| 103 | + "Authorization": authorization, |
| 104 | + "Content-Type": "application/json; charset=utf-8", |
| 105 | + "Host": host, |
| 106 | + "X-TC-Action": action, |
| 107 | + "X-TC-Timestamp": timestamp, |
| 108 | + "X-TC-Version": version, |
| 109 | + } |
| 110 | + if region: |
| 111 | + headers["X-TC-Region"] = region |
| 112 | + if token: |
| 113 | + headers["X-TC-Token"] = token |
| 114 | + |
| 115 | + try: |
| 116 | + req = HTTPSConnection(host) |
| 117 | + req.request("POST", "/", headers=headers, body=payload.encode("utf-8")) |
| 118 | + resp = req.getresponse() |
| 119 | + return resp.read() |
| 120 | + except Exception as err: |
| 121 | + print(err) |
| 122 | + |
| 123 | + |
| 124 | +def hunyuan_generate_cover(your_file_path): |
| 125 | + submit_action = "SubmitHunyuanImageJob" |
| 126 | + with open(your_file_path, "rb") as image_file: |
| 127 | + data = base64.b64encode(image_file.read()).decode("utf-8") |
| 128 | + payload = f'{{"Prompt":"这是一个视频截图，请尝试根据该图生成对应的动漫类型的封面","Style":"riman","ContentImage":{{"ImageBase64":"data:image/png;base64,{data}"}}}}' |
| 129 | + submit_return = post_request(submit_action, payload).decode("utf-8") |
| 130 | + job_id = json.loads(submit_return)["Response"]["JobId"] |
| 131 | + query_action = "QueryHunyuanImageJob" |
| 132 | + payload = f'{{"JobId":"{job_id}"}}' |
| 133 | + max_retries = 30 |
| 134 | + retries = 0 |
| 135 | + while retries < max_retries: |
| 136 | + query_return = post_request(query_action, payload).decode("utf-8") |
| 137 | + if json.loads(query_return)["Response"]["JobStatusCode"] == "5": |
| 138 | + image_url = json.loads(query_return)["Response"]["ResultImage"][0] |
| 139 | + img_data = requests.get(image_url).content |
| 140 | + cover_name = time.strftime("%Y%m%d%H%M%S") + ".png" |
| 141 | + temp_cover_path = os.path.join(os.path.dirname(your_file_path), cover_name) |
| 142 | + with open(temp_cover_path, "wb") as handler: |
| 143 | + handler.write(img_data) |
| 144 | + os.remove(your_file_path) |
| 145 | + return temp_cover_path |
| 146 | + else: |
| 147 | + time.sleep(1) |
| 148 | + retries += 1 |
| 149 | + return None |
| 150 | + |
| 151 | + |
| 152 | +if __name__ == "__main__": |
| 153 | + print(hunyuan_generate_cover("")) |
0 commit comments