import os, re from natsort import natsorted, ns import shutil import exifread import time import datetime from databases import PhotoRecord, SqlQuery, CRUD import settings import copy import json import requests from service.pic_deal import Picture import xlsxwriter 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"] class DealImage(BaseDealImage): def __init__(self, image_dir=None): 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) # 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) if "EXIF DateTimeOriginal" in tags: return str(tags["EXIF DateTimeOriginal"]) else: return False def get_goods_art_no(self, goods_art_no): # time_array = time.strptime(date_time_original, "%Y:%m:%d %H:%M:%S") # time_array = time.mktime(time_array) # datetime_obj = datetime.datetime.fromtimestamp(time_array) session = SqlQuery() configModel = CRUD(PhotoRecord) result = configModel.read( session, conditions={"goods_art_no": goods_art_no, "delete_time": None}, order_by="id", ascending=True, ) session.close() if result: 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") # time_array = time.mktime(time_array) # datetime_obj = datetime.datetime.fromtimestamp(time_array) session = SqlQuery() configModel = CRUD(PhotoRecord) result = configModel.read_all( session, 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] 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 = [] # 单次请求数少于20个 goods_art_no_dict = {} while _goods_art_no_list: goods_art_no = _goods_art_no_list.pop() _list.append(goods_art_no) if len(_list) == 20 or len(_goods_art_no_list) == 0: online_goods_art_data = self.get_goods_art_no_info( goods_art_list=_list ) if online_goods_art_data: for _goods_art_no in online_goods_art_data: goods_art_no_dict[_goods_art_no] = online_goods_art_data[ _goods_art_no ] _list = [] return goods_art_no_dict def get_goods_art_no_info(self, numbers_list=None, goods_art_list=None): # 获取商品基础信息，入参为商品的编号 url = "{domain}/api/backend/goods_client/goods_query".format( domain=settings.DOMAIN ) data = { 'goods_art_list': goods_art_list } data = json.dumps(data) _s = requests.session().post(url=url, data=data, headers=self.header) response_data = _s.json() goods_number_data = {} # ["", "", "", "", "", "", "", "", "", "", "", ] if "data" not in response_data: return {} for data in response_data["data"]: goods_number_data[data["goods_art_no"]] = {} goods_number_data[data["goods_art_no"]]["商品货号"] = data["goods_art_no"].upper() goods_number_data[data["goods_art_no"]]["款号"] = data["goods_number"].upper() goods_number_data[data["goods_art_no"]]["商品面料"] = data["fabric"] goods_number_data[data["goods_art_no"]]["商品内里"] = data["lining"] goods_number_data[data["goods_art_no"]]["商品鞋底"] = data["sole"] goods_number_data[data["goods_art_no"]]["鞋垫"] = data["insole"] goods_number_data[data["goods_art_no"]]["颜色名称"] = data["color"] return goods_number_data def get_data_from_hqt(self, goods_number_list): _goods_number_list = copy.deepcopy(goods_number_list) _list = [] # 单次请求数少于20个 goods_number_dict = {} while _goods_number_list: goods_art_no = _goods_number_list.pop() if "NUM" in goods_art_no: goods_art_no = goods_art_no.replace("NUM", "") _list.append(goods_art_no) if len(_list) == 20 or len(_goods_number_list) == 0: online_goods_art_data = self.get_goods_art_no_info( numbers_list=_list ) if online_goods_art_data: for number in online_goods_art_data: goods_number_dict["NUM" + number] = online_goods_art_data[ number ] _list = [] return goods_number_dict def create_folder(self, path): def check_folder(__path): if not os.path.exists(__path): os.makedirs(__path) return False return True # 文件夹不存在，创建货号子集文件夹 if not check_folder(path): for name in ["原始图", "原始图_已抠图", "800x800", "200images"]: other_path = path + "/" + name check_folder(other_path) def move_images(self, goods_art_no, goods_art_no_path, old_image_path): """ 步骤： 1、移动到原始图 Args: goods_art_no: goods_art_no_path: old_image_path: Returns: """ # 移动到原始图 file = os.path.split(old_image_path)[1] # 扩展名 e = os.path.splitext(file)[1] print("self.goods_images_count_dict", self.goods_images_count_dict) # 获取图片序列 self.goods_images_count_dict[goods_art_no] += 1 # A9999（1）.jpg new_file_name = "{}（{}）".format(goods_art_no, self.goods_images_count_dict[goods_art_no]) original_image_path = "{}/原始图/{}{}".format(goods_art_no_path, new_file_name, e) try: 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): """ 步骤： 1、移动到原始图 Args: goods_art_no: goods_art_no_path: old_image_path: Returns: """ # 移动到原始图 file = os.path.split(old_image_path)[1] # 扩展名 e = os.path.splitext(file)[1] print("self.goods_images_count_dict", self.goods_images_count_dict) # 获取图片序列 self.goods_images_count_dict[goods_art_no] += 1 # A9999（1）.jpg new_file_name = "{}（{}）".format(goods_art_no, self.goods_images_count_dict[goods_art_no]) check_path("{}/原始图".format(goods_art_no_path)) original_image_path = "{}/原始图/{}{}".format(goods_art_no_path, new_file_name, e) try: shutil.copy2(old_image_path, original_image_path) 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: 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): # -----图片清洗 file_path = image_dir + "/" + file if os.path.isdir(file_path): # 忽略文件夹 continue file_name, suffix = os.path.splitext(file) if suffix not in _Type: # 非图片进行移除 shutil.move(file_path, image_dir + "/历史/" + file) continue date_time_original = self.get_date_time_original(file_path) # 获取照片拍照时间 if date_time_original: # 基于照片的时间，与数据库匹配goods_art_no _data = self.get_goods_art_no(goods_art_no) 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) original_photo_list.append({"file_path": file_path, "file": file, "date_time_original": date_time_original, "goods_art_no": goods_art_no, "image_index": file_idx, "real_goods_art_no": "", "real_goods_number": "", }) else: # 匹配不上报错 # self.show_progress_detail("图片：{} 无法对应货号，不做处理".format(file)) if callback_func: callback_func("图片：{} 无对应货号".format(file)) # shutil.move(photo_dict["file_path"], self.image_dir + "/历史/" + photo_dict["file"]) continue else: shutil.move(file_path, image_dir + "/历史/" + file) if not original_photo_list: raise UnicornException("没有任何匹配的图片") return False, "没有任何匹配的图片" # 排序需要基于拍照的文件序号进行处理 # original_photo_list.sort( # key=lambda x: "{}-{}-{}".format(x["goods_art_no"], x["image_index"], x["file"])) # print(original_photo_list) # 对有拍摄信息的图片进行数据库比对，如有比对上，则移动至货号文件夹，否则移入历史文件夹 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))) # 遍历每个匹配好的数据进行处理 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"] # 输出货号文件夹 if photo_dict["real_goods_art_no"]: goods_art_no = "{}@{}".format(photo_dict["real_goods_art_no"], photo_dict["real_goods_number"]) 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)) self.move_images(goods_art_no, goods_art_no_path, original_image_path) # 货号、货号文件路径、原始图路径 time.sleep(0.2) # self.progress_sign.emit({"type": "移动原始图片", "progress_bar_value": int(n / total_num * 100)}) # self.show_progress_detail("货号{} 相关文件夹创建完成,已移动原图~".format(goods_art_no)) if callback_func: callback_func("货号{} 相关文件夹创建完成,已移动原图~".format(goods_art_no)) print("已完成移动处理") if n != 0: # if settings.MattingPics: # # 检查所有未处理的货号文件夹，查看是否有完成图片加工处理 # self.deal_images() # 自动生成一个货号表 print("output_path", output_path) self.deal(output_path) # 完成处理 # self.set_state(state_value=2) return True, output_path 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, "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 deal(cls, dir_path): # print("dir_path", dir_path) out_excel_data = [] for goods_art_no_folder in os.listdir(dir_path): # 遍历货号文件夹集合 if not os.path.isdir( "{}/{}".format(dir_path, goods_art_no_folder) ): # 非文件夹进行过滤 continue if "软件" in goods_art_no_folder: continue # print("goods_art_no_folder", goods_art_no_folder) # 如果存在800的主图，则优先进行使用 big_image_folder_path = "{}/{}/800x800".format( dir_path, goods_art_no_folder ) if not os.path.exists(big_image_folder_path): os.mkdir(big_image_folder_path) all_big_images = os.listdir(big_image_folder_path) goods_pic_total = len(all_big_images) _Type = [".png", ".PNG", ".jpg", ".JPG", ".gif", ".GIF", ".jpge", ".JPGE"] thumb_image_file_path = None # print("all_big_images",all_big_images) if all_big_images: for _file in all_big_images: # print(_file) file_name, e = os.path.splitext(_file) # print(file_name, e) if e in _Type: thumb_image_file_path = "{}/{}/800x800/{}".format( dir_path, goods_art_no_folder, _file ) break # 如果不存在主图则进行使用原始图 if thumb_image_file_path is None: _path = "{}/{}/原始图".format(dir_path, goods_art_no_folder) if not os.path.exists(_path): 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) goods_pic_total = len(sort_result) if not sort_result: continue image_file = sort_result[0] # 取第一个货号图 image_file_path = "{}/{}/原始图/{}".format( dir_path, goods_art_no_folder, image_file ) # print("\033[1;32;40m 需要处理的200images,路径： \033[0m",image_file_path,all_original_images,sort_result) if not os.path.exists( "{}/{}/200images".format(dir_path, goods_art_no_folder) ): os.mkdir("{}/{}/200images".format(dir_path, goods_art_no_folder)) thumb_image_file_path = "{}/{}/200images/{}".format( dir_path, goods_art_no_folder, image_file ) if not os.path.exists(thumb_image_file_path): # 开始触发进行压缩生成文件 shutil.copy(image_file_path, thumb_image_file_path) # 复制文件 pic = Picture(thumb_image_file_path) pic.resize(width=600) pic.save_img(thumb_image_file_path) # print("thumb_image_file_path", thumb_image_file_path) goods_number = "" if "@" in goods_art_no_folder: _ = goods_art_no_folder.split("@") goods_art_no_folder = _[0] goods_number = _[1].replace("NUM", "") out_excel_data.append( [ goods_number, goods_art_no_folder, thumb_image_file_path, goods_pic_total, ] ) if out_excel_data: out_excel_path = "{}/货号表-{}.xlsx".format(dir_path, time.time()) options = { "default_format_properties": { "align": "left", "valign": "vcenter", "text_wrap": True, } } book = xlsxwriter.Workbook(filename=out_excel_path, options=options) sheet = book.add_worksheet("sheet1") # sheet.freeze_panes(1, 2) sheet.set_column("B:B", 17) sheet.set_column("D:D", 20) sheet.write_row("A1", ["编号", "原货号", "新货号", "缩略图", "原始图张数"]) for index, data in enumerate(out_excel_data): # print(data) goods_number, goods_art_no, image_file, goods_pic_total = data try: im = Image.open(image_file) im_x, im_y = im.size image_width = 100 image_height = int(im_y * image_width / im_x) sheet.set_row(index + 1, 95) x_scale = round( image_width / im_x, 2 ) # 固定宽度/要插入的原始图片宽 y_scale = round( image_height / im_y, 2 ) # 固定高度/要插入的原始图片高 sheet.insert_image( index + 1, 3, image_file, { "x_scale": x_scale, "y_scale": y_scale, "x_offset": 5, "y_offset": 5, "object_position": 1, }, ) except: pass sheet.write_row("A{}".format(index + 2), [goods_number]) sheet.write_row("B{}".format(index + 2), [goods_art_no]) sheet.write_row("E{}".format(index + 2), [goods_pic_total])