"""Pinterest 简报池 + 并发生成流水线。 分析节点产出简报后立即推入简报池，后台 worker 逐条并发生成： 生成设计(compose) → 三合一(product) → OSS上传 → 生成种草图(seed_shot) 不等全部分析完，边分析边生成，显著缩短总耗时。 集成： pinterest_init 创建 PinterestPipeline（存 state["pinterest_pipeline"]） pinterest_analyze 每批产出简报 → pipe.add_briefs(new_briefs) pinterest_finalize → pipe.finish()（排空 + 合并产品 + 补写报告）→ template_export 并发上限与 product_node 一致（默认 5），避免压垮图像网关。 """ import concurrent.futures import random import re import threading import time from pathlib import Path from typing import Any, Dict, List, Optional from graph.paths import project_root, runtime_root class PinterestPipeline: def __init__(self, state: Dict[str, Any]): self.config = state["config"] self.country = state.get("country", "") self.output_dir = Path(state["output_dir"]) self.cache_dir = Path(state.get("cache_dir") or self.output_dir) self.task_timestamp = str(state.get("task_timestamp") or time.strftime("%Y%m%d%H%M%S")) # 简报池 + 信号 self._lock = threading.Lock() self._cond = threading.Condition(self._lock) self._briefs: List[Dict[str, Any]] = [] self._done = False self._cursor = 0 # 正在处理（已提交线程池、尚未完成）的简报数，供路由判断简报池是否空闲 self._in_flight = 0 # 结果：每完成一个产品立即落盘追加写入 products_pending.jsonl， # 内存列表仅作缓存（finish 时再读盘合并），中途崩溃也不丢已完成产品。 self._products: List[Dict[str, Any]] = [] self._products_lock = threading.Lock() self._errors: List[Dict[str, Any]] = [] self._errors_lock = threading.Lock() # 致命图像服务错误（53/账户不可用）：置位后终止分发、丢弃未完成简报，仅保留已完成产品 self._fatal_lock = threading.Lock() self._fatal_503 = False # 已完成产品落盘文件（JSONL 追加写）：output///products_pending.jsonl self._pending_file = self.output_dir / "products_pending.jsonl" # 路径解析（复用 product_node 的 _abs 逻辑：运行根优先，其次数据根） pcfg = self.config.get("product") or {} def _abs(key: str, default: str) -> Path: p = Path(pcfg.get(key, default)) if p.is_absolute(): return p for root in (runtime_root(), project_root()): cand = root / p if cand.exists(): return cand return project_root() / p self._db_path = _abs("db_path", "db/spu_sku.db") self._basemap_root = _abs("basemap_dir", "basemap") self._material_root = _abs("material_library_dir", "material_library") self._category = pcfg.get("model_category", "T-shirt") self._prefix = str(pcfg.get("code_prefix") or "DG").strip() # 任务清单（spu_tasks → [(spu, skus)]，简报按序号绑定） self._worklist = self._build_worklist() # 图像后端（compose + product 共用）/ 标题后端 self._ib = self._init_image_backend() self._title_backend = self._init_title_backend() # 分析后端（失败/侵权时从图池补充图片重新分析用） self._analyze_backend = self._init_analyze_backend() self._country_config = state.get("country_config") or {} # 补充重试次数：设计生成失败/侵权时，从图池取新图重新分析的最多尝试次数 self._supply_attempts = int((self.config.get("pinterest") or {}).get("supply_attempts", 3)) # 400 计数（per 种子词）：多模态 + 生图模型合计，超限放弃当前种子词 self._err400_lock = threading.Lock() self._err400_count = 0 self._err400_limit = int((self.config.get("pinterest") or {}).get("err400_limit", 15)) self._err400_aborted = False self._err400_term = "" # 模特分配（一个 SPU 一个模特，SPU>模特数循环兜底） self._model_assign = self._assign_models() # 材质映射（seed_shot 用） self._material_map = self._load_materials() # 类目 → 性别（seed_shot 用）：模版「类目」表头值含「男」→ 男模；含「女」→ 女模；都不含 → 全部随机 self._gender = None tp = str((pcfg.get("template_path") or "") or "").strip() if tp: from graph.seed_shot import read_template_category, gender_from_category category = read_template_category(tp) self._gender = gender_from_category(category) if self._gender: label = {"male": "男", "female": "女", "boy_kids": "男童", "girl_kids": "女童"}.get(self._gender, self._gender) print(f"[pinterest_pipeline] 类目「{category[:30]}…」检测到 {label} → 固定 {self._gender} 模特") elif category: print(f"[pinterest_pipeline] 类目「{category[:30]}…」无男/女 → 男女模特随机") # OSS self._oss_cfg = self.config.get("oss") or {} self._oss_enabled = bool(self._oss_cfg.get("enabled", True)) and bool( self._oss_cfg.get("oss_bucket")) self.oss_seq = int(state.get("oss_seq") or 0) self._oss_lock = threading.Lock() # 并发线程池（默认 5，与 product_node 上限一致） concurrency = int(pcfg.get("concurrency") or 0) or 5 self._pool = concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) # 分发线程：拉简报 → 逐条提交线程池 self._dispatcher = threading.Thread(target=self._dispatch, daemon=True) self._dispatcher.start() print(f"[pinterest_pipeline] 简报池启动：{len(self._worklist)} 个产品任务，" f"并发 {concurrency}（{self.country}）") # ------------------------------------------------------------------ # # 初始化辅助 # ------------------------------------------------------------------ # def _build_worklist(self) -> List[tuple]: pcfg = self.config.get("product") or {} spu_tasks = pcfg.get("spu_tasks") or [] worklist: List[tuple] = [] if not spu_tasks: return worklist try: from graph.product import list_spus spus = list_spus(str(self._db_path)) except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] SPU 读取失败: {e}") spus = [] for t in spu_tasks: code = (t.get("spu") or t.get("spu_code") or "").strip() spu = next((s for s in spus if s["code"] == code), None) if spu is None: print(f"[pinterest_pipeline] 任务款号 {code} 不在 db，跳过") continue worklist.append((spu, (t.get("skus") or "").strip())) return worklist def _init_image_backend(self): compose_cfg = self.config.get("compose") or {} backend_name = (compose_cfg.get("backend") or "").strip() if not backend_name: return None try: from graph.backends import get_image_backend ib = get_image_backend(backend_name) if ib is not None: ib.bind_config(compose_cfg) return ib except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 图像后端不可用: {e}") return None def _init_title_backend(self): ls_cfg = self.config.get("llm_screen") or {} if (ls_cfg.get("provider") or "") in ("", "mock"): return None try: from graph.llms import get_backend as _glb tb = _glb(ls_cfg.get("provider")) tb.bind_config(ls_cfg) if getattr(tb, "has_key", False): return tb except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 标题后端初始化失败: {e}") return None def _init_analyze_backend(self): """图片分析后端（失败/侵权时从图池补充图片重新分析用）。""" pcfg = self.config.get("pinterest") or {} provider = str(pcfg.get("provider") or "openai").strip().lower() if provider == "static": return None try: from graph.llms import get_backend llm = get_backend(provider) if hasattr(llm, "bind_config"): llm.bind_config(self.config.get("llm_screen") or {}) if provider not in ("mock",) and not getattr(llm, "has_key", False): print(f"[pinterest_pipeline] {provider} 未配置 API key，降级 mock") llm = get_backend("mock") return llm except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 分析后端初始化失败: {e}") return None def _assign_models(self) -> Dict[str, Any]: """任务级图源分配：按每个任务 spu.mark 从可配置的「模特图/平铺图」文件夹间随机抽图。 每个任务独立随机抽一张：先合并该 mark 对应「有图的」模特/平铺目录的全部合格图，再从其中随机抽一张， 抽到哪个文件夹的图就返回对应 kind（model/flat），供 product_node 用对应提示词合成。 某类目录无图则只用另一类；两类都无图则该任务无图源（跳过合成）。 返回 {task_key: {"img": Path, "kind": "model"|"flat", "prompts": {…}}}。 """ import random as _random assign: Dict[str, Any] = {} pcfg = self.config.get("product") or {} mark_dirs = pcfg.get("mark_dirs") or {} try: from graph.product import build_mark_sources except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 图源映射导入失败: {e}") return assign # 每个出现过的 mark 各建一个图源池，避免多 mark 错配 pool_by_mark: Dict[str, list] = {} for _i, (spu, _skus) in enumerate(self._worklist): mark = str(spu.get("mark") or "").strip() or "1" if mark in pool_by_mark: continue try: sources = build_mark_sources(self._material_root, mark_dirs, self._category, mark=mark) except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] mark={mark} 图源构建失败: {e}") sources = {"model": [], "flat": []} pool = [] for kind in ("model", "flat"): for p in sources.get(kind) or []: pool.append((p, kind)) pool_by_mark[mark] = pool if pool: print(f"[pinterest_pipeline] mark={mark} 图源池：{len(pool)} 张（模特/平铺）") for _i, (spu, _skus) in enumerate(self._worklist): key = f"task_{_i}" mark = str(spu.get("mark") or "").strip() or "1" pool = pool_by_mark.get(mark) or [] if not pool: continue img, kind = _random.choice(pool) # 每任务独立随机抽一张（含 kind） assign[key] = {"img": img, "kind": kind} return assign def _load_materials(self) -> Dict[str, str]: material_map: Dict[str, str] = {} try: from graph.product import list_spus for s in list_spus(str(self._db_path)): m = " ".join(str(s.get("material", "")).replace("\r", " ").replace("\n", " ").split()) material_map[s["code"]] = m except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 材质读取失败（用空）: {e}") return material_map # ------------------------------------------------------------------ # # 对外接口 # ------------------------------------------------------------------ # def reset_400(self, term: str = "") -> None: """新一批图爬取完成后调用：重置 400 计数并记录当前种子词。""" with self._err400_lock: self._err400_count = 0 self._err400_aborted = False self._err400_term = term or "" def record_400(self) -> bool: """记录一次 400（含内容/图片）。返回 True 表示本次触发放弃当前种子词。""" with self._err400_lock: self._err400_count += 1 if self._err400_count > self._err400_limit and not self._err400_aborted: self._err400_aborted = True return True return False def is_400_aborted(self) -> bool: with self._err400_lock: return self._err400_aborted # ------------------------------------------------------------------ # # 致命图像服务错误（503 / 账户不可用）：重试无效，提前终止整个任务 # ------------------------------------------------------------------ # @staticmethod def is_fatal_503(exc) -> bool: """判断异常是否为「图像服务不可用」类致命错误（503 / No available compatible accounts）。 这类错误说明账户配额耗尽或网关故障，重试必然失败，应提前终止任务而非无意义重试。 """ msg = str(exc) if "503" in msg: return True low = msg.lower() return "no available compatible accounts" in low or "account" in low and "not available" in low def record_503(self) -> bool: """记录一次致命 503：首次触发即置位终止标志（后续请求直接短路不再提交）。 返回 True 表示本次触发终止（调用方应立即停止当前链路）。 """ with self._fatal_lock: first = not self._fatal_503 self._fatal_503 = True if first: print("[pinterest_pipeline] ⛔ 检测到图像服务 503（No available compatible accounts），" "重试无效 → 提前终止任务，未完成产品将废弃，仅保留已完成产品") return first def is_fatal_503_aborted(self) -> bool: with self._fatal_lock: return self._fatal_503 def abort_unfinished(self) -> None: """终止分发：丢弃简报池中所有未完成简报（已完成的落盘产品保留）。""" with self._cond: dropped = len(self._briefs) self._briefs = [] self._done = True self._cond.notify_all() if dropped: print(f"[pinterest_pipeline] 503 终止：丢弃未完成简报 {dropped} 条（未完成产品废弃）") def _abort_current_term(self) -> None: """放弃当前种子词：清空其未完成简报 + 图池未消费图片（已完成的保留）。""" term = self._err400_term with self._cond: kept = [b for b in self._briefs if not self._brief_of_term(b, term)] dropped = len(self._briefs) - len(kept) self._briefs = kept if dropped: print(f"[pinterest_pipeline] 放弃「{term}」未完成简报 {dropped} 条") self._drop_term_pool(term) @staticmethod def _brief_of_term(b: Dict[str, Any], term: str) -> bool: """简报是否属于某种子词（topic 去掉 #N 后缀后 == term）。""" if not term: return False topic = str(b.get("topic") or "").strip() base = re.sub(r"\s+#\d+$", "", topic).strip().lower() return bool(base) and base == term.strip().lower() def _drop_term_pool(self, term: str) -> None: """清除图池中属于当前种子词的图片，并把它们的 md5 全部拉黑（used_images.json）。 400 超限说明这批图反复触发内容/图片 400，整批拉黑防止下次重新爬取到相同图再次触发。 """ if not term: return try: from graph.pinterest import ( load_image_pool, save_image_pool, load_used_images, save_used_images, ) pool = load_image_pool(str(self.output_dir), self.country) imgs = pool.get("images") or [] term_imgs = [img for img in imgs if str(img.get("term") or "").strip().lower() == term.strip().lower()] kept = [img for img in imgs if img not in term_imgs] if len(kept) < len(imgs): pool["images"] = kept save_image_pool(str(self.output_dir), self.country, pool) print(f"[pinterest_pipeline] 清除图池「{term}」图片 {len(imgs) - len(kept)} 张") md5s = [str(img.get("md5") or "").strip().lower() for img in term_imgs] md5s = [m for m in md5s if m] if md5s: used = load_used_images(str(self.output_dir), self.country) before = len(used) used.update(md5s) if len(used) > before: save_used_images(str(self.output_dir), self.country, used) print(f"[pinterest_pipeline] 400 超限：拉黑「{term}」图片 md5 {len(md5s)} 个") except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 清除图池失败: {e}") def add_briefs(self, briefs: List[Dict[str, Any]]) -> None: if not briefs: return with self._cond: self._briefs.extend(briefs) self._cond.notify_all() print(f"[pinterest_pipeline] 简报池 +{len(briefs)} 条（待处理 {len(self._briefs)}）") def pending_count(self) -> int: """简报池中待处理 + 正在处理的简报数（供路由判断是否需要补图/补分析）。""" with self._cond: queued = len(self._briefs) return queued + self._in_flight def wait_idle(self, timeout: Optional[float] = None) -> bool: """阻塞等待简报池消化完（无待处理且无在途），返回是否已空闲。 用 Condition 等待（_process_one 完成时 notify_all 唤醒），而非轮询 sleep， 避免 wait 循环疯狂刷屏。timeout 为 None 时无限等待（受 _done 保护）。 """ with self._cond: while (self._briefs or self._in_flight > 0) and not self._done: if timeout is not None: deadline = time.time() + timeout remaining = deadline - time.time() if remaining <= 0: return False self._cond.wait(min(remaining, 1.0)) else: self._cond.wait() return not self._briefs and self._in_flight <= 0 def __enter__(self): return self def __exit__(self, exc_type, exc, tb): # 异常路径也确保排空并释放线程池，避免 dispatcher/worker 泄漏 try: self.finish() except Exception: # noqa: BLE001 pass return False def finish(self) -> tuple: """排空简报池、等待全部产品完成，返回 (products, errors)。 产品来源：手动已完成（内存缓存）+ 落盘文件（products_pending.jsonl） 按货号去重合并——即使中途 503 终止/崩溃，已完成产品也不丢。 """ with self._cond: self._done = True self._cond.notify_all() self._dispatcher.join() self._pool.shutdown(wait=True) # 读盘 + 内存合并去重（内存为准，但以落盘为最终权威——崩溃恢复后走落盘） pending = self.load_pending() merged = {str(p.get("img_code", "")): p for p in pending} with self._products_lock: for p in self._products: merged[str(p.get("img_code", ""))] = p products = [merged[k] for k in merged if k] with self._errors_lock: errors = list(self._errors) print(f"[pinterest_pipeline] 收尾：完成 {len(products)} 个产品，错误 {len(errors)}" f"{'（含落盘恢复 ' + str(len(pending)) + '）' if pending else ''}") return products, errors # ------------------------------------------------------------------ # # 后台线程 # ------------------------------------------------------------------ # def _dispatch(self) -> None: while True: with self._cond: while not self._briefs and not self._done: self._cond.wait() if self._done and not self._briefs: break if self._fatal_503: # 致命 503：不再分发新简报（未完成的废弃，仅保留已完成落盘产品） self._briefs = [] self._done = True self._cond.notify_all() break batch = self._briefs self._briefs = [] # 与 _briefs 清空同一临界区递增在途数，避免 wait_idle 误判空闲 self._in_flight += len(batch) for b in batch: with self._cond: idx = self._cursor self._cursor += 1 self._pool.submit(self._process_one, b, idx) # ------------------------------------------------------------------ # # 单条简报完整链路：设计 → 三合一 → OSS → 种草图 # ------------------------------------------------------------------ # def _process_one(self, brief: Dict[str, Any], idx: int) -> None: try: if self.is_fatal_503_aborted(): return # 0) 先定货号：整条链路（设计/三合一/种草图）都用它命名与匹配，避免序号错位 if idx >= len(self._worklist): print(f"[pinterest_pipeline] 简报 {idx} 无对应产品任务，跳过") return spu, skus = self._worklist[idx] img_code = f"{self._prefix}{idx:03d}" # 1) 生成设计（compose）——直接按货号命名 designs/{img_code}_design.png design_path = self._gen_design(brief, img_code) if self.is_fatal_503_aborted(): return if self.is_400_aborted(): # 当前种子词 400 超限已放弃：正在生成的当个也放弃，不进入后续链路 print(f"[pinterest_pipeline] 当前种子词 400 超限已放弃，跳过简报 {idx}") return if not design_path: # 生成失败/侵权（MD5 全局重复/API 错误）→ 从图池补充图片重新分析，最多尝试 N 次； # 图池不足 → 返回 None，由路由在下一轮触发搜索 for _ in range(self._supply_attempts): new_brief = self._supply_from_pool( reason=f"简报「{brief.get('topic','')}」设计生成失败") if new_brief is None: return brief = new_brief design_path = self._gen_design(brief, img_code) if self.is_fatal_503_aborted(): return if design_path: break if not design_path: return brief["design_path"] = design_path # 2) 三合一（product）——同一货号；task_idx=本次简报序号，模特按任务独立随机 prod = self._process_spu(brief, spu, skus, img_code, design_path, task_idx=idx) if self.is_fatal_503_aborted(): return if not prod: return # 3) OSS 上传 self._upload_product(prod) # 4) 种草图——同一货号 self._seed_shot(prod) # 去重记录 try: from graph.nodes.product_node import _record_used _record_used(self.cache_dir, prod) except Exception: # noqa: BLE001 pass # 5) 落盘：每完成一个产品立即追加写入 products_pending.jsonl（不依赖内存，崩溃不丢） self._persist_product(prod) with self._products_lock: self._products.append(prod) print(f"[pinterest_pipeline] 产品完成: {prod.get('img_code', '')}" f"（累计 {len(self._products)}）") except Exception as e: # noqa: BLE001 with self._errors_lock: self._errors.append({"node": "pinterest_pipeline", "type": type(e).__name__, "message": f"简报 {idx} 处理失败: {e}", "trace": ""}) print(f"[pinterest_pipeline] 简报 {idx} 处理失败: {e}") finally: with self._cond: self._in_flight = max(0, self._in_flight - 1) self._cond.notify_all() def _persist_product(self, prod: Dict[str, Any]) -> None: """把已完成产品追加写入 products_pending.jsonl（JSONL 每行一个产品）。 并发安全：写入持 _products_lock，整行一次写（含换行），避免并发 append 交织； 落盘失败不阻塞主流程（仅告警）；finish() 时读盘合并，保证已完成产品不丢。 """ try: import json as _json self._pending_file.parent.mkdir(parents=True, exist_ok=True) line = _json.dumps(prod, ensure_ascii=False) + "\n" with self._products_lock: with open(self._pending_file, "a", encoding="utf-8") as f: f.write(line) except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 产品落盘失败（不影响流程）: {e}") def load_pending(self) -> List[Dict[str, Any]]: """读回 products_pending.jsonl 中已落盘的产品（进程重启/崩溃恢复用）。""" import json as _json out: List[Dict[str, Any]] = [] if not self._pending_file.exists(): return out try: for line in self._pending_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: out.append(_json.loads(line)) except Exception: # noqa: BLE001 continue except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 读回落盘产品失败: {e}") return out def _gen_design(self, brief: Dict[str, Any], img_code: str) -> Optional[str]: if self._ib is None: return None try: from graph.nodes.compose_node import generate_design design_dir = self.output_dir / "designs" design_dir.mkdir(parents=True, exist_ok=True) def _on_400(): if self.record_400(): self._abort_current_term() def _on_503(): self.record_503() self.abort_unfinished() return generate_design(self._ib, brief, design_dir, img_code, self._errors, on_400=_on_400, on_503=_on_503, size=str((self.config.get("compose") or {}).get("design_size") or "1024x1024")) except Exception as e: # noqa: BLE001 if self.is_fatal_503(e): self.record_503() self.abort_unfinished() return None with self._errors_lock: self._errors.append({"node": "compose", "type": type(e).__name__, "message": f"设计稿生成失败 {brief.get('topic', '')}: {e}", "trace": ""}) return None def _supply_from_pool(self, reason: str) -> Optional[Dict[str, Any]]: """生成失败/侵权时，从图池取一张未消费图片重新分析，产出新简报。 图池不足 → 返回 None（由路由在下一轮触发搜索）。该图片分析后 md5 一律拉黑 （合适/不合适都拉黑），避免重复分析。 """ from graph.pinterest import ( compress_image, load_image_pool, load_used_images, pool_unused_images, save_used_images, ) pool = load_image_pool(str(self.output_dir), self.country) used = load_used_images(str(self.output_dir), self.country) unused = pool_unused_images(pool, used) if not unused: print(f"[pinterest_pipeline] 图池无未消费图片，无法补充（{reason}），等待路由搜索") return None img = unused[0] compressed = compress_image(img["path"]) llm = self._analyze_backend if llm is None or not hasattr(llm, "analyze_pinterest_images"): return None try: def _on_400(): if self.record_400(): self._abort_current_term() res = llm.analyze_pinterest_images([compressed], img.get("term", ""), self.country, on_400=_on_400) or [] except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 补充分析失败: {e}") res = [] # 该图片已消费 → 拉黑（合适/不合适都拉黑） if img.get("md5"): used.add(str(img["md5"]).lower()) save_used_images(str(self.output_dir), self.country, used) if not res or not isinstance(res[0], dict): return None b = res[0] from graph.nodes.pinterest_analyze_node import _brief_suitable if not _brief_suitable(b): print(f"[pinterest_pipeline] 补充简报侵权/不适合印花，丢弃: {b.get('topic','')}") return None b["ref_images"] = [img["path"]] b["source_md5"] = str(img.get("md5") or "").strip().lower() try: from graph.nodes.pinterest_analyze_node import _enrich_briefs from graph.nodes.prompt_node import prompt_node screened = _enrich_briefs([b], self.country) if not screened: return None r = prompt_node({ "config": self.config, "country": self.country, "country_config": self._country_config, "screened": screened, }) new_briefs = r.get("briefs") or [] if new_briefs: print(f"[pinterest_pipeline] 图池补充成功（{reason}）: {new_briefs[0].get('topic','')}") return new_briefs[0] except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 补充简报装配失败: {e}") return None def _model_source(self, task_idx: int, spu=None) -> Dict[str, Any]: """返回某任务给 _process_spu 的图源参数：model_img / model_kind / prompts。""" src = self._model_assign.get(f"task_{task_idx}") or {} img = src.get("img") if img is None: return {} if spu is not None and int(spu.get("mark") or 0) != 1: return {"model_img": img} # 非 mark=1 走旧逻辑（不传 kind，product_node 按其 mark 自行判定） return {"model_img": img, "model_kind": src.get("kind", "model")} def _process_spu(self, brief: Dict[str, Any], spu, skus: str, img_code: str, design_path: str, task_idx: int = 0) -> Optional[Dict[str, Any]]: from graph.nodes.product_node import _process_spu as _ps prod_dir = self.output_dir / "product" prod_dir.mkdir(parents=True, exist_ok=True) # 设计稿已按货号命名（designs/{img_code}_design.png），直接复用，无需再拷贝 brief = dict(brief) brief["design_path"] = design_path r = _ps(self._db_path, self._basemap_root, self._material_root, self._category, prod_dir, brief, self._ib, spu, skus, self.config.get("product") or {}, self._errors, design_path, self._title_backend, self.country, img_code=img_code, on_503=lambda: (self.record_503(), self.abort_unfinished()), **self._model_source(task_idx, spu)) if r: r["img_code"] = img_code return r def _upload_product(self, r: Dict[str, Any]) -> None: if not self._oss_enabled: return from graph.oss_upload import build_oss_key, compress_for_oss, upload_to_oss from graph.nodes.oss_upload_node import KIND_ORDER, _gen_rand4 # 货号直接取产品 img_code：同一货号的所有图片（合成/平铺/底图）共用同一货号， # 避免独立计数器在有产品被跳过时与 img_code 错位 base_code = r.get("img_code") or r.get("oss_code") or "" if not base_code: return for kind in KIND_ORDER: src = r.get(f"{kind}_path") if not src or not Path(src).exists(): continue try: compressed = compress_for_oss(src, str(Path(src).with_suffix(".oss.jpg"))) key = build_oss_key(self.country, self.task_timestamp, base_code, _gen_rand4()) url = upload_to_oss(self._oss_cfg, compressed, key) if url: r[f"{kind}_url"] = url r["oss_code"] = base_code except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] OSS 上传失败 {src}: {e}") # 多色：首色用主图 url/code，额外色单独上传（同一货号） comps = r.get("color_composites") or [] if comps and r.get("composite_url"): comps[0]["url"] = r["composite_url"] comps[0]["code"] = r.get("oss_code", "") for cc in comps[1:]: src = cc.get("composite_path") if not src or not Path(src).exists(): continue try: compressed = compress_for_oss(src, str(Path(src).with_suffix(".oss.jpg"))) key = build_oss_key(self.country, self.task_timestamp, base_code, _gen_rand4()) url = upload_to_oss(self._oss_cfg, compressed, key) if url: cc["url"] = url cc["code"] = base_code except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] OSS 颜色图上传失败 {src}: {e}") def _seed_shot(self, r: Dict[str, Any]) -> None: ss_cfg = self.config.get("seed_shot") or {} count = int(ss_cfg.get("count", 1)) if count <= 0 or not bool(ss_cfg.get("enabled", True)) or self._ib is None: return from graph.nodes.seed_shot_node import _plan_seed_shots from graph.seed_shot import generate_seed_shots from graph.oss_upload import build_oss_key, compress_for_oss, upload_to_oss from graph.nodes.oss_upload_node import MAX_CODE, _gen_rand4 comps = r.get("color_composites") or [] if not comps and r.get("composite_path") and Path(r["composite_path"]).exists(): comps = [{"sku_code": r.get("sku_code"), "color": r.get("color", ""), "composite_path": r["composite_path"]}] if not comps: return plan = _plan_seed_shots(comps, count) cn = (r.get("cn_title") or "").strip() or r.get("topic", "") material = self._material_map.get(r.get("spu_code", ""), "") base_prefix = r.get("img_code") or r.get("oss_code") or "" pfx = base_prefix or "seed" shot_dir = self.output_dir / "seed_shots" shot_dir.mkdir(parents=True, exist_ok=True) size = str(ss_cfg.get("size") or "1536x2048") paths: List[str] = [] for cc, n in plan: base = cc.get("composite_path") if not base or not Path(base).exists(): print(f"[pinterest_pipeline] {r.get('spu_code', '')} 参考图缺失，跳过该色种草图") continue try: generated = generate_seed_shots(self._ib, base, cn, material, n, str(shot_dir), r.get("composite_negative", ""), size=size, prefix=pfx, gender=self._gender, retries=int(ss_cfg.get("retries", 3))) except Exception as e: # noqa: BLE001 if self.is_fatal_503(e): self.record_503() self.abort_unfinished() return print(f"[pinterest_pipeline] 种草图生成失败（跳过该色）: {e}") continue paths.extend(generated) if not paths: return r["seed_shot_paths"] = paths urls: List[str] = [] for pth in paths: with self._oss_lock: if not base_prefix: if self.oss_seq >= MAX_CODE: break code = f"{self._prefix}{self.oss_seq:03d}" self.oss_seq += 1 else: code = base_prefix if self._oss_enabled: try: compressed = compress_for_oss(pth, str(Path(pth).with_suffix(".oss.jpg"))) url = upload_to_oss(self._oss_cfg, compressed, build_oss_key(self.country, self.task_timestamp, code, _gen_rand4())) if url: urls.append(url) r["seed_shot_urls"] = urls except Exception as e: # noqa: BLE001 print(f"[pinterest_pipeline] 种草图上传失败 {pth}: {e}")