TuKouKouClient/python/services/other/remove_bg_ali.py

import copy
import json
import os
from PIL import Image
from alibabacloud_imageseg20191230.client import Client as imageseg20191230Client
from alibabacloud_imageseg20191230.models import SegmentCommodityAdvanceRequest
from alibabacloud_imageseg20191230 import models as imageseg_20191230_models
from alibabacloud_tea_util.models import RuntimeOptions
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.models import Config
from alibabacloud_tea_util import models as util_models
import requests
from io import BytesIO
import cv2
import numpy as np
from .module_online_data import GetOnlineData
from Tea.exceptions import TeaException
from middleware import UnicornException

# todo 获取密钥
# key_id,key_secret = GetOnlineData().get_cutout_image_config()
# 惠利玛的key
AccessKeyId = "LTAI5tBdDVT9Wc5idJXdGHjw"
AccessKeySecret = "bCSotQ7eAztOxx6AqHwJJPsb0hkECe"
ALI_MAPPING = {"tops": "上衣", "coat": "外套", "skirt": "裙装", "pants": "裤装"}
ALI_CLOTH_CLASSES = ["tops", "coat", "skirt", "pants"]

# https://help.aliyun.com/zh/viapi/developer-reference/python?spm=a2c4g.11186623.0.i0#task-2252575
# pip install alibabacloud_goodstech20191230
# pip install alibabacloud_tea_openapi
# pip install alibabacloud_tea_util


class Segment(object):
    def __init__(self):
        self.client = self.create_client()

    def get_no_bg_common(self, file_path):
        # 初始化RuntimeObject
        runtime_option = RuntimeOptions()
        try:
            # 场景一：文件在本地
            img = open(file_path, "rb")
            request = SegmentCommodityAdvanceRequest()
            request.image_urlobject = img
            response = self.client.segment_common_image_advance(request, runtime_option)
            # 获取整体结果
            # print(response.body)
            img.close()
            return response.body
        except Exception as error:
            # 获取整体报错信息
            print("error", error)
            return None

    def get_no_bg_goods(self, file_path=None, _im=None):
        im = _im
        img = BytesIO()
        im.save(img, format="JPEG")  # format: PNG or JPEG
        img.seek(0)  # rewind to the start
        request = imageseg_20191230_models.SegmentCommodityAdvanceRequest()
        request.image_urlobject = img
        client = self.create_client()
        # 5、调用api，注意，recognize_bank_card_advance需要更换为相应能力对应的方法名。方法名是根据能力名称按照一定规范形成的，如能力名称为SegmentCommonImage，对应方法名应该为segment_common_image_advance。
        runtime = util_models.RuntimeOptions()
        response = client.segment_commodity_advance(request, runtime)
        return response.body

    def get_no_bg_cloths(self, _im=None):
        im = _im
        img = BytesIO()
        im.save(img, format="JPEG")  # format: PNG or JPEG
        img.seek(0)  # rewind to the start
        request = imageseg_20191230_models.SegmentClothAdvanceRequest()
        request.image_urlobject = img
        request.cloth_class = ALI_CLOTH_CLASSES
        client = self.create_client()
        # 5、调用api，注意，recognize_bank_card_advance需要更换为相应能力对应的方法名。方法名是根据能力名称按照一定规范形成的，如能力名称为SegmentCommonImage，对应方法名应该为segment_common_image_advance。
        runtime = util_models.RuntimeOptions()
        response = client.segment_cloth_advance(request, runtime)
        return response.body

    def create_client(self):
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 必填，您的 AccessKey ID,
            access_key_id=AccessKeyId,
            # 必填，您的 AccessKey Secret,
            access_key_secret=AccessKeySecret,
        )
        # 访问的域名
        config.endpoint = f"imageseg.cn-shanghai.aliyuncs.com"
        return imageseg20191230Client(config)


class Picture:
    def __init__(self, in_path, im=None):
        if im:
            self.im = im
        else:
            self.im = Image.open(in_path).convert("RGB")
        self.x, self.y = self.im.size
        # print(self.x, self.y)

    def save_img(self, outpath, quality=90):
        # self.im = self.im.convert("RGB")
        self.im.save(outpath, quality=quality)

    def resize(self, width):
        re_x = int(width)
        re_y = int(self.y * re_x / self.x)
        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
        self.x, self.y = self.im.size

    def resize_by_heigh(self, heigh):
        re_y = int(heigh)
        re_x = int(self.x * re_y / self.y)
        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
        self.x, self.y = self.im.size


class RemoveBgALi(object):
    def __init__(self):
        self.segment = Segment()

    def get_image_cut(self, file_path, out_file_path=None, original_im=None):
        if original_im:
            original_pic = Picture(in_path=None, im=original_im)
        else:
            original_pic = Picture(file_path)

        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)

        # new_pic.im.show()
        body = self.segment.get_no_bg_goods(file_path=None, _im=new_pic.im)
        body = eval(str(body))
        try:
            image_url = body["Data"]["ImageURL"]
        except BaseException as e:
            print("阿里抠图错误：", e)
            # todo 处理失败，需要删除过程图片
            return None
        # 字节流转PIL对象
        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)
            # _img_im.show("11111111111111111111111")
        if out_file_path:
            _img_im.save(out_file_path)
        return _img_im

    def get_image_cut_cloths(self, file_path, out_file_path=None, original_im=None):
        if original_im:
            original_pic = Picture(in_path=None, im=original_im)
        else:
            original_pic = Picture(file_path)

        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)
        try:
            body = self.segment.get_no_bg_cloths(_im=new_pic.im)
            body = eval(str(body))
            clothsArray = []
            Elements = body["Data"]["Elements"]
            for idx, clothItem in enumerate(Elements):
                if idx == 0:
                    ImageURL = clothItem.get("ImageURL")
                    print("clothItem", clothItem)
                    response = requests.get(ImageURL)
                    pic = response.content
                    _img_im = Image.open(BytesIO(pic))
                    clothsArray.append(
                        {"type": "all", "cn_name": "合并图", "pilImage": _img_im}
                    )
                else:
                    ClassUrl = clothItem.get("ClassUrl")
                    classKeys = ClassUrl.keys()
                    for classKey in classKeys:
                        keyUrl = ClassUrl.get(classKey)
                        response = requests.get(keyUrl)
                        pic = response.content
                        mask_image = Image.open(BytesIO(pic))
                        if self.isEmptyMask(mask_image.convert("RGB")):
                            continue
                        if mask_image.mode != "L":
                            mask_image = mask_image.convert("L")
                        # 使用蒙版图像对原图进行抠图
                        result_image = new_pic.im.copy()
                        result_image.putalpha(mask_image)
                        clothsArray.append(
                            {
                                "type": classKey,
                                "cn_name": ALI_MAPPING[classKey],
                                "pilImage": result_image,
                            }
                        )
        except TeaException as teaError:
            raise UnicornException(teaError.data.get("Message"))
            return None
        except BaseException as e:
            # todo 处理失败，需要删除过程图片
            print("Exception",e)
            return None
        image_arrs = []
        for _, item in enumerate(clothsArray):
            _img_im = item["pilImage"]
            type = item["type"]
            cn_name = item["cn_name"]
            # 原图更大，则需要执行CV处理
            if after_need_resize:
                _img_im = self.clothsDispose(
                    _img_im, original_pic=original_pic, out_file_path=out_file_path
                )
            image_arrs.append({"image_obj": _img_im, "type": type, "cn_name": cn_name})
        return image_arrs

    def isEmptyMask(self, img):
        data = np.array(img)
        max_index = np.unravel_index(np.argmax(data, axis=None), data.shape)
        max_value = data[max_index]
        return True if max_value == 0 else False

    def clothsDispose(self, _img_im, original_pic, out_file_path):
        # 将抠图结果转成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)
        # _img_im.show("11111111111111111111111")
        return _img_im

    def get_image_cut1(self, file_path, out_file_path=None):
        original_pic = Picture(file_path)
        new_pic = copy.copy(original_pic)
        if new_pic.x > 2000:
            new_pic.resize(2000)
        # new_pic.im.show()
        body = self.segment.get_no_bg_goods(file_path=out_file_path, _im=new_pic.im)
        body = eval(str(body))
        try:
            image_url = body["Data"]["ImageURL"]
        except BaseException as e:
            print("阿里抠图错误：", e)
            # todo 处理失败，需要删除过程图片
            return None
        # 字节流转PIL对象
        response = requests.get(image_url)
        pic = response.content
        _img_im = Image.open(BytesIO(pic))  # 阿里返回的抠图结果 已转PIL对象

        if original_pic.x > 2000:
            # 原图更大，则需要执行CV处理
            # 对mask进行放大，然后进行抠图处理
            print("对mask进行放大，然后进行抠图处理")
            transparent_im = Image.new("RGBA", original_pic.im.size, (0, 0, 0, 0))
            _img_im = _img_im.resize((original_pic.x, original_pic.y))
            transparent_im.paste(original_pic.im, (0, 0), mask=_img_im)
            _img_im = transparent_im
            pass

        _img_im.save(out_file_path)
        return _img_im

    def download_picture(self, url, out_path):
        response = requests.get(url)
        pic = response.content
        with open(out_path, "wb") as f:
            f.write(pic)


if __name__ == "__main__":
    r = RemoveBgALi()
    path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\auto_capture_V2\IPC\test\171112057820408.png"
    out_path = "{}._no_bg-out.png".format(path)
    r.get_image_cut(path, out_file_path=out_path)