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 func_timeout import func_set_timeout from func_timeout import FunctionTimedOut from .multi_threaded_image_saving import ImageSaver import settings # 自己的 AccessKeyId = "LTAI5tCk4p881X8hymj2FYFk" AccessKeySecret = "yBYIYzX8CL24r5ZgEx2AgZyDBmFkIK" def uploadImage(im: Image) -> 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"] # 惠利玛公司的KEY # AccessKeyId = 'LTAI5tCk4p881X8hymj2FYFk' # AccessKeySecret = 'rQMgHwciTN4Gusbpt8CM8tflgsxh1V' # 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') # 使用完成之后记得调用img.close()关闭流 # 场景二，使用任意可访问的url # url = 'https://viapi-test-bj.oss-cn-beijing.aliyuncs.com/viapi-3.0domepic/ocr/RecognizeBankCard/yhk1.jpg' # img = io.BytesIO(urlopen(url).read()) # 4、初始化Request，这里只是以RecognizeBankCard为例，其他能力请使用相应能力对应的类 request = SegmentCommodityAdvanceRequest() request.image_urlobject = img # 5、调用api，注意，recognize_bank_card_advance需要更换为相应能力对应的方法名。方法名是根据能力名称按照一定规范形成的，如能力名称为SegmentCommonImage，对应方法名应该为segment_common_image_advance。 response = self.client.segment_common_image_advance(request, runtime_option) # 获取整体结果 # print(response.body) img.close() return response.body # 获取单个字段，这里只是一个例子，具体能力下的字段需要看具体能力的文档 # print(response.body.data.card_number) # tips: 可通过response.body.__dict__查看属性名称 except Exception as error: # 获取整体报错信息 print("error", error) return None # 获取单个字段 # print(error.code) # tips: 可通过error.__dict__查看属性名称 def get_image_orientation(self, img): # 获取EXIF数据 exif = img._getexif() if exif is not None: # EXIF标签274对应的是Orientation orientation = exif.get(0x0112) print("orientation", orientation) if orientation == 2: # 水平翻转 img = img.transpose(Image.FLIP_LEFT_RIGHT) elif orientation == 3: # 旋转180度 img = img.rotate(180, expand=True) elif orientation == 4: # 垂直翻转 img = img.transpose(Image.FLIP_TOP_BOTTOM) elif orientation == 5: # 水平翻转后顺时针旋转90度 img = img.transpose(Image.FLIP_LEFT_RIGHT).transpose(Image.ROTATE_270) elif orientation == 6: # 顺时针旋转90度 img = img.transpose(Image.ROTATE_270) elif orientation == 7: # 水平翻转后逆时针旋转90度 img = img.transpose(Image.FLIP_LEFT_RIGHT).transpose(Image.ROTATE_90) elif orientation == 8: # 逆时针旋转90度 img = img.transpose(Image.ROTATE_90) else: print("没有EXIF数据或没有方向信息") orientation = 1 return img def get_bo_bg_goods_ultra_background(self, im, is_shadow=False, api_url=None): imageUrl = uploadImage(im) # imageUrl = imageUrl + "?x-oss-process=image/auto-orient,1" if not settings.USER_TOKEN: print("错误：USER_TOKEN 未配置或为空") return None print("图片上传成功", imageUrl) post_headers = {"Authorization": "Bearer " + settings.USER_TOKEN} data = { "image_url": imageUrl, "is_shadow": is_shadow } response = requests.post( api_url, json=data, headers=post_headers ) 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为空") return None cutout_image = data.get("cutout_image") if cutout_image is None: print("旗舰版抠图返回的cutout_image为空") return None return cutout_image def get_no_bg_goods(self, file_path=None, _im=None): # file_path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\_MG_9061.jpg" # file_path_1 = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\_MG_9061_resize.png" # if file_path: # img = open(file_path, 'rb') # if _im: # https://blog.csdn.net/weixin_43411585/article/details/107780941 im = _im # im.save(file_path) img = BytesIO() im.save(img, format='JPEG') # format: PNG or JPEG img.seek(0) # rewind to the start # img = img_byte.getvalue() # im对象转为二进制流 # with open(file_path, "wb") as binary_file: # binary_file.write(im.tobytes()) # file_path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\1.png" # img = open(file_path, 'rb') 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) # img.close() # print("1111111111111", response.body) 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) 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.saver = ImageSaver() self.segment = Segment() @func_set_timeout(40) def get_image_cut_new(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": original_pic.im = original_pic.im.convert("RGB") 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) # 处理失败，需要删除过程图片 return None # 字节流转PIL对象 response = requests.get(image_url) pic = response.content _img_im = Image.open(BytesIO(pic)) # 阿里返回的抠图结果 已转PIL对象 box_size = _img_im.getbbox() new_pp4_im = _img_im.crop(box_size) byte_io = BytesIO() new_pp4_im.save(byte_io, format='PNG') # 将图像保存为 PNG 格式到 BytesIO 对象 byte_io.seek(0) # 将指针重置到流的开头，以便后续读取 return byte_io @func_set_timeout(40) 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) # 处理失败，需要删除过程图片 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: self.saver.save_image( image=_img_im, file_path=out_file_path, quality=100, dpi=(350, 350), _format="PNG" ) # _img_im.save(out_file_path) 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) # 处理失败，需要删除过程图片 return None # 字节流转PIL对象 response = requests.get(image_url) pic = response.content _img_im = Image.open(BytesIO(pic)) # 阿里返回的抠图结果 已转PIL对象 if original_pic.x > 2000: # 原图更大，则需要执行CV处理 # _img_im.show() # 对mask进行放大，然后进行抠图处理 print("对mask进行放大，然后进行抠图处理") transparent_im = Image.new('RGBA', original_pic.im.size, (0, 0, 0, 0)) # original_pic.im.show() # mask.show() _img_im = _img_im.resize((original_pic.x, original_pic.y)) # _img_im.show() transparent_im.paste(original_pic.im, (0, 0), mask=_img_im) # transparent_im.show() # transparent_im.show() _img_im = transparent_im # 上述抠图结果进行拼接 # _img_im.paste(transparent_im, (0, 0), 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) class RemoveUltraBackground: def __init__(self): self.api_url = '/api/ai_image/segment_shadow/segment_service' self.headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } self.saver = ImageSaver() self.segment = Segment() @func_set_timeout(40) def get_image_cut(self, file_path, out_file_path=None): 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 # 字节流转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处理 if out_file_path: self.saver.save_image( image=_img_im, file_path=out_file_path, quality=100, dpi=(350, 350), _format="PNG" ) return _img_im if __name__ == '__main__': r = RemoveUltraBackground() path = r"C:\Users\15001\Desktop\miniso\hb\原始图\hb（1）.JPG" out_path = "{}._no_bg-out.png".format(path) r.get_image_cut(path, out_file_path=out_path)