import json import os import re import pandas as pd import requests from natsort.natsort import natsorted from PIL import Image, ImageDraw, ImageFont import settings from logger import logger PLATE_LONG = 2048 PLATE_SIZE = (1366, 2048) PLATE_COLOR = (186, 186, 186) OUTPUT_NAME = "场景套图" MARKER_NAME = "场景套图_任务.json" TEMP_NAME = "_场景套图临时" LAYOUT_PLATE = { "handheld": {"height_ratio": 0.30, "bottom_ratio": 0.10, "max_width_ratio": 0.66}, "floor": {"height_ratio": 0.44, "bottom_ratio": 0.02, "max_width_ratio": 0.90}, } ANCHOR_LABELS = { "top-left": "左上", "top-right": "右上", "bottom-left": "左下", "bottom-right": "右下", "center": "中间", } SKIP_FOLDERS = {"原始图", "原始图_已抠图", "800x800", "阴影图处理", "200images", OUTPUT_NAME, TEMP_NAME} def prepare_scene_set(payload): goods_dir = resolve_goods_dir(payload) cutout_dir = os.path.join(goods_dir, "原始图_已抠图") if not os.path.isdir(cutout_dir): raise RuntimeError("没有找到原始图_已抠图，不会重新抠图") files = natsorted([ name for name in os.listdir(cutout_dir) if name.lower().endswith(".png") and os.path.isfile(os.path.join(cutout_dir, name)) ]) if not files: raise RuntimeError("原始图_已抠图里没有 PNG，不会重新抠图") layout = payload.get("layout") or "handheld" slots = [slot for slot in (payload.get("slots") or []) if (slot.get("layout") or layout) == layout] slots = sorted(slots, key=lambda slot: int(slot.get("seq") or 0)) cutouts = [] for index, name in enumerate(files): slot = slots[index] if index < len(slots) else {} cutouts.append({ "file": name, "path": os.path.join(cutout_dir, name), "index": index + 1, "slot_id": slot.get("id") or 0, "name": slot.get("name") or "未识别", "role": slot.get("role") or "", "rotation": int(slot.get("rotation") or 0), "use_long_scene": int(slot.get("use_long_scene") or 0), "use_three_view": int(slot.get("use_three_view") or 0), "use_detail": int(slot.get("use_detail") or 0), "use_for_scene": int(slot.get("use_for_scene") or 0) if slot else 0, "crops": slot.get("crops") or [], }) output_dir = os.path.join(goods_dir, OUTPUT_NAME) logger.info("场景套图使用已有抠图 goods=%s dir=%s count=%s", payload.get("goods_art_no"), goods_dir, len(cutouts)) return { "goods_dir": goods_dir, "cutout_dir": cutout_dir, "output_dir": output_dir, "output_exists": os.path.isdir(output_dir), "cutouts": cutouts, "message": "使用已有抠图，没有重新抠图", } def run_scene_set(payload): goods_dir = resolve_goods_dir(payload) output_dir = os.path.join(goods_dir, OUTPUT_NAME) if os.path.isdir(output_dir): return { "skipped": True, "reason": "已有场景套图目录，要重出请先删除该目录", "output_dir": output_dir, "goods_dir": goods_dir, } marker = os.path.join(goods_dir, MARKER_NAME) if os.path.isfile(marker) and not int(payload.get("force") or 0): saved = json.loads(open(marker, "r", encoding="utf-8").read()) if saved.get("job_id"): return { "resumed": True, "job_id": saved["job_id"], "goods_dir": goods_dir, "output_dir": output_dir, "reason": "继续上次未完成的任务", } cutouts = payload.get("cutouts") or [] source = next((item for item in cutouts if int(item.get("use_for_scene") or 0) and int(item.get("use_long_scene") or 0)), None) if not source: raise RuntimeError("没有可用于场景的机位") layout = payload.get("layout") or "handheld" temp_dir = os.path.join(goods_dir, TEMP_NAME) os.makedirs(temp_dir, exist_ok=True) plate_path = os.path.join(temp_dir, "plate.jpg") png_path = os.path.join(temp_dir, "cutout.png") build_plate(source["path"], int(source.get("rotation") or 0), layout, png_path, plate_path) logger.info("场景套图上传抠图 PNG，不转成 JPEG %s", png_path) upload_file(png_path) plate_url = upload_file(plate_path) materials = collect_materials(payload) body = { "category_id": int(payload.get("category_id") or 0), "goods_art_no": payload.get("goods_art_no") or "", "basis": payload.get("basis") or "selling_point", "scene_tags": payload.get("scene_tags") or [], "materials": materials, "cutouts": [public_cutout(item) for item in cutouts], "copy_anchor": payload.get("copy_anchor") or "top-left", "plate_url": plate_url, "options": { "layout": layout, "height_cm": payload.get("height_cm") or "", "missing_height": payload.get("missing_height") or "skip", "default_tier": payload.get("default_tier") or "M", "visible_features": payload.get("visible_features") or [], }, } data = backend_post("/product_scene/create_job", body) if data.get("job_id"): with open(marker, "w", encoding="utf-8") as file: json.dump({"job_id": data["job_id"]}, file, ensure_ascii=False) data["goods_dir"] = goods_dir data["output_dir"] = output_dir data["message"] = "使用已有抠图，没有重新抠图" return data def save_scene_set(payload): goods_dir = payload.get("goods_dir") or "" if not goods_dir or not os.path.isdir(goods_dir): raise RuntimeError("货号目录不存在") job = payload.get("job") or {} cutouts = payload.get("cutouts") or [] anchor = job.get("copy_anchor") or payload.get("copy_anchor") or "top-left" output_dir = os.path.join(goods_dir, OUTPUT_NAME) os.makedirs(output_dir, exist_ok=True) saved = [] notes = list(job.get("progress_notes") or []) for image in job.get("images") or []: kind = image.get("kind") if kind == "scene": if int(image.get("status") or 0) != 2 or not image.get("result_url"): notes.append((image.get("scene_name") or "场景") + " 未出图：" + (image.get("error_msg") or "失败")) continue stem = "场景_{:02d}_{}".format(int(image.get("sort") or 1), safe_name(image.get("scene_name") or "场景")) jpg_path = os.path.join(output_dir, stem + ".jpg") download(image["result_url"], jpg_path) png_path = os.path.join(output_dir, stem + ".png") compose_copy(Image.open(jpg_path).convert("RGB"), anchor, image.get("title") or "", image.get("lines") or []).save(png_path, "PNG") saved.append(jpg_path) saved.append(png_path) elif kind == "three_view": path = os.path.join(output_dir, "三视图.png") render_three_view(cutouts, image, anchor).save(path, "PNG") saved.append(path) elif kind == "detail": path = os.path.join(output_dir, "细节.png") render_detail(cutouts, image, anchor).save(path, "PNG") saved.append(path) elif kind in ("size", "param"): filename = "尺寸.png" if kind == "size" else "参数.png" path = os.path.join(output_dir, filename) render_text_page(image.get("title") or filename[:-4], image.get("lines") or [], anchor).save(path, "PNG") saved.append(path) txt_path = os.path.join(output_dir, "卖点.txt") with open(txt_path, "w", encoding="utf-8") as file: file.write(selling_text(job, anchor, notes)) saved.append(txt_path) marker = os.path.join(goods_dir, MARKER_NAME) if os.path.isfile(marker): os.remove(marker) logger.info("场景套图已写入 %s", output_dir) return {"output_dir": output_dir, "files": saved, "notes": notes} def resolve_goods_dir(payload): goods_dir = payload.get("goods_dir") or "" if goods_dir and os.path.isdir(goods_dir) and has_cutout_png(goods_dir): return os.path.abspath(goods_dir) image_dir = goods_dir_from_image(payload.get("image_path") or "") if image_dir and has_cutout_png(image_dir): return image_dir found = find_output_goods_dir(payload.get("goods_art_no") or "") if found: return found if image_dir: return image_dir if goods_dir and os.path.isdir(goods_dir): return os.path.abspath(goods_dir) raise RuntimeError("没有找到货号目录") def goods_dir_from_image(image_path): if not image_path or not os.path.exists(image_path): return "" parent = os.path.dirname(os.path.abspath(image_path)) if os.path.basename(parent) in SKIP_FOLDERS: parent = os.path.dirname(parent) return parent def has_cutout_png(goods_dir): cutout_dir = os.path.join(goods_dir, "原始图_已抠图") if not os.path.isdir(cutout_dir): return False for name in os.listdir(cutout_dir): path = os.path.join(cutout_dir, name) if name.lower().endswith(".png") and os.path.isfile(path): return True return False def find_output_goods_dir(goods_art_no): goods_art_no = str(goods_art_no or "").strip() output_dir = str(settings.OUTPUT_DIR) if not goods_art_no or not os.path.isdir(output_dir): return "" matches = [] try: names = os.listdir(output_dir) except OSError: return "" for date_dir in names: date_path = os.path.join(output_dir, date_dir) goods_path = os.path.join(date_path, goods_art_no) if os.path.isdir(date_path) and has_cutout_png(goods_path): matches.append(os.path.abspath(goods_path)) if not matches: return "" matches.sort(key=lambda path: os.path.basename(os.path.dirname(path))) return matches[-1] def build_plate(source_path, rotation, layout, png_path, plate_path): image = Image.open(source_path).convert("RGBA") if rotation: image = image.rotate(-int(rotation), expand=True) image = trim_alpha(image) image = fit_long_edge(image, PLATE_LONG) image.save(png_path, "PNG") ratios = LAYOUT_PLATE.get(layout) or LAYOUT_PLATE["handheld"] canvas = Image.new("RGB", PLATE_SIZE, PLATE_COLOR) width, height = image.size scale = min( ratios["height_ratio"] * canvas.height / height, ratios["max_width_ratio"] * canvas.width / width, ) resized = image.resize((max(1, int(width * scale)), max(1, int(height * scale))), Image.Resampling.LANCZOS) x = int((canvas.width - resized.width) / 2) y = int(canvas.height - ratios["bottom_ratio"] * canvas.height - resized.height) canvas.paste(resized, (x, y), resized.getchannel("A")) canvas.save(plate_path, "JPEG", quality=90) def trim_alpha(image, threshold=16, pad=12): alpha = image.getchannel("A") mask = alpha.point(lambda value: 255 if value > threshold else 0) box = mask.getbbox() if not box: return image left, top, right, bottom = box left = max(0, left - pad) top = max(0, top - pad) right = min(image.width, right + pad) bottom = min(image.height, bottom + pad) return image.crop((left, top, right, bottom)) def fit_long_edge(image, long_edge): width, height = image.size longest = max(width, height) if longest <= long_edge: return image scale = long_edge / float(longest) return image.resize((max(1, int(width * scale)), max(1, int(height * scale))), Image.Resampling.LANCZOS) def upload_file(path): ext = os.path.splitext(path)[1].lower() content_type = "image/png" if ext == ".png" else "image/jpeg" with open(path, "rb") as file: result = requests.post( settings.DOMAIN + "/api/upload", files={"file": (os.path.basename(path), file, content_type)}, timeout=120, ).json() url = ((result or {}).get("data") or {}).get("url") if not url: raise RuntimeError("图片上传失败") return url def collect_materials(payload): fields = dict(payload.get("materials") or {}) selling = (payload.get("selling_points") or "").strip() excel_path = payload.get("excel_path") or "" if excel_path: parsed = read_excel(excel_path, payload.get("goods_art_no") or "") for key, value in parsed["fields"].items(): fields.setdefault(key, value) if not selling: selling = parsed["selling_points"] return {"fields": fields, "selling_points": selling} def read_excel(path, goods_art_no): if not os.path.isfile(path): raise RuntimeError("资料文件不存在") frame = pd.read_excel(path) frame.columns = [str(column).strip() for column in frame.columns] if {"字段", "内容"}.issubset(set(frame.columns)): fields = {} selling = [] for _, row in frame.iterrows(): name = "" if pd.isna(row["字段"]) else str(row["字段"]).strip() value = "" if pd.isna(row["内容"]) else str(row["内容"]).strip() if not name or not value: continue if name in ("卖点", "卖点文案"): selling.append(value) else: fields[name] = value return {"fields": fields, "selling_points": "\n".join(selling)} row = frame.iloc[0] if "货号" in frame.columns and goods_art_no: matched = frame[frame["货号"].astype(str).str.strip() == str(goods_art_no).strip()] if matched.empty: raise RuntimeError("资料里没有这个货号") row = matched.iloc[0] fields = {} selling = [] for column, value in row.items(): if pd.isna(value): continue text = str(value).strip() if not text or column == "货号": continue if column in ("卖点", "卖点文案"): selling.append(text) else: fields[column] = text return {"fields": fields, "selling_points": "\n".join(selling)} def public_cutout(item): return { "name": item.get("name") or "", "role": item.get("role") or "", "use_three_view": int(item.get("use_three_view") or 0), "use_detail": int(item.get("use_detail") or 0), "use_for_scene": int(item.get("use_for_scene") or 0), "use_long_scene": int(item.get("use_long_scene") or 0), "crops": item.get("crops") or [], } def backend_post(path, payload): response = requests.post( settings.DOMAIN + "/api/ai_image/camera_machine" + path, headers={ "Authorization": "Bearer {}".format(settings.USER_TOKEN or ""), "Content-Type": "application/json", }, json=payload, timeout=180, ) try: body = response.json() except Exception: raise RuntimeError("后端没有返回结果") if body.get("code") not in (0, "0"): raise RuntimeError(body.get("message") or body.get("msg") or "后端请求失败") return body.get("data") or {} def download(url, path): response = requests.get(url, timeout=180) response.raise_for_status() with open(path, "wb") as file: file.write(response.content) def compose_copy(base, anchor, title, lines): canvas = base.convert("RGBA") scale = canvas.width / 750.0 title_lines = split_title(title) body = [str(line).strip() for line in lines if str(line).strip() and not str(line).startswith("卖点")] if not title_lines and not body: return canvas.convert("RGB") title_font = load_font(int(34 * scale), bold=True) body_font = load_font(int(20 * scale), bold=False) draw = ImageDraw.Draw(canvas) max_width = int(canvas.width * 0.72) wrapped = [] for line in title_lines: wrapped.extend(wrap_text(line, title_font, max_width, draw)) body_wrapped = [] for line in body: body_wrapped.extend(wrap_text(line, body_font, max_width, draw)) block = measure_block(draw, wrapped, title_font, body_wrapped, body_font, scale) if anchor == "center": origin = (int((canvas.width - block[0]) / 2), int((canvas.height - block[1]) / 2)) plate = Image.new("RGBA", canvas.size, (0, 0, 0, 0)) plate_draw = ImageDraw.Draw(plate) pad = int(28 * scale) plate_draw.rounded_rectangle( [origin[0] - pad, origin[1] - pad, origin[0] + block[0] + pad, origin[1] + block[1] + pad], radius=int(18 * scale), fill=(0, 0, 0, 140), ) canvas = Image.alpha_composite(canvas, plate) else: canvas = Image.alpha_composite(canvas, shade_layer(canvas.size, anchor)) origin = text_origin(canvas.size, block, anchor, scale) draw = ImageDraw.Draw(canvas) y = origin[1] for line in wrapped: x = align_x(origin[0], block[0], draw.textlength(line, font=title_font), anchor) draw.text((x, y), line, font=title_font, fill=(255, 255, 255, 255)) y += int(46 * scale) if wrapped and body_wrapped: y += int(8 * scale) for line in body_wrapped: x = align_x(origin[0], block[0], draw.textlength(line, font=body_font), anchor) draw.text((x, y), line, font=body_font, fill=(255, 255, 255, 230)) y += int(32 * scale) return canvas.convert("RGB") def render_three_view(cutouts, image, anchor): canvas = Image.new("RGB", (1500, 2400), (243, 240, 235)) plan = ((image.get("extra") or {}).get("plan") or []) margin, gap = 48, 28 card_w = 1500 - margin * 2 card_h = (2400 - margin * 2 - gap * 2) // 3 y = margin for item in plan: card = product_card(find_cutout(cutouts, item.get("role")), card_w, card_h, item.get("label") or "") canvas.paste(card, (margin, y)) y += card_h + gap return compose_copy(canvas, anchor, image.get("title") or "", image.get("lines") or []) def render_detail(cutouts, image, anchor): canvas = Image.new("RGB", (1500, 2400), (243, 240, 235)) plan = ((image.get("extra") or {}).get("plan") or [])[:4] if not plan: return compose_copy(canvas, anchor, image.get("title") or "", image.get("lines") or []) margin, gap = 48, 24 card_w = 1500 - margin * 2 card_h = (2400 - margin * 2 - gap * (len(plan) - 1)) // len(plan) y = margin for item in plan: cutout = find_cutout(cutouts, item.get("role"), item.get("name")) cropped = crop_box(open_cutout(cutout), item.get("box") or [0, 0, 1, 1]) if cutout else None card = product_card_image(cropped, card_w, card_h, item.get("label") or "细节") canvas.paste(card, (margin, y)) y += card_h + gap return compose_copy(canvas, anchor, image.get("title") or "", image.get("lines") or []) def render_text_page(title, lines, anchor): canvas = Image.new("RGB", (1500, 2400), (243, 240, 235)) return compose_copy(canvas, anchor, title, lines) def product_card(cutout, width, height, label): image = open_cutout(cutout) if cutout else None return product_card_image(image, width, height, label) def product_card_image(image, width, height, label): card = Image.new("RGB", (width, height), (255, 255, 255)) draw = ImageDraw.Draw(card) font = load_font(28, bold=False) label_h = 64 if image: fitted = fit_box(image, width - 48, height - label_h - 48) x = int((width - fitted.width) / 2) y = int((height - label_h - fitted.height) / 2) card.paste(fitted, (x, max(16, y)), fitted.getchannel("A")) text = label or "" draw.text(((width - draw.textlength(text, font=font)) / 2, height - label_h), text, font=font, fill=(40, 40, 40, 255)) return card def open_cutout(cutout): if not cutout or not cutout.get("path") or not os.path.isfile(cutout["path"]): return None image = Image.open(cutout["path"]).convert("RGBA") rotation = int(cutout.get("rotation") or 0) if rotation: image = image.rotate(-rotation, expand=True) return trim_alpha(image) def crop_box(image, box): if image is None: return None left, top, right, bottom = box width, height = image.size return image.crop(( int(float(left) * width), int(float(top) * height), int(float(right) * width), int(float(bottom) * height), )) def find_cutout(cutouts, role, name=""): for item in cutouts: if role and item.get("role") == role: return item for item in cutouts: if name and item.get("name") == name: return item return None def fit_box(image, width, height): scale = min(width / image.width, height / image.height) return image.resize((max(1, int(image.width * scale)), max(1, int(image.height * scale))), Image.Resampling.LANCZOS) def shade_layer(size, anchor): width, height = size mask = Image.new("L", size, 0) draw = ImageDraw.Draw(mask) span = int(height * 0.46) if anchor in ("top-left", "top-right"): for y in range(span): draw.line([(0, y), (width, y)], fill=int(170 * (1 - y / float(span)))) elif anchor in ("bottom-left", "bottom-right"): start = height - span for y in range(start, height): draw.line([(0, y), (width, y)], fill=int(170 * ((y - start) / float(span)))) layer = Image.new("RGBA", size, (0, 0, 0, 0)) layer.putalpha(mask) return layer def split_title(title): text = str(title or "").replace("卖点：", "").replace("卖点:", "").strip() if not text: return [] return [part.strip() for part in re.split(r"\s*/\s*", text) if part.strip()] def wrap_text(text, font, max_width, draw): lines = [] current = "" for char in str(text): trial = current + char if draw.textlength(trial, font=font) <= max_width: current = trial else: if current: lines.append(current) current = char if current: lines.append(current) return lines def measure_block(draw, title_lines, title_font, body_lines, body_font, scale): width = 0 for line in title_lines: width = max(width, draw.textlength(line, font=title_font)) for line in body_lines: width = max(width, draw.textlength(line, font=body_font)) height = len(title_lines) * int(46 * scale) + len(body_lines) * int(32 * scale) if title_lines and body_lines: height += int(8 * scale) return int(width), int(height) def text_origin(size, block, anchor, scale): pad = int(42 * scale) width, height = size block_w, block_h = block if anchor == "top-left": return pad, pad if anchor == "top-right": return width - pad - block_w, pad if anchor == "bottom-left": return pad, height - pad - block_h if anchor == "bottom-right": return width - pad - block_w, height - pad - block_h return int((width - block_w) / 2), int((height - block_h) / 2) def align_x(origin_x, block_w, line_w, anchor): if anchor in ("top-right", "bottom-right"): return origin_x + block_w - line_w if anchor == "center": return origin_x + (block_w - line_w) / 2 return origin_x def load_font(size, bold=False): path = r"C:\Windows\Fonts\msyhbd.ttc" if bold else r"C:\Windows\Fonts\msyh.ttc" if not os.path.isfile(path): path = r"C:\Windows\Fonts\msyh.ttc" if not os.path.isfile(path): raise RuntimeError("没有找到微软雅黑") return ImageFont.truetype(path, max(12, int(size)), index=0) def selling_text(job, anchor, notes): rows = ["来源：{}".format(job.get("selling_source") or ""), "文案位置：{}".format(ANCHOR_LABELS.get(anchor, anchor)), "", "条目"] points = job.get("selling_points") or [] if not points: rows.append("（无）") for index, point in enumerate(points, start=1): tags = "、".join(point.get("tags") or []) rows.append("{}. {}".format(index, point.get("text") or "")) rows.append("依据：{}".format(point.get("evidence") or "")) if tags: rows.append("标签：{}".format(tags)) rows.append("") rows.append("图片") for image in job.get("images") or []: status = {0: "等待", 1: "生成中", 2: "成功", 4: "本地", -1: "失败"}.get(int(image.get("status") or 0), str(image.get("status"))) rows.append("{} {} {}".format(image.get("kind") or "", image.get("scene_name") or "", status)) if image.get("lines"): rows.append("文案：" + " / ".join(image.get("lines") or [])) if image.get("error_msg"): rows.append("说明：" + image.get("error_msg")) if notes: rows.append("") rows.append("进度") rows.extend(notes) return "\n".join(rows) + "\n" def safe_name(name): return re.sub(r'[\\/:*?"<>|]', "_", str(name))[:40]