""" 简化的图片导出服务 - 使用Pillow生成图片 支持分批并发处理，适配低配服务器（2核4G） """ import io import logging import re import asyncio from typing import List, Dict, Optional, Callable from PIL import Image, ImageDraw, ImageFont import requests from io import BytesIO from app.utils.fonts import get_font, get_fonts, find_available_font logger = logging.getLogger(__name__) class SimpleImageExportService: """简化的图片导出服务""" def __init__(self): self.image_width = 800 self.image_height = 600 self.padding = 20 self.line_height = 30 def _get_headers(self) -> Dict[str, str]: """获取图片下载请求头（淘宝图片需要特定请求头）""" return { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'image/webp,image/apng,image/*,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Referer': 'https://item.taobao.com/', 'Connection': 'keep-alive' } async def _download_image_async(self, url: str, timeout: int = 15) -> Optional[bytes]: """ 异步下载单张图片 Args: url: 图片URL timeout: 超时时间（秒），默认15秒 Returns: 图片字节数据，失败返回 None """ if not url or not url.startswith(('http://', 'https://')): return None try: import aiohttp async with aiohttp.ClientSession() as session: async with session.get( url, headers=self._get_headers(), timeout=aiohttp.ClientTimeout(total=timeout) ) as resp: if resp.status == 200: return await resp.read() else: logger.warning(f"图片下载HTTP {resp.status}: {url}") except asyncio.TimeoutError: logger.warning(f"图片下载超时 ({timeout}s): {url}") except Exception as e: logger.warning(f"图片下载失败: {url}, 错误: {e}") return None async def generate_sku_images_batched( self, selected_skus: List[Dict], max_concurrent: int = 3, batch_size: int = 10, progress_callback: Optional[Callable[[int, int, str], None]] = None ) -> List[Dict]: """ 分批并发生成 SKU 图片（适配低配服务器 2核4G） Args: selected_skus: SKU 列表 max_concurrent: 最大并发数（默认3，适配2核CPU） batch_size: 每批处理数量（默认10，控制内存使用） progress_callback: 进度回调函数 (current, total, message) Returns: 图片数据列表 [{'filename': str, 'data': bytes}, ...] """ import os results = [] total = len(selected_skus) semaphore = asyncio.Semaphore(max_concurrent) async def process_single_sku(sku: Dict, idx: int) -> Optional[Dict]: """处理单个 SKU""" async with semaphore: try: # 1. 异步下载图片 image_url = sku.get('image_url', '') image_bytes = None if image_url and image_url.startswith(('http://', 'https://')): image_bytes = await self._download_image_async(image_url) # 2. PIL 图片处理放入线程池（CPU密集型操作） combined_image_bytes = await asyncio.to_thread( self._create_combined_image_from_bytes, sku, idx, image_bytes ) if combined_image_bytes: # 生成导出文件名 product_code = sku.get('product_code', f'SKU_{idx+1}') color = str(sku.get('color', '')).replace('/', '_').replace(' ', '_') if 'size_details' in sku and len(sku.get('size_details', [])) > 0: total_qty = sum(detail.get('quantity', 0) for detail in sku['size_details']) export_filename = f"{product_code}_{color}_({total_qty}).jpg" else: size = str(sku.get('size', '')).replace('/', '_').replace(' ', '_') quantity = sku.get('quantity', 0) if size: export_filename = f"{product_code}_{color}_{size}_({quantity}).jpg" else: export_filename = f"{product_code}_{color}_({quantity}).jpg" return { 'filename': export_filename, 'data': combined_image_bytes } except Exception as e: logger.error(f"处理 SKU {idx+1} 失败: {str(e)}") return None # 分批处理 for batch_start in range(0, total, batch_size): batch_end = min(batch_start + batch_size, total) batch = selected_skus[batch_start:batch_end] if progress_callback: progress_callback(batch_start, total, f"正在处理第 {batch_start+1}-{batch_end} 张...") # 并发处理当前批次 tasks = [ process_single_sku(sku, batch_start + i) for i, sku in enumerate(batch) ] batch_results = await asyncio.gather(*tasks, return_exceptions=True) # 收集成功的结果 for r in batch_results: if isinstance(r, dict) and r is not None: results.append(r) elif isinstance(r, Exception): logger.error(f"批次处理异常: {r}") # 批次间休息，让系统喘息（除最后一批外） if batch_end < total: await asyncio.sleep(0.3) if progress_callback: progress_callback(total, total, f"处理完成，共 {len(results)} 张图片") logger.info(f"分批并发处理完成: {len(results)}/{total} 张图片") return results def _create_combined_image_from_bytes( self, sku: Dict, index: int, image_bytes: Optional[bytes] = None ) -> bytes: """ 从已下载的图片字节数据创建组合图片 Args: sku: SKU 数据 index: 索引 image_bytes: 已下载的图片字节数据（可选） Returns: 组合图片的字节数据 """ import os try: # 设置图片尺寸 product_image_height = 800 info_height = 180 total_height = 980 total_width = 800 # 创建高分辨率白色背景的组合图片 scale_factor = 2 high_res_width = total_width * scale_factor high_res_height = total_height * scale_factor combined_img = Image.new('RGB', (high_res_width, high_res_height), 'white') # 加载产品图片 product_img = None # 优先使用已下载的字节数据 if image_bytes: try: product_img = Image.open(BytesIO(image_bytes)) if product_img.mode != 'RGB': product_img = product_img.convert('RGB') logger.debug(f"从预下载数据加载图片成功") except Exception as e: logger.warning(f"从字节数据加载图片失败: {e}") product_img = None # 如果没有预下载数据，尝试本地路径 if product_img is None: image_url = sku.get('image_url', '') if image_url: product_img = self._try_load_local_image(image_url) # 如果有产品图片，将其添加到组合图片中 if product_img: img_ratio = product_img.width / product_img.height if img_ratio > 1: new_width = high_res_width new_height = int(high_res_width / img_ratio) else: new_height = product_image_height * scale_factor new_width = int(product_image_height * scale_factor * img_ratio) product_img = product_img.resize((new_width, new_height), Image.Resampling.LANCZOS) paste_x = (high_res_width - new_width) // 2 paste_y = (product_image_height * scale_factor - new_height) // 2 combined_img.paste(product_img, (paste_x, paste_y)) else: draw = ImageDraw.Draw(combined_img) font = get_font(72) draw.text((high_res_width//2 - 120, product_image_height * scale_factor//2 - 40), "暂无图片", fill='gray', font=font) # 在下方绘制商品信息 draw = ImageDraw.Draw(combined_img) large_font, small_font = get_fonts(84, 42) draw.line([(0, product_image_height * scale_factor), (high_res_width, product_image_height * scale_factor)], fill='#e0e0e0', width=2) margin = 40 info_y = product_image_height * scale_factor + 30 max_width = high_res_width - 2 * margin # 第一行：品牌 货号 颜色 尺寸 first_line_parts = [] brand = sku.get('brand', '') if brand: first_line_parts.append(brand) product_code = sku.get('product_code', '') if product_code: first_line_parts.append(product_code) color = sku.get('color', '') if color: first_line_parts.append(color) size_details = [] if 'size_details' in sku and sku['size_details']: for detail in sku['size_details']: size_val = detail.get('size', '') qty = detail.get('quantity', 0) if size_val: size_val = self._clean_measurement_brackets(size_val) if size_val: size_details.append(f"{size_val}({qty})") else: size = sku.get('size', '') quantity = sku.get('quantity', sku.get('pending_procurement_qty', 0)) if size: size = self._clean_measurement_brackets(size) if size: size_details.append(f"{size}({quantity})") if size_details: first_line_parts.append(' '.join(size_details)) first_line_text = ' '.join(first_line_parts) wrapped_first_line = self._wrap_text(first_line_text, large_font, max_width) for line in wrapped_first_line: draw.text((margin, info_y), line, fill='red', font=large_font) info_y += 90 # 缩小图片到目标尺寸 final_img = combined_img.resize((total_width, total_height), Image.Resampling.LANCZOS) buffer = BytesIO() final_img.save(buffer, format='JPEG', quality=95, optimize=True, dpi=(300, 300)) return buffer.getvalue() except Exception as e: logger.error(f"创建组合图片失败: {str(e)}") return self._create_sku_image(sku, index) def _try_load_local_image(self, image_url: str) -> Optional[Image.Image]: """尝试从本地路径加载图片""" import os try: if image_url.startswith('/api/v1/static/images/'): image_filename = image_url.split('/api/v1/static/images/')[-1] current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, 'images', image_filename) if os.path.exists(image_path): img = Image.open(image_path) if img.mode != 'RGB': img = img.convert('RGB') return img elif image_url.startswith('images/'): current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, image_url) if os.path.exists(image_path): img = Image.open(image_path) if img.mode != 'RGB': img = img.convert('RGB') return img elif not image_url.startswith(('http://', 'https://')): image_filename = None if 'images/' in image_url: image_filename = image_url.split('images/')[-1] elif image_url: image_filename = os.path.basename(image_url) if image_filename: current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, 'images', image_filename) if os.path.exists(image_path): img = Image.open(image_path) if img.mode != 'RGB': img = img.convert('RGB') return img except Exception as e: logger.warning(f"加载本地图片失败 {image_url}: {e}") return None def _clean_measurement_brackets(self, text: str) -> str: """去除包含胸围/腰围的中括号内容""" # 匹配包含胸围或腰围的中括号，例如 [胸围90 腰围70] pattern = r'\[.*?(?:胸围|腰围).*?\]' cleaned = re.sub(pattern, '', text) # 清理多余的空格 cleaned = re.sub(r'\s+', ' ', cleaned).strip() return cleaned async def generate_sku_images(self, selected_skus: List[Dict]) -> List[Dict]: """导出包含商品信息的产品图片""" import os import base64 images = [] for idx, sku in enumerate(selected_skus): try: # 生成包含图片和商品信息的组合图片 combined_image_bytes = self._create_combined_image(sku, idx) if combined_image_bytes: # 生成导出文件名 product_code = sku.get('product_code', f'SKU_{idx+1}') color = str(sku.get('color', '')).replace('/', '_').replace(' ', '_') # 如果有多个尺寸，在文件名中包含总数量信息 if 'size_details' in sku and len(sku.get('size_details', [])) > 0: total_qty = sum(detail.get('quantity', 0) for detail in sku['size_details']) export_filename = f"{product_code}_{color}_({total_qty}).jpg" else: # 单个尺寸，包含尺寸信息 size = str(sku.get('size', '')).replace('/', '_').replace(' ', '_') quantity = sku.get('quantity', 0) if size: export_filename = f"{product_code}_{color}_{size}_({quantity}).jpg" else: export_filename = f"{product_code}_{color}_({quantity}).jpg" images.append({ 'filename': export_filename, 'data': combined_image_bytes }) logger.info(f"Generated combined image {idx+1}/{len(selected_skus)}: {export_filename}") except Exception as e: logger.error(f"Failed to generate image for SKU {idx+1}: {str(e)}") continue logger.info(f"Successfully generated {len(images)} images") return images def _create_combined_image(self, sku: Dict, index: int) -> bytes: """创建包含原始图片和商品信息的组合图片""" import os import textwrap try: # 获取图片URL image_url = sku.get('image_url', '') # 设置图片尺寸 product_image_height = 800 # 产品图片高度 800x800 info_height = 180 # 信息区域高度（增加到180px，支持3行文字完整显示） total_height = 980 # 总高度（800产品图+180信息区） total_width = 800 # 创建高分辨率白色背景的组合图片（2倍分辨率，最后缩放） scale_factor = 2 # 2倍分辨率以提高文字清晰度 high_res_width = total_width * scale_factor high_res_height = total_height * scale_factor combined_img = Image.new('RGB', (high_res_width, high_res_height), 'white') # 加载产品图片 product_img = None if image_url: logger.info(f"Loading image from URL: {image_url}") try: if image_url.startswith('http://') or image_url.startswith('https://'): # 从HTTP URL下载图片 import requests # 对于淘宝图片，添加请求头避免403错误 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'image/webp,image/apng,image/*,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Referer': 'https://item.taobao.com/', 'Connection': 'keep-alive' } # 多次重试机制 max_retries = 3 retry_count = 0 while retry_count < max_retries: try: response = requests.get(image_url, headers=headers, timeout=60) if response.status_code == 200: product_img = Image.open(BytesIO(response.content)) logger.info(f"Successfully loaded image from {image_url}") break else: logger.warning(f"HTTP {response.status_code} for {image_url}, retry {retry_count+1}/{max_retries}") except Exception as retry_error: logger.warning(f"Retry {retry_count+1} failed: {retry_error}") retry_count += 1 if retry_count < max_retries: import time time.sleep(1) # 等待1秒后重试 elif image_url.startswith('/api/v1/static/images/'): # 本地静态文件路径 - 使用相对路径，适配本地和Docker环境 image_filename = image_url.split('/api/v1/static/images/')[-1] # 获取当前文件所在目录，计算 backend/images 路径 current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, 'images', image_filename) if os.path.exists(image_path): product_img = Image.open(image_path) logger.info(f"Loaded local image from {image_path}") elif image_url.startswith('images/'): # 相对路径格式（如 images/xxx.jpg） current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, image_url) if os.path.exists(image_path): product_img = Image.open(image_path) logger.info(f"Loaded relative path image from {image_path}") else: # 尝试从URL提取图片文件名 image_filename = None if 'images/' in image_url: image_filename = image_url.split('images/')[-1] elif image_url: image_filename = os.path.basename(image_url) if image_filename: # 构建实际的图片路径 - 使用相对路径 current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) image_path = os.path.join(current_dir, 'images', image_filename) # 检查文件是否存在并加载 if os.path.exists(image_path): product_img = Image.open(image_path) logger.info(f"Loaded fallback path image from {image_path}") # 转换为RGB模式（避免RGBA等模式问题） if product_img and product_img.mode != 'RGB': product_img = product_img.convert('RGB') except Exception as img_error: logger.warning(f"Failed to load image from {image_url}: {img_error}") product_img = None # 如果有产品图片，将其添加到组合图片中 if product_img: # 等比例缩放产品图片以适应800x800区域（高分辨率） img_ratio = product_img.width / product_img.height if img_ratio > 1: # 图片太宽，按宽度缩放 new_width = high_res_width new_height = int(high_res_width / img_ratio) else: # 图片太高，按高度缩放 new_height = product_image_height * scale_factor new_width = int(product_image_height * scale_factor * img_ratio) # 缩放图片到高分辨率 product_img = product_img.resize((new_width, new_height), Image.Resampling.LANCZOS) # 居中粘贴产品图片 paste_x = (high_res_width - new_width) // 2 paste_y = (product_image_height * scale_factor - new_height) // 2 combined_img.paste(product_img, (paste_x, paste_y)) else: # 没有图片时显示占位文字 draw = ImageDraw.Draw(combined_img) font = get_font(72) # 2倍大小，跨平台支持 draw.text((high_res_width//2 - 120, product_image_height * scale_factor//2 - 40), "暂无图片", fill='gray', font=font) # 在下方绘制商品信息（白底） draw = ImageDraw.Draw(combined_img) # 加载字体 - 使用跨平台字体工具 large_font, small_font = get_fonts(84, 42) # 高分辨率字体大小（2倍） # 绘制分割线（高分辨率） draw.line([(0, product_image_height * scale_factor), (high_res_width, product_image_height * scale_factor)], fill='#e0e0e0', width=2) # 商品信息起始位置（高分辨率） margin = 40 # 2倍 info_y = product_image_height * scale_factor + 30 # 2倍 max_width = high_res_width - 2 * margin # 第一行：品牌 货号 颜色 尺寸（红色字体，去掉标签） first_line_parts = [] # 品牌 - 不清理 brand = sku.get('brand', '') if brand: first_line_parts.append(brand) # 货号 - 不清理 product_code = sku.get('product_code', '') if product_code: first_line_parts.append(product_code) # 颜色 - 不清理 color = sku.get('color', '') if color: first_line_parts.append(color) # 尺寸和数量（处理多尺寸情况） size_details = [] if 'size_details' in sku and sku['size_details']: # 有多个尺寸 - 汇总在一起显示 for detail in sku['size_details']: size_val = detail.get('size', '') qty = detail.get('quantity', 0) if size_val: # 清理尺寸中的测量数据 size_val = self._clean_measurement_brackets(size_val) if size_val: # 只有清理后还有内容才添加 size_details.append(f"{size_val}({qty})") else: # 单个尺寸 size = sku.get('size', '') quantity = sku.get('quantity', sku.get('pending_procurement_qty', 0)) if size: # 清理尺寸中的测量数据 size = self._clean_measurement_brackets(size) if size: # 只有清理后还有内容才添加 size_details.append(f"{size}({quantity})") if size_details: first_line_parts.append(' '.join(size_details)) # 组合第一行文本 first_line_text = ' '.join(first_line_parts) # 自动换行处理（第一行） wrapped_first_line = self._wrap_text(first_line_text, large_font, max_width) for line in wrapped_first_line: draw.text((margin, info_y), line, fill='red', font=large_font) info_y += 90 # 行间距（2倍） # 第二行内容已移除 - 不再显示产品名称和销售属性 # 缩小图片到目标尺寸（高质量缩放） final_img = combined_img.resize((total_width, total_height), Image.Resampling.LANCZOS) # 保存为高质量JPEG buffer = BytesIO() # 提高质量到95，减少压缩损失 final_img.save(buffer, format='JPEG', quality=95, optimize=True, dpi=(300, 300)) return buffer.getvalue() except Exception as e: logger.error(f"Failed to create combined image: {str(e)}") # 如果创建组合图片失败，返回原始的简单图片 return self._create_sku_image(sku, index) def _wrap_text(self, text, font, max_width): """文本自动换行""" lines = [] if not text: return lines # 简单的换行处理（基于字符数估算） # 估算每个字符的平均宽度 try: # 获取单个中文字符的宽度 char_width = font.getlength('中') max_chars = int(max_width / char_width) except: max_chars = 30 # 默认值 words = text current_line = "" for i, char in enumerate(words): current_line += char try: if font.getlength(current_line) > max_width: # 当前行太长，需要换行 lines.append(current_line[:-1]) # 不包括最后一个字符 current_line = char # 新行从当前字符开始 except: # 如果getlength不可用，使用字符数估算 if len(current_line) > max_chars: lines.append(current_line[:-1]) current_line = char if current_line: lines.append(current_line) return lines if lines else [text] def _create_sku_image(self, sku: Dict, index: int) -> bytes: """创建单个SKU的图片""" # 创建白色背景图片 img = Image.new('RGB', (self.image_width, self.image_height), 'white') draw = ImageDraw.Draw(img) # 使用跨平台字体工具 title_font = get_font(24) normal_font = get_font(18) small_font = get_font(14) y_position = self.padding # 标题 title = f"采购单 #{index + 1}" draw.text((self.padding, y_position), title, fill='black', font=title_font) y_position += 40 # 绘制分割线 draw.line([(self.padding, y_position), (self.image_width - self.padding, y_position)], fill='gray', width=1) y_position += 20 # SKU信息 info_items = [ ('产品名称', sku.get('product_name', 'N/A')), ('品牌', sku.get('brand', 'N/A')), ('货号', sku.get('product_code', 'N/A')), ('颜色', sku.get('color', 'N/A')), ('尺寸', sku.get('size', 'N/A')), ('数量', str(sku.get('pending_procurement_qty', 0))), ('平均价格', f"¥{sku.get('avg_price', 0):.2f}"), ] for label, value in info_items: # 绘制标签 draw.text((self.padding, y_position), f"{label}:", fill='gray', font=normal_font) # 绘制值 draw.text((self.padding + 120, y_position), str(value), fill='black', font=normal_font) y_position += self.line_height # 添加底部信息 y_position = self.image_height - 60 draw.line([(self.padding, y_position), (self.image_width - self.padding, y_position)], fill='gray', width=1) y_position += 10 # 店铺信息 shops = sku.get('shops', []) if shops: shops_text = f"关联店铺: {', '.join(shops[:3])}" if len(shops) > 3: shops_text += f" 等{len(shops)}个" draw.text((self.padding, y_position), shops_text, fill='gray', font=small_font) # 如果有图片URL，尝试加载第一张图片 image_urls = sku.get('image_urls', []) if image_urls and len(image_urls) > 0: try: # 在右侧绘制产品图片区域 img_x = self.image_width - 250 - self.padding img_y = 100 img_width = 250 img_height = 250 # 绘制图片占位框 draw.rectangle( [(img_x, img_y), (img_x + img_width, img_y + img_height)], outline='gray', width=2 ) # 如果是本地图片路径，尝试加载 first_url = image_urls[0] if first_url and not first_url.startswith('http'): # 本地图片，添加占位文字 draw.text( (img_x + img_width//2 - 30, img_y + img_height//2 - 10), "产品图片", fill='gray', font=normal_font ) except Exception as e: logger.warning(f"Failed to process image: {str(e)}") # 保存为JPEG buffer = BytesIO() img.save(buffer, format='JPEG', quality=85) return buffer.getvalue()