| 1 | """Pinterest search via ScrapeCreators API for /last30days. |
| 2 | |
| 3 | Uses ScrapeCreators REST API to search Pinterest by keyword, extract |
| 4 | engagement metrics (saves, comments), and return pin descriptions. |
| 5 | |
| 6 | Requires SCRAPECREATORS_API_KEY in config. 100 free API calls, then PAYG. |
| 7 | API docs: https://scrapecreators.com/docs |
| 8 | """ |
| 9 | |
| 10 | import re |
| 11 | import sys |
| 12 | from typing import Any, Dict, List, Optional, Set |
| 13 | |
| 14 | from . import dates, http, log |
| 15 | |
| 16 | SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/pinterest" |
| 17 | |
| 18 | # Depth configurations: how many results to fetch |
| 19 | DEPTH_CONFIG = { |
| 20 | "quick": {"results_per_page": 10}, |
| 21 | "default": {"results_per_page": 20}, |
| 22 | "deep": {"results_per_page": 40}, |
| 23 | } |
| 24 | |
| 25 | from .relevance import token_overlap_relevance as _compute_relevance |
| 26 | |
| 27 | |
| 28 | def _extract_core_subject(topic: str) -> str: |
| 29 | """Extract core subject from verbose query for Pinterest search.""" |
| 30 | from .query import VIRAL_NOISE, extract_core_subject |
| 31 | return extract_core_subject(topic, noise=VIRAL_NOISE) |
| 32 | |
| 33 | |
| 34 | def _log(msg: str): |
| 35 | log.source_log("Pinterest", msg, tty_only=False) |
| 36 | |
| 37 | |
| 38 | def _parse_items(raw_items: List[Dict[str, Any]], core_topic: str) -> List[Dict[str, Any]]: |
| 39 | """Parse raw Pinterest items into normalized dicts. |
| 40 | |
| 41 | Pinterest pins are visual content with descriptions. Saves are the |
| 42 | primary engagement signal (analogous to upvotes/likes on other platforms). |
| 43 | """ |
| 44 | items = [] |
| 45 | for raw in raw_items: |
| 46 | if not isinstance(raw, dict): |
| 47 | continue |
| 48 | |
| 49 | pin_id = str(raw.get("id", raw.get("pin_id", ""))) |
| 50 | description = str(raw.get("description") or raw.get("title") or "") |
| 51 | |
| 52 | # Engagement metrics - saves are the primary signal |
| 53 | save_count = raw.get("save_count") or raw.get("saves") or raw.get("repin_count") or 0 |
| 54 | comment_count = raw.get("comment_count") or raw.get("comments") or 0 |
| 55 | |
| 56 | # Author info |
| 57 | pinner = raw.get("pinner") or raw.get("creator") or raw.get("user") or {} |
| 58 | if isinstance(pinner, dict): |
| 59 | author_name = pinner.get("username") or pinner.get("full_name") or "" |
| 60 | elif isinstance(pinner, str): |
| 61 | author_name = pinner |
| 62 | else: |
| 63 | author_name = "" |
| 64 | |
| 65 | # URL |
| 66 | url = raw.get("link") or raw.get("url") or "" |
| 67 | if not url and pin_id: |
| 68 | url = f"https://www.pinterest.com/pin/{pin_id}/" |
| 69 | |
| 70 | # Board info (container for pins) |
| 71 | board = raw.get("board") or {} |
| 72 | board_name = board.get("name", "") if isinstance(board, dict) else "" |
| 73 | |
| 74 | # Compute relevance |
| 75 | relevance = _compute_relevance(core_topic, description, []) |
| 76 | |
| 77 | items.append({ |
| 78 | "pin_id": pin_id, |
| 79 | "description": description, |
| 80 | "url": url, |
| 81 | "author": author_name, |
| 82 | "board": board_name, |
| 83 | "engagement": { |
| 84 | "saves": save_count, |
| 85 | "comments": comment_count, |
| 86 | }, |
| 87 | "relevance": relevance, |
| 88 | "why_relevant": f"Pinterest: {description[:60]}" if description else f"Pinterest: {core_topic}", |
| 89 | }) |
| 90 | return items |
| 91 | |
| 92 | |
| 93 | def parse_pinterest_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 94 | """Parse Pinterest search response to normalized format. |
| 95 | |
| 96 | Returns: |
| 97 | List of item dicts ready for normalization. |
| 98 | """ |
| 99 | return response.get("items", []) |
| 100 | |
| 101 | |
| 102 | def search_pinterest( |
| 103 | topic: str, |
| 104 | from_date: str, |
| 105 | to_date: str, |
| 106 | depth: str = "default", |
| 107 | token: str = None, |
| 108 | ) -> Dict[str, Any]: |
| 109 | """Search Pinterest via ScrapeCreators API. |
| 110 | |
| 111 | Args: |
| 112 | topic: Search topic |
| 113 | from_date: Start date (YYYY-MM-DD) |
| 114 | to_date: End date (YYYY-MM-DD) |
| 115 | depth: 'quick', 'default', or 'deep' |
| 116 | token: ScrapeCreators API key |
| 117 | |
| 118 | Returns: |
| 119 | Dict with 'items' list and optional 'error'. |
| 120 | """ |
| 121 | if not token: |
| 122 | return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} |
| 123 | |
| 124 | config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 125 | core_topic = _extract_core_subject(topic) |
| 126 | |
| 127 | _log(f"Searching Pinterest for '{core_topic}' (depth={depth}, count={config['results_per_page']})") |
| 128 | |
| 129 | try: |
| 130 | data = http.get( |
| 131 | f"{SCRAPECREATORS_BASE}/search", |
| 132 | params={"query": core_topic}, |
| 133 | headers=http.scrapecreators_headers(token), |
| 134 | timeout=30, |
| 135 | retries=2, |
| 136 | ) |
| 137 | except Exception as e: |
| 138 | _log(f"ScrapeCreators error: {e}") |
| 139 | return {"items": [], "error": f"{type(e).__name__}: {e}"} |
| 140 | |
| 141 | # Extract items from response - try common SC response shapes |
| 142 | raw_items = data.get("pins") or data.get("results") or data.get("data") or data.get("items") or [] |
| 143 | |
| 144 | # Limit to configured count |
| 145 | raw_items = raw_items[:config["results_per_page"]] |
| 146 | |
| 147 | # Parse items |
| 148 | items = _parse_items(raw_items, core_topic) |
| 149 | |
| 150 | # Sort by saves descending (primary engagement signal) |
| 151 | items.sort(key=lambda x: x["engagement"]["saves"], reverse=True) |
| 152 | |
| 153 | _log(f"Found {len(items)} Pinterest pins") |
| 154 | return {"items": items} |
| 155 |