模板导出增强 + 模特性别分组 + 三合一提示词精简
1) 模板导出：识别「基码表-胸围」填 sku.bust（多个胸围列都填）；申报价格模糊匹配多列统一按加价后价格填写；详情图文不再拼接 img_url_2；SPU 款式来源统一填「现货款」；商品产地国家简称映射（沙特→沙特阿拉伯） 2) 模特性别分组：model_features 按男女分组，按模板类目含男/女固定取对应性别模特（含 Pinterest 模式 pipeline） 3) 三合一提示词：去掉 DESIGN CONTENT 四要素描述（设计已由设计稿提供） 4) 生图尺寸：全部改为读 config 不再硬编码（设计图 compose.design_size / 合成图 compose.size / 种草图 seed_shot.size）
This commit is contained in:
+224
-1
@@ -1,10 +1,13 @@
|
||||
"""Pinterest 参考模式共享辅助：种子词加载、已用搜索词持久化、搜索词全局过滤。
|
||||
"""Pinterest 参考模式共享辅助：种子词加载、已用搜索词持久化、搜索词全局过滤、设计图全局 MD5 去重。
|
||||
|
||||
独立于 Google Trends 采集链路，供 pinterest_search / scrape / analyze 节点复用。
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import random
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
|
||||
@@ -114,3 +117,223 @@ def merge_used(existing: List[str], new_terms: List[str]) -> List[str]:
|
||||
seen.add(k)
|
||||
out.append(str(t).strip())
|
||||
return out
|
||||
|
||||
|
||||
# —— 设计图全局 MD5 去重（跨国家、跨运行；对所有国家生效）——
|
||||
_MD5_LOCK = threading.Lock()
|
||||
|
||||
|
||||
def global_md5_path() -> Path:
|
||||
"""全局设计图 MD5 过滤文件（.cache 随打包同步，跨版本保留）。"""
|
||||
return runtime_root() / ".cache" / "global_design_md5.json"
|
||||
|
||||
|
||||
def load_global_md5() -> set:
|
||||
"""读全局设计图 MD5 集合。"""
|
||||
try:
|
||||
p = global_md5_path()
|
||||
if p.exists():
|
||||
data = json.loads(p.read_text(encoding="utf-8"))
|
||||
return {str(m).strip().lower() for m in (data.get("md5s") or []) if str(m).strip()}
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 全局 MD5 过滤读取失败: {e}")
|
||||
return set()
|
||||
|
||||
|
||||
def add_global_md5(md5: str) -> bool:
|
||||
"""把设计图 MD5 加入全局过滤；返回 True=新增可用，False=已存在（全局重复，应跳过）。"""
|
||||
md5 = str(md5 or "").strip().lower()
|
||||
if not md5:
|
||||
return False
|
||||
with _MD5_LOCK:
|
||||
try:
|
||||
p = global_md5_path()
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
md5s = load_global_md5()
|
||||
if md5 in md5s:
|
||||
return False
|
||||
md5s.add(md5)
|
||||
p.write_text(json.dumps(
|
||||
{"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "md5s": sorted(md5s)},
|
||||
ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
return True
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 全局 MD5 过滤写入失败: {e}")
|
||||
return True # 写入失败不阻塞：按新增处理，避免误跳过
|
||||
|
||||
|
||||
def design_md5_ok(image_path: str) -> bool:
|
||||
"""计算设计图 MD5 并加入全局过滤；返回 True=新增可用，False=全局重复（应跳过）。"""
|
||||
try:
|
||||
md5 = hashlib.md5(Path(image_path).read_bytes()).hexdigest()
|
||||
return add_global_md5(md5)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 设计图 MD5 计算失败: {e}")
|
||||
return True
|
||||
|
||||
|
||||
# —— 图池（Image Pool）：跨轮次/跨运行持久化的爬取图片池 + 已消费图片 MD5 拉黑 ——
|
||||
# 图池 = 所有已爬取图片的注册表（path + md5 + term）；已消费（分析过）的图片 MD5 记入
|
||||
# used_images，不再复用。分析从图池取未消费图片，池不足时由路由触发新一轮搜索。
|
||||
|
||||
_IMG_EXTS = (".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp")
|
||||
|
||||
|
||||
def image_pool_path(output_dir: str, country: str) -> Path:
|
||||
return Path(output_dir) / "pinterest_ref" / country / "image_pool.json"
|
||||
|
||||
|
||||
def used_images_path(output_dir: str, country: str) -> Path:
|
||||
return Path(output_dir) / "pinterest_ref" / country / "used_images.json"
|
||||
|
||||
|
||||
def load_image_pool(output_dir: str, country: str) -> Dict[str, Any]:
|
||||
"""读图池注册表；无文件时回退扫描文件系统重建。"""
|
||||
try:
|
||||
p = image_pool_path(output_dir, country)
|
||||
if p.exists():
|
||||
data = json.loads(p.read_text(encoding="utf-8")) or {}
|
||||
imgs = data.get("images") or []
|
||||
if imgs:
|
||||
return {"updated_at": data.get("updated_at", ""), "images": imgs}
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 图池读取失败: {e}")
|
||||
return {"updated_at": "", "images": scan_scraped_images(output_dir, country)}
|
||||
|
||||
|
||||
def save_image_pool(output_dir: str, country: str, pool: Dict[str, Any]) -> None:
|
||||
"""持久化图池注册表。"""
|
||||
try:
|
||||
p = image_pool_path(output_dir, country)
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(json.dumps(
|
||||
{"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "images": pool.get("images") or []},
|
||||
ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 图池保存失败: {e}")
|
||||
|
||||
|
||||
def scan_scraped_images(output_dir: str, country: str) -> List[Dict[str, Any]]:
|
||||
"""扫描 pinterest_ref/<country>/*/ 下所有图片，重建图池注册表（含 md5）。"""
|
||||
base = Path(output_dir) / "pinterest_ref" / country
|
||||
out: List[Dict[str, Any]] = []
|
||||
if not base.exists():
|
||||
return out
|
||||
for term_dir in sorted(base.iterdir()):
|
||||
if not term_dir.is_dir():
|
||||
continue
|
||||
term = term_dir.name
|
||||
for f in sorted(term_dir.iterdir()):
|
||||
if not f.is_file() or f.suffix.lower() not in _IMG_EXTS:
|
||||
continue
|
||||
out.append({"path": str(f), "md5": image_md5(str(f)), "term": term})
|
||||
return out
|
||||
|
||||
|
||||
def image_md5(path: str) -> str:
|
||||
"""计算图片文件 MD5（失败返回空串）。"""
|
||||
try:
|
||||
return hashlib.md5(Path(path).read_bytes()).hexdigest()
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 图片 MD5 计算失败 {path}: {e}")
|
||||
return ""
|
||||
|
||||
|
||||
def load_used_images(output_dir: str, country: str) -> set:
|
||||
"""读已消费图片 MD5 集合（拉黑，不再复用）。"""
|
||||
try:
|
||||
p = used_images_path(output_dir, country)
|
||||
if p.exists():
|
||||
data = json.loads(p.read_text(encoding="utf-8")) or {}
|
||||
return {str(m).strip().lower() for m in (data.get("md5s") or []) if str(m).strip()}
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 已消费图片读取失败: {e}")
|
||||
return set()
|
||||
|
||||
|
||||
def save_used_images(output_dir: str, country: str, md5s: set) -> None:
|
||||
"""持久化已消费图片 MD5 集合。"""
|
||||
try:
|
||||
p = used_images_path(output_dir, country)
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(json.dumps(
|
||||
{"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "md5s": sorted(md5s)},
|
||||
ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 已消费图片保存失败: {e}")
|
||||
|
||||
|
||||
def pool_unused_images(pool: Dict[str, Any], used: set) -> List[Dict[str, Any]]:
|
||||
"""从图池取未消费图片（md5 不在 used 集合），保序。"""
|
||||
out = []
|
||||
for img in pool.get("images") or []:
|
||||
if not isinstance(img, dict):
|
||||
continue
|
||||
m = str(img.get("md5") or "").strip().lower()
|
||||
if not m or m in used:
|
||||
continue
|
||||
if not Path(str(img.get("path") or "")).exists():
|
||||
continue
|
||||
out.append(img)
|
||||
return out
|
||||
|
||||
|
||||
def compress_image(path: str, max_dim: int = 1024, max_bytes: int = 1_500_000,
|
||||
out_dir: str = "") -> str:
|
||||
"""压缩大图：超过 max_dim 边长或 max_bytes 体积时缩放/重编码，返回压缩后路径。
|
||||
|
||||
- 内存占用过大（分辨率过高）→ 等比缩放到 max_dim 内；
|
||||
- 文件过大 → 转 JPEG 重编码（质量自适应）；
|
||||
- 无需压缩 → 返回原路径。压缩产物存 out_dir（默认图片同目录 .compressed/）。
|
||||
"""
|
||||
try:
|
||||
from PIL import Image
|
||||
p = Path(path)
|
||||
if not p.exists():
|
||||
return path
|
||||
size = p.stat().st_size
|
||||
try:
|
||||
with Image.open(p) as im:
|
||||
w, h = im.size
|
||||
except Exception: # noqa: BLE001 无法解析的图片（损坏/非标准）直接返回原路径
|
||||
return path
|
||||
if w <= max_dim and h <= max_dim and size <= max_bytes:
|
||||
return path
|
||||
out_root = Path(out_dir) if out_dir else (p.parent / ".compressed")
|
||||
out_root.mkdir(parents=True, exist_ok=True)
|
||||
out = out_root / f"{p.stem}_c.jpg"
|
||||
with Image.open(p) as im:
|
||||
im = im.convert("RGB")
|
||||
im.thumbnail((max_dim, max_dim), Image.LANCZOS)
|
||||
quality = 85
|
||||
while quality >= 40:
|
||||
im.save(out, "JPEG", quality=quality, optimize=True)
|
||||
if out.stat().st_size <= max_bytes:
|
||||
break
|
||||
quality -= 15
|
||||
print(f"[pinterest] 图片压缩 {p.name} ({w}x{h}, {size // 1024}KB) → "
|
||||
f"{out.name} ({out.stat().st_size // 1024}KB)")
|
||||
return str(out)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[pinterest] 图片压缩失败 {path}: {e}")
|
||||
return path
|
||||
|
||||
|
||||
def is_400_content_image(exc) -> bool:
|
||||
"""HTTP 400 且错误信息模糊匹配到「内容」或「图片」才算（用户要求）。
|
||||
|
||||
生图后端抛 RuntimeError("图像 API 400: {body}")，body 在消息里；
|
||||
多模态后端抛 requests.HTTPError，响应体在 exc.response.text。
|
||||
"""
|
||||
msg = str(exc or "")
|
||||
resp = getattr(exc, "response", None)
|
||||
if resp is not None:
|
||||
try:
|
||||
body = resp.text or ""
|
||||
except Exception: # noqa: BLE001
|
||||
body = ""
|
||||
if body:
|
||||
msg = f"{msg} {body}"
|
||||
if "400" not in msg:
|
||||
return False
|
||||
return ("内容" in msg or "图片" in msg)
|
||||
|
||||
Reference in New Issue
Block a user