1

utils/authenticator.py

@@ -1,20 +1,22 @@
 # -*- coding: utf-8 -*-
 """
-认证器模块
+封装认证器
 """
 
 import hashlib
 import hmac
 import json
-import sys
+from pathlib import Path
 import threading
 import time
-from pathlib import Path
 from typing import Optional, Tuple
 
-sys.path.append(Path(__file__).parent.as_posix())
 from request import Request
 
+import sys
+
+sys.path.append(Path(__file__).parent.as_posix())
+
 
 class Authenticator:
     """
@@ -37,6 +39,9 @@ class Authenticator:
                     ensure_ascii=False,
                 )
 
+        # 初始化所有访问凭证
+        self.certifications = {}
+
         # 初始化请求客户端
         self.request = Request()
 
@@ -75,9 +80,11 @@ class Authenticator:
             if time.time() > expired_timestamp:
                 match servicer:
                     case "szkt":
-                        token, expired_timestamp = self._get_szkt_certification()
-                    case "hlyj":
-                        token, expired_timestamp = self._get_hlyj_certification()
+                        token, expired_timestamp = (
+                            self._get_inspirvision_certification()
+                        )
+                    case "hollycrm":
+                        token, expired_timestamp = self._get_hollycrm_certification()
                     case "feishu":
                         token, expired_timestamp = self._get_feishu_certification()
                     case _:
@@ -97,7 +104,7 @@ class Authenticator:
 
             return token
 
-    def _get_szkt_certification(self) -> tuple[str, float]:
+    def _get_inspirvision_certification(self) -> Tuple[str, float]:
         """
         获取深圳快瞳访问凭证
         :return: 访问令牌和失效时间戳
@@ -115,7 +122,7 @@ class Authenticator:
             time.time() + response["data"]["expires_in"],
         )
 
-    def _get_hlyj_certification(self) -> Tuple[str, float]:
+    def _get_hollycrm_certification(self) -> Tuple[str, float]:
         """获取合力亿捷访问凭证"""
         # 企业访问标识
         access_key_id = "25938f1c190448829dbdb5d344231e42"
@@ -129,20 +136,19 @@ class Authenticator:
             f"{access_key_id}{secret_access_key}{timestamp}".encode("utf-8"),
             hashlib.sha256,
         ).hexdigest()
+
         response = self.request.get(
             url=f"https://kms.7x24cc.com/api/v1/corp/auth/token?access_key_id={access_key_id}&timestamp={timestamp}&signature={signature}"
         )
-
         # 若非响应成功则抛出异常
         if not response["success"]:
             raise RuntimeError("获取合力亿捷访问凭证发生异常")
-
         return (
             response["data"],
             time.time() + 1 * 3600,  # 访问令牌有效期为1小时
         )
 
-    def _get_feishu_certification(self) -> tuple[str, float]:
+    def _get_feishu_certification(self) -> Tuple[str, float]:
         """获取飞书访问凭证"""
         response = self.request.post(
             url="https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
@@ -151,11 +157,9 @@ class Authenticator:
                 "app_secret": "vZXGZomwfmyaHXoG8s810d1YYGLsIqCA",
             },
         )
-
         # 若非响应成功则抛出异常
         if not response["code"] == 0:
             raise RuntimeError("获取飞书访问凭证发生异常")
-
         return (
             response["tenant_access_token"],
             time.time() + response["expire"],

utils/certifications.json

@@ -1 +1 @@
-{"szkt": {"token": "a62c56ace614a6546191d5af8ca8b1513cfaeaea7ce67d0a37de994ab6c2aa4e2a0b058e0da575ff376dd51dc19c5ad353ab2761cb6d9db4d521b83adeee2979b78f7ae70765b26985165b6266d084b75f2f918008966e72a116d8bca5ec4c7cecc5223f78fa47b4d40aa9cf5277a11b0b967ad06e84ef7c4acbc53ccdef936c062b2d037ae0dad8c29d50426b668ec349cc8c0099a0270e16f97d31e4f058bc086334468f88d934c7fd1464ed3800833d2f486dc06f0689b99abbb78a8ebf4a3877bd82d0dd765dc09b7a1594fa8849d51f59282a81048c52e82e8320d1ad042a6c307ca831647cba4356564704780f", "expired_timestamp": 1859201579.393386}}
+{"szkt": {"token": "a62c56ace614a6546191d5af8ca8b1513cfaeaea7ce67d0a37de994ab6c2aa4e2a0b058e0da575ff376dd51dc19c5ad353ab2761cb6d9db4d521b83adeee2979b78f7ae70765b26985165b6266d084b75f2f918008966e72a116d8bca5ec4c7cecc5223f78fa47b4d40aa9cf5277a11b0b967ad06e84ef7c4acbc53ccdef936c062b2d037ae0dad8c29d50426b668ec349cc8c0099a0270e16f97d31e4f058bc086334468f88d934c7fd1464ed3800833d2f486dc06f0689b99abbb78a8ebf4a3877bd82d0dd765dc09b7a1594fa8849d51f59282a81048c52e82e8320d1ad042a6c307ca831647cba4356564704780f", "expired_timestamp": 1859201579.393386}, "feishu": {"token": "t-g1043hkKMBTKBZNZJPNCVDO7JMNOLWMM56ZCWKR6", "expired_timestamp": 1773758800.9476614}}

utils/feishu.py

@@ -1,30 +1,34 @@
 # -*- coding: utf-8 -*-
 """
-飞书客户端模块
+封装飞书客户端
 """
 
-import re
-import time
 from email.parser import BytesParser
 from email.policy import default
 from email.utils import parsedate_to_datetime
 from imaplib import IMAP4_SSL
+import re
+import time
+from typing import Dict, Any
 
 import pandas
-from request import Request
+
 from authenticator import Authenticator
+from request import Request
 
 
 class Feishu:
     """飞书客户端"""
 
     def __init__(self):
+        # 初始化认证器
         self.authenticator = Authenticator()
+        # 初始化请求客户端
         self.http_client = Request()
 
-    def _headers(self):
-        """请求头"""
-        # 装配飞书访问凭证
+    def _get_headers(self) -> Dict[str, Any]:
+        """获取请求头"""
+        # 构建身份认证请求头
         return {
             "Authorization": f"Bearer {self.authenticator.get_token(servicer='feishu')}",
         }

短视频智能生成/main.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 """
-主程序
+程序入口
+根据商品主图智能生成短视频广告
 """
 import warnings
 
@@ -16,33 +17,36 @@ from time import sleep
 from typing import Any, Dict, List
 from uuid import uuid4
 
-from volcenginesdkarkruntime import Ark
-from volcenginesdkarkruntime.types.responses import (
-    ResponseOutputMessage,
-    ResponseOutputText,
-)
-
 sys.path.append(Path(__file__).parent.parent.as_posix())
 
 from utils.request import Request
 from create_draft import JianYingDraftGenerator
 
-
-# 初始化火山引擎 Ark 客户端
-ark_client = Ark(
-    base_url="https://ark.cn-beijing.volces.com/api/v3",
-    api_key="2c28ab07-888c-45be-84a2-fc4b2cb5f3f2",
-)  # 本人火山引擎账密
-
 # 初始化请求客户端
-request_client = Request()
+request = Request(
+    default_headers={
+        "Content-Type": "application/json; charset=utf-8",
+        "Authorization": "Bearer 2c28ab07-888c-45be-84a2-fc4b2cb5f3f2",
+    }
+)
 
 
-def get_product_image() -> List[str]:
+def generate_task_id() -> str:
+    """
+    生成任务标识
+    :return: 任务ID
+    """
+    return uuid4().hex.upper().replace("-", "")
+
+
+def get_product_images() -> None:
     """
     获取产品图片
-    :return: 产品图片
+    :return: 无
     """
+
+
+
     try:
         with open(
             file=Path(__file__).parent / "brand_words.txt", mode="r", encoding="utf-8"
@@ -56,14 +60,6 @@ def get_product_image() -> List[str]:
         raise exception
 
 
-def generate_task_id() -> str:
-    """
-    生成任务标识
-    :return: 任务ID
-    """
-    return uuid4().hex.upper().replace("-", "")
-
-
 def get_storyboard(brand_word: str) -> Dict[str, Any]:
     """
     获取分镜脚本