51 lines
2.1 KiB
Python
51 lines
2.1 KiB
Python
"""Pinterest 爬取封装：把 pinterest_image_capture 的 CLI 逻辑封装成可调用函数，供 UI/节点后台线程调用。
|
|
|
|
用法：
|
|
from pinterest_scraper.scraper import scrape_pinterest
|
|
files = scrape_pinterest("vintage 70s", count=40, save_dir="output/pinterest_ref/US/vintage_70s")
|
|
"""
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from pinterest_scraper.pinterest_image_capture import (
|
|
detect_proxy,
|
|
download_all,
|
|
get_system_proxy,
|
|
scrape,
|
|
)
|
|
|
|
|
|
def scrape_pinterest(keyword: str, count: int = 40, save_dir: Optional[str] = None,
|
|
proxy: Optional[str] = None, headless: bool = False,
|
|
cdp_url: Optional[str] = None) -> List[str]:
|
|
"""按关键词爬取 Pinterest 图片并下载到 save_dir，返回下载成功的文件路径列表。
|
|
|
|
- proxy 为 None 时自动探测（环境变量/系统代理/本地常见端口）；空字符串显式禁用。
|
|
- 首次运行会弹出本地 Chrome 窗口，需手动登录 Pinterest 一次，登录态缓存在
|
|
pinterest_scraper/.chrome_session，之后自动复用。
|
|
"""
|
|
# 代理解析：显式传入 > 自动探测 > 直连
|
|
if proxy is None:
|
|
proxy = detect_proxy() or get_system_proxy()
|
|
if proxy:
|
|
print(f"[pinterest] 使用代理: {proxy}")
|
|
else:
|
|
print("[pinterest] 未检测到代理，将直连（如无法访问请手动指定代理）")
|
|
proxy = proxy or None
|
|
|
|
out = save_dir or os.path.join(os.getcwd(), "output", "img", keyword)
|
|
os.makedirs(out, exist_ok=True)
|
|
print(f"[pinterest] 关键词: {keyword} | 数量: {count} | 保存目录: {out}")
|
|
|
|
imgs_url = asyncio.run(scrape(keyword, count, headless, proxy, cdp_url))
|
|
if not imgs_url:
|
|
print(f"[pinterest] 未收集到图片（可能未登录或搜索无结果）: {keyword}")
|
|
return []
|
|
asyncio.run(download_all(imgs_url, out, proxy))
|
|
|
|
files = [str(p) for p in sorted(Path(out).iterdir()) if p.is_file()]
|
|
print(f"[pinterest] 完成，共下载 {len(files)} 张图片: {keyword}")
|
|
return files
|