优化上传图片与视频

service/doudian_service.py

@@ -31,6 +31,7 @@ def generate_a_bogus(params:str | dict,data: dict,ua: str, params_type: int=1):
 
 
 def get_schema(headers: dict, proxy_url: str, category_id: str):
+    """获取商品类目"""
     params = {
         'appid': '1',
         '_bid': 'ffa_goods',
@@ -59,6 +60,7 @@ def get_schema(headers: dict, proxy_url: str, category_id: str):
     return response
 
 def create_global_promotion(params:str | dict,data: dict,headers: dict, proxy_url: str):
+    """创建全域推广"""
     proxies = check_proxy(proxy_url)
     if headers.get("User-Agent"):
         ua = headers['User-Agent']
@@ -74,7 +76,7 @@ def create_global_promotion(params:str | dict,data: dict,headers: dict, proxy_ur
     return response
 
 def search_category(product_name, cookies, headers, proxies):
-    """you"""
+    """搜索商品分类"""
     if 15 > len(product_name) > 60:
         raise Exception('商品标题长度必须在15-60字符之间')
     params = {
@@ -87,8 +89,8 @@ def search_category(product_name, cookies, headers, proxies):
     response = doudian_request("GET", url, proxies, params, headers=headers, cookies=cookies)
     return response['data']
 
-# AI视频生成
 def generate_video(task: dict):
+    """生成视频"""
     if type(task.get("proxies")) == dict:
         addr = task.get("proxies")["addr"]
         port = task.get("proxies")["port"]
@@ -105,24 +107,15 @@ def generate_video(task: dict):
     generate_video_data = task.get('generate_video_data')
     if not generate_video_data:
         raise ParamsError('缺少generate_video_data参数')
-    # generate_video_data = {
-    #     'product_name': "自粘地板贴家用瓷砖加厚耐磨防水地胶地垫",
-    #     'customized_images': [
-    #         "https://p3-aio.ecombdimg.com/obj/ecom-shop-material/jpeg_m_9ce719a34608aff3b11827271a67da85_sx_99050_www810-1080",
-    #         "https://p3-aio.ecombdimg.com/obj/ecom-shop-material/jpeg_m_4e88344c8a6746ab6bd79cebd34778a1_sx_187733_www1080-980",
-    #         "https://p3-aio.ecombdimg.com/obj/ecom-shop-material/jpeg_m_995815fcd3378e7f48765b3ef3fc9bec_sx_119123_www810-1080",
-    #         "https://p3-aio.ecombdimg.com/obj/ecom-shop-material/jpeg_m_b1fe80bb8228d713eabc880561120823_sx_53964_www810-1080",
-    #         "https://p3-aio.ecombdimg.com/obj/ecom-shop-material/jpeg_m_cc78dc386c3359945173fc9cb57297c4_sx_137165_www810-1080"],
-    # }
     params = {
         'appid': '1',
         '_bid': 'ffa_goods',
     }
     product_name =  generate_video_data['product_name']
     customized_images =  generate_video_data['customized_images']
-    logger.info(len(customized_images))
     if len(customized_images) < 3:
         raise Exception('生成视频的图片数量至少为3张')
+    logger.info(f"商品名称 --> {product_name}")
     category_list = search_category(product_name, cookies, headers, proxies)
     if not category_list:
         raise Exception('未找到商品分类')
@@ -131,7 +124,7 @@ def generate_video(task: dict):
     third_name = category_list[0].get('third_name', '')
     fourth_name = category_list[0].get('fourth_name', '')
     product_industrys = [first_name, second_name, third_name, fourth_name]
-    logger.info(f"商品分类：{product_industrys}")
+    logger.info(f"商品分类 --> {product_industrys}")
     json_data = {
         'product_id': '',
         'optimize_strategy': '视频一键生成',
@@ -165,6 +158,7 @@ def generate_video(task: dict):
     return result
 
 def get_task_result(task_id: str, headers: dict, tool_source: str, ua, proxies, cookies):
+    """获取任务结果"""
     params = {
         'task_id': task_id,
         'app_id': '',

service/upload_image_and_video.py

@@ -223,7 +223,8 @@ async def upload_single_video(task: dict, local_path: any, original_url: str, cu
     """上传单个视频"""
     task['file_path'] = local_path
     task['video_url'] = original_url
-    product_name = task['title'].encode('utf-8').decode('unicode_escape')
+    product_name = repr(task['title'])
+    logger.info(f"product_name：{product_name}")
     task['generate_video_data'] = {
         "product_name": product_name,
         "customized_images": customized_images

service/upload_video.py

@@ -681,6 +681,9 @@ def upload_video_with_multithreading(task):
 
 def prepare_video_file(task):
     """准备视频文件"""
+    video_url = task.get("video_url")
+    if not video_url:
+        raise VideoError("video_url is None")
     if not task.get("file_path"):
         if not os.path.exists(VIDEO_PATH):
             os.makedirs(VIDEO_PATH)