对上传的图片进行校验

requirements.txt

@@ -29,4 +29,5 @@ python-dateutil
 python-socks~=2.7.1
 python-socks~=2.7.1
 PySocks
 PySocks
 
 
-requests-toolbelt~=1.0.0
+requests-toolbelt~=1.0.0
\ No newline at end of file
+pillow~=11.2.1
\ No newline at end of file

service/upload_image_and_video.py

@@ -4,6 +4,7 @@ import json
 import os
 import os
 
 
 import requests
 import requests
+from PIL import Image
 from loguru import logger
 from loguru import logger
 from requests_toolbelt.multipart.encoder import MultipartEncoder
 from requests_toolbelt.multipart.encoder import MultipartEncoder
 from config import settings
 from config import settings
@@ -12,6 +13,52 @@ from service.upload_video import upload_video_with_multithreading
 from utils.common import check_proxy, callback_task
 from utils.common import check_proxy, callback_task
 
 
 
 
+# 图片转换成正方形
+def convert_rect_to_square(image_path):
+    # 打开原始图片并转换为 RGBA（支持透明通道）
+    original = Image.open(image_path).convert("RGBA")
+    width, height = original.size
+    if width != height:
+        logger.info(f"{image_path} --> 图片不是正方形")
+        # 计算正方形边长（取长宽中的最大值）
+        square_size = max(width, height)
+
+        # 创建新的正方形透明背景图
+        square_image = Image.new("RGBA", (square_size, square_size), (0, 0, 0, 0))
+
+        # 计算粘贴位置（居中）
+        paste_x = (square_size - width) // 2
+        paste_y = (square_size - height) // 2
+
+        # 将原始图片粘贴到正方形中心
+        square_image.paste(original, (paste_x, paste_y), original)
+        output_path = os.path.join(os.path.dirname(image_path), f"square_{os.path.basename(image_path)}")
+        # 保存结果
+        square_image.save(output_path, "PNG")
+        logger.info(f"{output_path} --> 图片转换完成")
+        return output_path
+    return image_path
+
+def check_image_size(image_path):
+    file_size = os.path.getsize(image_path)
+    max_bytes = 5 * 1024 * 1024
+    if file_size > max_bytes:
+        raise Exception(f"{image_path} --> 图片大小大于5M")
+
+
+def check_image_width_height(image_path, image_type: int = 1):
+    original = Image.open(image_path).convert("RGBA")
+    width, height = original.size
+    if image_type == 1:
+        if width < 600 or height < 600:
+            raise Exception(f"{image_path} --> 图片尺寸小于600*600")
+    elif image_type == 2:
+        if width > 5000 or height > 20000:
+            raise Exception(f"{image_path} --> 图片尺寸大于5000*20000")
+    else:
+        raise Exception(f"{image_path} --> 图片类型错误")
+
+
 def get_local_path(item_id, url):
 def get_local_path(item_id, url):
     folder_path = os.path.join(settings.BASE_PATH, str(item_id))
     folder_path = os.path.join(settings.BASE_PATH, str(item_id))
     if not os.path.exists(folder_path):
     if not os.path.exists(folder_path):
@@ -98,6 +145,8 @@ async def uploadImageAndVideo(task: dict = None):
                 if img_url:
                 if img_url:
                     md5_key = hashlib.md5(img_url.encode()).hexdigest()
                     md5_key = hashlib.md5(img_url.encode()).hexdigest()
                     local_path = get_local_path(item_id, img_url)
                     local_path = get_local_path(item_id, img_url)
+                    check_image_size(local_path)
+                    local_path = convert_rect_to_square(local_path)
                     sku_image_list.append({md5_key: local_path})
                     sku_image_list.append({md5_key: local_path})
 
 
     # 准备主图上传
     # 准备主图上传
@@ -105,6 +154,8 @@ async def uploadImageAndVideo(task: dict = None):
     for url in task.get('images', []):
     for url in task.get('images', []):
         md5_key = hashlib.md5(url.encode()).hexdigest()
         md5_key = hashlib.md5(url.encode()).hexdigest()
         local_path = get_local_path(item_id, url)
         local_path = get_local_path(item_id, url)
+        check_image_size(local_path)
+        check_image_width_height(local_path, 1)
         image_list.append({md5_key: local_path})
         image_list.append({md5_key: local_path})
 
 
     # 准备详情图上传
     # 准备详情图上传
@@ -112,6 +163,8 @@ async def uploadImageAndVideo(task: dict = None):
     for url in task.get('description', []):
     for url in task.get('description', []):
         md5_key = hashlib.md5(url.encode()).hexdigest()
         md5_key = hashlib.md5(url.encode()).hexdigest()
         local_path = get_local_path(item_id, url)
         local_path = get_local_path(item_id, url)
+        check_image_size(local_path)
+        check_image_width_height(local_path, 2)
         description_list.append({md5_key: local_path})
         description_list.append({md5_key: local_path})
     try:
     try:
         # 并行处理所有上传任务
         # 并行处理所有上传任务