feat(api): 添加拍照机商品资料上传功能

- 新增uuid和BytesIO导入模块
- 实现cutOutImage抠图函数，集成RemoveUltraBackground服务
- 实现uploadImageIM图片上传函数，支持PNG/JPEG格式处理
- 添加/shoes_compose接口，实现鞋类主图合成上传功能
- 集成compose_shoe_main_image模板合成功能
- 添加阿里云背景移除服务超时控制和错误处理
- 优化图像尺寸调整和掩码处理逻辑
- 更新环境配置从prod到dev模式
- 完善数据库查询条件格式化和代码样式规范

python/api.py

@@ -1,3 +1,4 @@
+import uuid
 from re import search, match
 from natsort.natsort import order_by_index, natsorted
 from sqlalchemy import func
@@ -14,6 +15,7 @@ from utils.utils_func import check_path
 from sockets.socket_client import socket_manager
 from mcu.DeviceControl import DeviceControl
 import time, shutil, os
+from io import BytesIO
 from sqlalchemy import and_, asc, desc
 from functools import partial
 from service.deal_image import DealImage
@@ -2401,3 +2403,84 @@ async def upload_item_to_open_api(params: OpenApiUploadItem):
         raise UnicornException(f"请求开放平台接口失败: {str(e)}")
     except Exception as e:
         raise UnicornException(f"上传异常: {str(e)}")
+
+
+async def cutOutImage(original_image_path):
+    from service.remove_bg_ali import RemoveUltraBackground
+    try:
+        remove_pic_ins = RemoveUltraBackground()
+        im_pillow = remove_pic_ins.ultra_segment_fast_pil(
+            file_path=original_image_path,
+        )
+    except Exception as e:
+        print("抠图异常", e)
+        raise UnicornException("抠图异常")
+    return im_pillow
+
+
+def uploadImageIM(im) -> str:
+    img_byte_io = BytesIO()
+    # 根据图片模式选择保存格式
+    if im.mode == 'RGBA':
+        im.save(img_byte_io, format='PNG')
+    else:
+        im.save(img_byte_io, format='JPEG')
+    img_byte_io.seek(0)  # 重置指针到开头
+
+    post_headers = {"Authorization": settings.USER_TOKEN}
+    url = settings.DOMAIN + "/api/upload"
+
+    # 使用字节流上传
+    resultData = requests.post(
+        url,
+        files={"file": ("image.jpg", img_byte_io, "image/jpeg")},
+        headers=post_headers
+    ).json()
+
+    return resultData["data"]["url"]
+
+
+@app.post("/shoes_compose", description="拍照机商品资料上传")
+async def shoeCompose(params: ShoesComposes):
+    from service.shoe_main_image_compose import compose_shoe_main_image  # noqa: E402
+    settings.USER_TOKEN = "60db012c9dd228dd0b08bea6e6fc915d1b495d8f"
+    # token = "Bearer " + settings.USER_TOKEN
+    tpl = "resources/detail_temp/scene/tpl_1.png"  # 图2：底图模板
+    postData = []
+    for itemIdx, item in enumerate(params.data):
+        print("item", item)
+        main_image_path = item.image_path
+        reference_image = item.reference_image
+        logo = item.logo
+        params.data[itemIdx].logo = logo
+        main_image_pil = await cutOutImage(main_image_path)
+        main_image_url = uploadImageIM(main_image_pil) + "?x-oss-process=image/format,jpg"
+        params.data[itemIdx].image_path = main_image_pil
+        for img in item.images:
+            sub_image_path = img.get("image_path")
+            img['image_path'] = await cutOutImage(sub_image_path)
+        result_url = compose_shoe_main_image(template=tpl,
+                                             data=item.__dict__,
+                                             label_pos="right",
+                                             shoe_scale=0.9,
+                                             font_path="resources/ttl/MiSans-Light.ttf",
+                                             upload=True,
+                                             info_bottom_margin=20,
+                                             label_font_size=30,
+                                             info_bar_height=40,
+                                             info_font_size=34)
+        postData.append({"base_image": main_image_url, "reference_image": reference_image, 'past_image': result_url})
+    headers = {
+        "Authorization": f"Bearer {settings.USER_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+    response = requests.request(
+        "POST",
+        f"{settings.DOMAIN}/api/ai_image/main/auto_photo_scene_images_zj",
+        headers=headers,
+        json={"data": postData},
+    )
+    print("response", response.content)
+    responseData = response.json()
+    return {"code": 0, "msg": "操作成功", "data": responseData.get("data", None)}

python/config.ini

@@ -10,7 +10,7 @@ app_run = api:app
 # 端口号
 port = 7074
 debug = false
-env = prod
+env = dev
 # 线程数
 works = 1
 project = 惠利玛

python/models.py

@@ -219,3 +219,27 @@ class OpenApiUploadItem(BaseModel):
     goods_art_no: str = Field(default=None, description="货号")
     company_id: str = Field(default=None, description="混淆企业ID")
     item_type: int = Field(default=None, description="上传类型：1商品 2材料")
+
+
+class ShoesComposeImageItem(BaseModel):
+    """拍照机商品资料上传"""
+    image_path: str = Field(default=None, description="货号")
+    color_name: str = Field(default=None, description="尺码")
+
+
+class ShoesComposeItem(BaseModel):
+    """拍照机商品资料上传"""
+    logo: str = Field(default=None, description="logo")
+    number_mid: dict = Field(default=None, description="中童货号")
+    size_mid: dict = Field(default=None, description="中童尺码")
+    number_big: dict = Field(default=None, description="大童货号")
+    size_big: dict = Field(default=None, description="大童尺码")
+    box: dict = Field(default=None, description="装箱数量")
+    image_path: str = Field(default=None, description="主图")
+    reference_image: str = Field(default=None, description="参考图")
+    images: list[dict] = Field(default=None, description="图片货号列表")
+
+
+class ShoesComposes(BaseModel):
+    """拍照机商品资料上传"""
+    data: list[ShoesComposeItem] = Field(default=None, description="结构体")

python/service/deal_image.py

@@ -1,11 +1,11 @@
-import os,re
+import os, re
 
-from  natsort import natsorted,ns
+from natsort import natsorted, ns
 import shutil
 import exifread
 import time
 import datetime
-from databases import PhotoRecord,SqlQuery,CRUD
+from databases import PhotoRecord, SqlQuery, CRUD
 import settings
 import copy
 import json
@@ -17,6 +17,7 @@ from PIL import Image
 from utils.utils_func import check_path
 from .base_deal import BaseDealImage
 from middleware import UnicornException
+
 _Type = ['.png', '.PNG', '.jpg', '.JPG', '.gif', '.GIF', ".jpge", ".JPGE"]
 
 
@@ -26,16 +27,20 @@ class DealImage(BaseDealImage):
         super().__init__()
         self.image_dir = image_dir
         self.header = None
+
     def check_path(self, image_dir: str):
         if not os.path.exists(image_dir):
             os.mkdir(image_dir)
         return True
+
     def list_dir(self, path):
         listdir = os.listdir(path)
-        sort_result = sorted(listdir, key=lambda x: int(re.findall(r'^(\d+)_', x)[0]) if re.findall(r'^(\d+)_', x) else 0)
+        sort_result = sorted(listdir,
+                             key=lambda x: int(re.findall(r'^(\d+)_', x)[0]) if re.findall(r'^(\d+)_', x) else 0)
         # print("listdir   排序排序排序排序排序排序", sort_result)
         # return natsorted(listdir, alg=ns.PATH)
         return sort_result
+
     def get_date_time_original(self, file_path):
         with open(file_path, 'rb') as file_data:
             tags = exifread.process_file(file_data)
@@ -54,7 +59,7 @@ class DealImage(BaseDealImage):
         configModel = CRUD(PhotoRecord)
         result = configModel.read(
             session,
-            conditions={"goods_art_no": goods_art_no,"delete_time": None},
+            conditions={"goods_art_no": goods_art_no, "delete_time": None},
             order_by="id",
             ascending=True,
         )
@@ -63,6 +68,7 @@ class DealImage(BaseDealImage):
             return result.goods_art_no, result.image_index, result.image_deal_mode
         else:
             return None
+
     def get_goods_art_no_files(self, goods_art_no):
         # time_array = time.strptime(date_time_original, "%Y:%m:%d %H:%M:%S")
 
@@ -73,15 +79,16 @@ class DealImage(BaseDealImage):
         configModel = CRUD(PhotoRecord)
         result = configModel.read_all(
             session,
-            conditions={"goods_art_no": goods_art_no,"delete_time": None},
+            conditions={"goods_art_no": goods_art_no, "delete_time": None},
             order_by="id",
             ascending=True,
         )
         session.close()
         if result:
-            return [{"image_path":item.image_path,"id":item.id,"image_index":item.image_index} for item in result]
+            return [{"image_path": item.image_path, "id": item.id, "image_index": item.image_index} for item in result]
         else:
             return []
+
     def get_data_from_hqt_with_goods_art_no(self, goods_art_no_list):
         _goods_art_no_list = copy.deepcopy(goods_art_no_list)
         _list = []
@@ -196,6 +203,7 @@ class DealImage(BaseDealImage):
             shutil.move(old_image_path, original_image_path)
         except Exception as e:
             print(f"文件操作异常:{e}")
+
     def copy_images(self, goods_art_no, goods_art_no_path, old_image_path):
         """
         步骤：
@@ -221,18 +229,19 @@ class DealImage(BaseDealImage):
         original_image_path = "{}/原始图/{}{}".format(goods_art_no_path, new_file_name, e)
         try:
             shutil.copy2(old_image_path, original_image_path)
-            return True,"移动成功"
+            return True, "移动成功"
         except Exception as e:
             print(f"文件操作异常:{e}")
-            return False,f"货号[{goods_art_no}]原始图{old_image_path}不存在或已被删除"
-    def dealMoveImage(self, image_dir: str, callback_func=None,goods_art_no=None) -> dict:
+            return False, f"货号[{goods_art_no}]原始图{old_image_path}不存在或已被删除"
+
+    def dealMoveImage(self, image_dir: str, callback_func=None, goods_art_no=None) -> dict:
         if not self.check_path(image_dir=image_dir + "/历史"):
             raise UnicornException("文件夹创建失败")
 
         # 遍历目标文件夹，获取有拍摄信息的图片，并按拍摄时间排序
         files = self.list_dir(image_dir)
         original_photo_list = []  # 原始图片列表
-        for file_idx,file in enumerate(files):
+        for file_idx, file in enumerate(files):
             # -----图片清洗
             file_path = image_dir + "/" + file
             if os.path.isdir(file_path):  # 忽略文件夹
@@ -249,7 +258,7 @@ class DealImage(BaseDealImage):
                 if _data:
                     # 能匹配上数据库
                     goods_art_no, image_index, image_deal_mode = _data
-                    print("832 与数据库匹配goods_art_no", file_path,file_name, date_time_original, goods_art_no)
+                    print("832 与数据库匹配goods_art_no", file_path, file_name, date_time_original, goods_art_no)
                     original_photo_list.append({"file_path": file_path,
                                                 "file": file,
                                                 "date_time_original": date_time_original,
@@ -280,14 +289,15 @@ class DealImage(BaseDealImage):
         total_num = len(original_photo_list)
         # 当天日期作为文件夹
         seconds = time.time()
-        output_path = "{}/{f_name}".format(settings.OUTPUT_DIR,f_name=time.strftime("%Y-%m-%d", time.localtime(seconds)))
+        output_path = "{}/{f_name}".format(settings.OUTPUT_DIR,
+                                           f_name=time.strftime("%Y-%m-%d", time.localtime(seconds)))
 
         # 遍历每个匹配好的数据进行处理
-        print(f"original_photo_list:",original_photo_list)
+        print(f"original_photo_list:", original_photo_list)
         n = 0
         for photo_dict in original_photo_list:
             n += 1
-            print("处理dict:",photo_dict)
+            print("处理dict:", photo_dict)
             # 进度条
             goods_art_no = photo_dict["goods_art_no"]
             original_image_path = photo_dict["file_path"]
@@ -325,13 +335,61 @@ class DealImage(BaseDealImage):
         # 完成处理
         # self.set_state(state_value=2)
         return True, output_path
-    def dealMoveImageV2(self,goods_art_no=None) -> dict:
+
+    def dealMoveImageV2(self, goods_art_no=None) -> dict:
+        # 遍历目标文件夹，获取有拍摄信息的图片，并按拍摄时间排序
+        original_photo_list = []  # 原始图片列表
+        _data = self.get_goods_art_no_files(goods_art_no)
+        if _data:
+            for file_idx, dataItem in enumerate(_data):
+                # 能匹配上数据库
+                file_path = dataItem['image_path']
+                original_photo_list.append({"file_path": file_path,
+                                            "goods_art_no": goods_art_no,
+                                            "image_index": file_idx,
+                                            })
+        if not original_photo_list:
+            raise UnicornException("没有任何匹配的图片")
+        # 当天日期作为文件夹
+        seconds = time.time()
+        output_path = "{}/{f_name}".format(settings.OUTPUT_DIR,
+                                           f_name=time.strftime("%Y-%m-%d", time.localtime(seconds)))
+
+        # 遍历每个匹配好的数据进行处理
+        print(f"original_photo_list:", original_photo_list)
+        n = 0
+        for photo_dict in original_photo_list:
+            n += 1
+            print("处理dict:", photo_dict)
+            # 进度条
+            goods_art_no = photo_dict["goods_art_no"]
+            original_image_path = photo_dict["file_path"]
+            goods_art_no_path = "{output_path}/{goods_art_no}".format(output_path=output_path,
+                                                                      goods_art_no=goods_art_no)
+            # 创建货号下的一系列文件夹
+            self.create_folder(goods_art_no_path)
+            # 重命名并进行移动
+            print("开始移动：{}  {} 命名为：{}".format(goods_art_no, original_image_path, goods_art_no_path))
+            result, message = self.copy_images(goods_art_no, goods_art_no_path, original_image_path)  # 货号、货号文件路径、原始图路径
+            if not result:
+                return result, message
+            time.sleep(0.1)
+            print("已完成移动处理")
+        if n != 0:
+            # 自动生成一个货号表
+            print("output_path", output_path)
+            self.deal(output_path)
+        # 完成处理
+        # self.set_state(state_value=2)
+        return True, output_path
+
+    def dealImagePath(self, goods_art_no=None) -> dict:
         # 遍历目标文件夹，获取有拍摄信息的图片，并按拍摄时间排序
         original_photo_list = []  # 原始图片列表
         _data = self.get_goods_art_no_files(goods_art_no)
         if _data:
             for file_idx, dataItem in enumerate(_data):
-            # 能匹配上数据库
+                # 能匹配上数据库
                 file_path = dataItem['image_path']
                 original_photo_list.append({"file_path": file_path,
                                             "goods_art_no": goods_art_no,
@@ -341,14 +399,15 @@ class DealImage(BaseDealImage):
             raise UnicornException("没有任何匹配的图片")
         # 当天日期作为文件夹
         seconds = time.time()
-        output_path = "{}/{f_name}".format(settings.OUTPUT_DIR,f_name=time.strftime("%Y-%m-%d", time.localtime(seconds)))
+        output_path = "{}/{f_name}".format(settings.OUTPUT_DIR,
+                                           f_name=time.strftime("%Y-%m-%d", time.localtime(seconds)))
 
         # 遍历每个匹配好的数据进行处理
-        print(f"original_photo_list:",original_photo_list)
+        print(f"original_photo_list:", original_photo_list)
         n = 0
         for photo_dict in original_photo_list:
             n += 1
-            print("处理dict:",photo_dict)
+            print("处理dict:", photo_dict)
             # 进度条
             goods_art_no = photo_dict["goods_art_no"]
             original_image_path = photo_dict["file_path"]
@@ -370,6 +429,7 @@ class DealImage(BaseDealImage):
         # 完成处理
         # self.set_state(state_value=2)
         return True, output_path
+
     def deal(cls, dir_path):
 
         # print("dir_path", dir_path)
@@ -412,7 +472,9 @@ class DealImage(BaseDealImage):
                     continue
 
                 all_original_images = os.listdir(_path)  # 遍历货号原始图文件夹
-                sort_result = sorted(all_original_images, key=lambda x: int(re.findall(r'（(\d+)）', x)[0]) if re.findall(r'（(\d+)）', x) else 0)
+                sort_result = sorted(all_original_images,
+                                     key=lambda x: int(re.findall(r'（(\d+)）', x)[0]) if re.findall(r'（(\d+)）',
+                                                                                                   x) else 0)
                 goods_pic_total = len(sort_result)
                 if not sort_result:
                     continue
@@ -492,9 +554,9 @@ class DealImage(BaseDealImage):
                             "y_scale": y_scale,
                             "x_offset": 5,
                             "y_offset": 5,
-                            "object_position":1,
+                            "object_position": 1,
                         },
-                        )
+                    )
 
                 except:
                     pass

python/service/remove_bg_ali.py

@@ -140,7 +140,7 @@ class Segment(object):
         if not settings.USER_TOKEN:
             print("错误：USER_TOKEN 未配置或为空")
             return None
-        print("图片上传成功", imageUrl)
+        # print("图片上传成功", imageUrl)
         post_headers = {"Authorization": "Bearer " + settings.USER_TOKEN}
         data = {
             "image_url": imageUrl,
@@ -149,19 +149,19 @@ class Segment(object):
         response = requests.post(
             api_url, json=data, headers=post_headers
         )
-        print(f"响应状态码: {response.status_code}")
-        print(f"响应内容: {response.text[:500]}")  # 只打印前500字符
+        # print(f"响应状态码: {response.status_code}")
+        # print(f"响应内容: {response.text[:500]}")  # 只打印前500字符
 
         resultData = response.json()
         print("旗舰版抠图请求", resultData)
         # 安全地获取返回值
         data = resultData.get("data")
         if data is None:
-            print("旗舰版抠图返回的data为空")
+            # print("旗舰版抠图返回的data为空")
             return None
         cutout_image = data.get("cutout_image")
         if cutout_image is None:
-            print("旗舰版抠图返回的cutout_image为空")
+            # print("旗舰版抠图返回的cutout_image为空")
             return None
         return cutout_image
 
@@ -726,6 +726,76 @@ class RemoveUltraBackground:
             )
         return _img_im
 
+    @func_set_timeout(40)
+    def ultra_segment_fast_pil(self, file_path):
+        '''
+        旗舰抠图快速版
+        '''
+        original_pic = Picture(file_path)
+        original_pic.im = self.segment.get_image_orientation(original_pic.im)
+        original_pic.x, original_pic.y = original_pic.im.size
+        if original_pic.im.mode != "RGB":
+            print("抠图图片不能是PNG")
+            return None
+
+        new_pic = copy.copy(original_pic)
+        after_need_resize = False
+        if new_pic.x > new_pic.y:
+            if new_pic.x > 2000:
+                after_need_resize = True
+                new_pic.resize(2000)
+        else:
+            if new_pic.y > 2000:
+                after_need_resize = True
+                new_pic.resize_by_heigh(heigh=2000)
+        # print("使用旗舰版抠图")
+        try:
+            api_url = f"{settings.DOMAIN}{self.api_url}"
+            image_url = self.segment.get_bo_bg_goods_ultra_background(im=new_pic.im, api_url=api_url)
+        except BaseException as e:
+            print("旗舰版抠图异常：", e)
+            # 处理失败，需要删除过程图片
+            return None
+        if image_url is None:
+            return None
+        # 字节流转PIL对象
+        # print("image_url", image_url)
+        response = requests.get(image_url)
+        pic = response.content
+        _img_im = Image.open(BytesIO(pic))  # 阿里返回的抠图结果 已转PIL对象
+        # 原图更大，则需要执行CV处理
+        if after_need_resize:
+            # 将抠图结果转成mask
+            # _img_im = Image.open(_path)
+            # 将抠图结果放大到原始图大小
+            _img_im = _img_im.resize(original_pic.im.size)
+            new_big_mask = Image.new('RGB', _img_im.size, (0, 0, 0))
+            white = Image.new('RGB', _img_im.size, (255, 255, 255))
+            new_big_mask.paste(white, mask=_img_im.split()[3])
+
+            # ---------制作选区缩小的mask
+            # mask = cv2.imread(mask_path)
+            # mask = cv2.cvtColor(mask, cv2.COLOR_BGR2GRAY)
+            mask = cv2.cvtColor(np.asarray(new_big_mask), cv2.COLOR_BGR2GRAY)  # 将PIL 格式转换为 CV对象
+            mask[mask != 255] = 0
+            # 黑白反转
+            # mask = 255 - mask
+            # 选区缩小10
+            kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (10, 10))
+            erode_im = cv2.morphologyEx(mask, cv2.MORPH_ERODE, kernel)
+
+            # -------再进行抠图处理
+            mask = Image.fromarray(cv2.cvtColor(erode_im, cv2.COLOR_GRAY2RGBA))  # CV 对象转 PIL
+            transparent_im = Image.new('RGBA', original_pic.im.size, (0, 0, 0, 0))
+            # original_pic.im.show()
+            # mask.show()
+            transparent_im.paste(original_pic.im, (0, 0), mask.convert('L'))
+            # transparent_im.show()
+            # 上述抠图结果进行拼接
+            _img_im.paste(transparent_im, (0, 0), transparent_im)
+        # 原图更大，则需要执行CV处理
+        return _img_im
+
     def run_ultra_segment(self, file_path, out_file_path):
         # 直接调用抠图
         time.sleep(0.01)