From 423a6e7b6f22d76ec8ed87f12c51159db0785578 Mon Sep 17 00:00:00 2001 From: liubiren Date: Tue, 17 Mar 2026 21:36:29 +0800 Subject: [PATCH] 1 --- utils/authenticator.py | 32 +++++++++++++++------------ utils/certifications.json | 2 +- utils/feishu.py | 18 +++++++++------ 短视频智能生成/main.py | 46 ++++++++++++++++++--------------------- 4 files changed, 51 insertions(+), 47 deletions(-) diff --git a/utils/authenticator.py b/utils/authenticator.py index a14934b..8261708 100644 --- a/utils/authenticator.py +++ b/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}×tamp={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"], diff --git a/utils/certifications.json b/utils/certifications.json index f8ea234..d177044 100644 --- a/utils/certifications.json +++ b/utils/certifications.json @@ -1 +1 @@ -{"szkt": {"token": "a62c56ace614a6546191d5af8ca8b1513cfaeaea7ce67d0a37de994ab6c2aa4e2a0b058e0da575ff376dd51dc19c5ad353ab2761cb6d9db4d521b83adeee2979b78f7ae70765b26985165b6266d084b75f2f918008966e72a116d8bca5ec4c7cecc5223f78fa47b4d40aa9cf5277a11b0b967ad06e84ef7c4acbc53ccdef936c062b2d037ae0dad8c29d50426b668ec349cc8c0099a0270e16f97d31e4f058bc086334468f88d934c7fd1464ed3800833d2f486dc06f0689b99abbb78a8ebf4a3877bd82d0dd765dc09b7a1594fa8849d51f59282a81048c52e82e8320d1ad042a6c307ca831647cba4356564704780f", "expired_timestamp": 1859201579.393386}} \ No newline at end of file +{"szkt": {"token": "a62c56ace614a6546191d5af8ca8b1513cfaeaea7ce67d0a37de994ab6c2aa4e2a0b058e0da575ff376dd51dc19c5ad353ab2761cb6d9db4d521b83adeee2979b78f7ae70765b26985165b6266d084b75f2f918008966e72a116d8bca5ec4c7cecc5223f78fa47b4d40aa9cf5277a11b0b967ad06e84ef7c4acbc53ccdef936c062b2d037ae0dad8c29d50426b668ec349cc8c0099a0270e16f97d31e4f058bc086334468f88d934c7fd1464ed3800833d2f486dc06f0689b99abbb78a8ebf4a3877bd82d0dd765dc09b7a1594fa8849d51f59282a81048c52e82e8320d1ad042a6c307ca831647cba4356564704780f", "expired_timestamp": 1859201579.393386}, "feishu": {"token": "t-g1043hkKMBTKBZNZJPNCVDO7JMNOLWMM56ZCWKR6", "expired_timestamp": 1773758800.9476614}} \ No newline at end of file diff --git a/utils/feishu.py b/utils/feishu.py index 54658c3..4e693d3 100644 --- a/utils/feishu.py +++ b/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')}", } diff --git a/短视频智能生成/main.py b/短视频智能生成/main.py index 6006d5c..1677f15 100644 --- a/短视频智能生成/main.py +++ b/短视频智能生成/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]: """ 获取分镜脚本