"""Pinterest 参考模式节点 2/3:爬取图片(pinterest_scrape)。 对 pinterest_search 生成的每个搜索词,调 pinterest_scraper.scraper.scrape_pinterest (Playwright 启动本地 Chrome)搜索 Pinterest 并下载图片到 output/pinterest_ref/<国家>/<搜索词>/。 - 单个搜索词失败(未登录/网络/无结果)跳过,不中断整批。 - 并发数由 config.pinterest.scrape_concurrency 控制(每个并发开一个 Chrome 窗口)。 - 已爬取过且图片数达标的搜索词跳过(断点续爬,避免重复开 Chrome)。 """ import concurrent.futures from pathlib import Path from typing import Any, Dict, List from graph.validate import with_fallback def _term_dir(output_dir: str, country: str, term: str) -> Path: safe = "".join(ch for ch in term if ch.isalnum() or ch in "-_ ").strip() or "term" return Path(output_dir) / "pinterest_ref" / country / safe def _already_scraped(term_dir: Path) -> bool: """该搜索词已爬取过(目录里已有 ≥1 张图)→ 跳过,避免重复开 Chrome。""" if not term_dir.exists(): return False return any(p.is_file() and p.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp") for p in term_dir.iterdir()) @with_fallback("pinterest_scrape") def pinterest_scrape_node(state: Dict[str, Any]) -> Dict[str, Any]: terms: List[str] = state.get("pinterest_search_terms") or [] if not terms: print("[pinterest_scrape] 无搜索词,跳过爬取") return {"pinterest_images": {}, "errors": state.get("errors") or []} country = state["country"] config = state["config"] output_dir = state["output_dir"] errors = list(state.get("errors") or []) pcfg = config.get("pinterest") or {} images_per_term = int(pcfg.get("images_per_term", 40)) concurrency = int(pcfg.get("scrape_concurrency", 2)) headless = bool(pcfg.get("headless", False)) proxy = pcfg.get("proxy") or None # 所有搜索词共享同一个 .chrome_session 登录态目录,Chrome 对同一 user-data-dir 是单例, # 并发启动会互相抢占导致 "browser has been closed",必须串行爬取。 if concurrency > 1: print(f"[pinterest_scrape] 共享登录态目录不支持并发,scrape_concurrency 强制为 1(原 {concurrency})") concurrency = 1 # 一次性探测并校验代理(Pinterest 需代理才能访问;代理失效时给出明确警告,避免逐词静默失败) if proxy is None: try: from pinterest_scraper.pinterest_image_capture import detect_proxy, get_system_proxy, _validate_proxy proxy = detect_proxy() or get_system_proxy() except Exception: # noqa: BLE001 proxy = None if proxy and not _validate_proxy(proxy): print(f"[pinterest_scrape] 警告:代理 {proxy} 无法连通外网,请检查代理/VPN 是否正常," f"否则 Pinterest 将无法访问(爬取会失败)") results: Dict[str, List[str]] = {} skipped: List[str] = [] def _one(term: str) -> None: term_dir = _term_dir(output_dir, country, term) if _already_scraped(term_dir): skipped.append(term) print(f"[pinterest_scrape] 已爬取过(跳过): {term}") return try: from pinterest_scraper.scraper import scrape_pinterest files = scrape_pinterest(term, count=images_per_term, save_dir=str(term_dir), proxy=proxy, headless=headless) results[term] = files except Exception as e: # noqa: BLE001 errors.append({"node": "pinterest_scrape", "type": type(e).__name__, "message": f"term[{term}]: {e}", "trace": ""}) print(f"[pinterest_scrape] 爬取失败(跳过): {term}: {e}") print(f"[pinterest_scrape] 开始爬取 {len(terms)} 个搜索词(并发 {concurrency})…") with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, concurrency)) as ex: list(ex.map(_one, terms)) total = sum(len(v) for v in results.values()) stats = dict(state.get("stats") or {}) stats["pinterest_scrape"] = { "terms": len(terms), "scraped": len(results), "skipped": len(skipped), "images": total, } print(f"[pinterest_scrape] 完成:{len(results)} 个搜索词,共 {total} 张图(跳过 {len(skipped)})") return {"pinterest_images": results, "stats": stats, "errors": errors}