import json import time import logging import threading import random from curl_cffi.requests import Session as CurlSession from urllib.parse import quote, quote_plus from .models import SearchResponse from .utils import current_epoch_ms USER_AGENTS = [ { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", "sec_ch_ua": '"Chromium";v="138", "Not/A)Brand";v="24", "Google Chrome";v="138"', "platform": '"Windows"', }, { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0", "sec_ch_ua": '"Chromium";v="137", "Not/A)Brand";v="24", "Microsoft Edge";v="137"', "platform": '"Windows"', }, { "ua": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", "sec_ch_ua": '"Chromium";v="138", "Not/A)Brand";v="24", "Google Chrome";v="138"', "platform": '"macOS"', }, { "ua": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.5 Safari/605.1.15", "sec_ch_ua": '"Not/A)Brand";v="24", "Safari";v="18"', "platform": '"macOS"', }, { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0", "sec_ch_ua": "", "platform": '"Windows"', }, { "ua": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", "sec_ch_ua": "", "platform": '"macOS"', }, { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36 OPR/120.0.0.0", "sec_ch_ua": '"Chromium";v="136", "Not/A)Brand";v="24", "Opera";v="120"', "platform": '"Windows"', }, { "ua": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", "sec_ch_ua": '"Chromium";v="138", "Not/A)Brand";v="24", "Google Chrome";v="138"', "platform": '"Linux"', }, { "ua": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0", "sec_ch_ua": "", "platform": '"Linux"', }, { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36 Edg/135.0.0.0", "sec_ch_ua": '"Chromium";v="135", "Not/A)Brand";v="24", "Microsoft Edge";v="135"', "platform": '"Windows"', }, { "ua": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36 Edg/136.0.0.0", "sec_ch_ua": '"Chromium";v="136", "Not/A)Brand";v="24", "Microsoft Edge";v="136"', "platform": '"macOS"', }, { "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Vivaldi/7.4", "sec_ch_ua": '"Chromium";v="138", "Not/A)Brand";v="24", "Vivaldi";v="7"', "platform": '"Windows"', }, ] IMPERSONATE_PROFILES = ["chrome", "chrome110", "chrome116", "chrome120", "safari15_5", "edge99"] class Pinterest: BASE_URL = "https://in.pinterest.com" _proxy_lock = threading.Lock() _shared_proxy_index = 0 def __init__(self, user_agent: str = "", proxies: dict = None, proxy_list: list = None, sleep_time=None, max_retries: int = 3): self.errors = [] self.max_retries = max_retries self.sleep_time = sleep_time if sleep_time is not None else (1.5, 4.0) self.proxies = proxies or {} self.proxy_list = proxy_list or [] self._fixed_ua = bool(user_agent) self.user_agent = user_agent or USER_AGENTS[0]["ua"] self.BASE_HEADERS = { 'Host': self.BASE_URL.replace('https://', ''), 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Ch-Ua': '"Chromium";v="137", "Not/A)Brand";v="24"', 'Sec-Ch-Ua-Model': '""', 'Sec-Ch-Ua-Mobile': '?0', 'X-Requested-With': 'XMLHttpRequest', 'Accept': 'application/json, text/javascript, */*, q=0.01', 'X-Pinterest-Source-Url': '', 'X-Pinterest-Appstate': 'active', 'Accept-Language': 'en-US,en;q=0.9', 'Screen-Dpr': '1', 'X-Pinterest-Pws-Handler': 'www/search/[scope].js', 'User-Agent': self.user_agent, 'Sec-Ch-Ua-Platform-Version': '""', 'Sec-Fetch-Site': 'same-origin', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Dest': 'empty', 'Referer': f'{self.BASE_URL}/', 'Priority': 'u=1, i', } self._init_session() # ----------------------------- # Session management (curl_cffi) # ----------------------------- def _init_session(self): profile = random.choice(IMPERSONATE_PROFILES) self.session = CurlSession(impersonate=profile) self._current_profile = profile def reset_session(self): try: self.session.close() except Exception: pass self._init_session() # ----------------------------- # Proxy rotation # ----------------------------- def _get_next_proxy(self) -> dict: if not self.proxy_list: return self.proxies with Pinterest._proxy_lock: proxy_url = self.proxy_list[Pinterest._shared_proxy_index % len(self.proxy_list)] Pinterest._shared_proxy_index += 1 return {"http": proxy_url, "https": proxy_url} # ----------------------------- # User-Agent rotation # ----------------------------- def _rotate_headers(self) -> dict: headers = self.BASE_HEADERS.copy() if not self._fixed_ua: profile = random.choice(USER_AGENTS) headers["User-Agent"] = profile["ua"] headers["Sec-Ch-Ua-Platform"] = profile["platform"] if profile["sec_ch_ua"]: headers["Sec-Ch-Ua"] = profile["sec_ch_ua"] else: headers.pop("Sec-Ch-Ua", None) return headers # ----------------------------- # Sleep with jitter # ----------------------------- def _jitter_sleep(self): if isinstance(self.sleep_time, (tuple, list)): min_s, max_s = self.sleep_time time.sleep(random.uniform(min_s, max_s)) elif self.sleep_time: time.sleep(self.sleep_time + random.uniform(0, 1.5)) # ----------------------------- # Request with exponential backoff # ----------------------------- def _request_with_backoff(self, url: str, headers: dict, proxies: dict): for attempt in range(self.max_retries): try: response = self.session.get(url, headers=headers, proxies=proxies, timeout=15) except Exception as e: if attempt < self.max_retries - 1: wait = (2 ** attempt) + random.uniform(1, 3) logging.warning(f"Request error: {e}, retrying in {wait:.1f}s ({attempt+1}/{self.max_retries})") time.sleep(wait) self.reset_session() headers = self._rotate_headers() continue raise if response.status_code in (429, 403): if attempt < self.max_retries - 1: wait = (2 ** attempt) + random.uniform(1, 5) logging.warning( f"Rate limited ({response.status_code}), backing off {wait:.1f}s " f"({attempt+1}/{self.max_retries})" ) time.sleep(wait) self.reset_session() headers = self._rotate_headers() continue return response return response return None # ----------------------------- # Pinterest API calls # ----------------------------- def search(self, query: str, page_size: int = 26) -> list: source_url = f"/search/pins/?q={quote(query)}&rs=typed" current_proxy = self._get_next_proxy() headers = self._rotate_headers() try: self._request_with_backoff( f"{self.BASE_URL}{source_url}", headers, current_proxy ) except Exception: pass payload = { "options": { "applied_unified_filters": None, "appliedProductFilters": "---", "article": None, "auto_correction_disabled": False, "corpus": None, "customized_rerank_type": None, "domains": None, "filters": None, "journey_depth": None, "page_size": f"{page_size}", "price_max": None, "price_min": None, "query_pin_sigs": None, "query": quote(query), "redux_normalize_feed": True, "request_params": None, "rs": "typed", "scope": "pins", "selected_one_bar_modules": None, "source_id": None, "source_module_id": None, "seoDrawerEnabled": False, "source_url": quote_plus(source_url), "top_pin_id": None, "top_pin_ids": None }, "context": {} } encoded = quote_plus(json.dumps(payload).replace(" ", "")) encoded = ( encoded.replace("%2520", "%20") .replace("%252F", "%2F") .replace("%253F", "%3F") .replace("%252520", "%2520") .replace("%253D", "%3D") .replace("%2526", "%26") ) epoch = int(time.time() * 1000) url = ( f"{self.BASE_URL}/resource/BaseSearchResource/get/" f"?source_url={quote_plus(source_url)}&data={encoded}&_={epoch}" ) headers = self._rotate_headers() headers["X-Pinterest-Source-Url"] = source_url try: response = self._request_with_backoff(url, headers, current_proxy) except Exception as e: logging.error(f"Request failed for '{query}': {e}") self.errors.append(str(e)) return [] if response is None: logging.error(f"All retries exhausted for '{query}'") return [] self._jitter_sleep() if response.status_code != 200: logging.warning(f"Image search failed for '{query}': {response.status_code}") return [] try: data = SearchResponse(**response.json()) except Exception as e: logging.error(f"Failed to parse response for '{query}': {e}") self.errors.append(str(e)) return [] if not data.resource_response.data: return [] results = data.resource_response.data.results output = [] for r in results: try: if not r.images or "orig" not in r.images: continue title = r.title or r.grid_title or r.seo_alt_text or query if isinstance(title, dict): title = title.get("text") or title.get("format") or query if not isinstance(title, str): title = query image_url = str(r.images["orig"].url) desc = r.description or "" if not isinstance(desc, str): desc = "" output.append({"title": title, "image_url": image_url, "description": desc}) except Exception: continue return output def search_descriptions(self, query: str, page_size: int = 26) -> list: source_url = f"/search/pins/?q={quote(query)}&rs=typed" current_proxy = self._get_next_proxy() headers = self._rotate_headers() try: self._request_with_backoff( f"{self.BASE_URL}{source_url}", headers, current_proxy ) except Exception: pass payload = { "options": { "applied_unified_filters": None, "appliedProductFilters": "---", "article": None, "auto_correction_disabled": False, "corpus": None, "customized_rerank_type": None, "domains": None, "filters": None, "journey_depth": None, "page_size": f"{page_size}", "price_max": None, "price_min": None, "query_pin_sigs": None, "query": quote(query), "redux_normalize_feed": True, "request_params": None, "rs": "typed", "scope": "pins", "selected_one_bar_modules": None, "source_id": None, "source_module_id": None, "seoDrawerEnabled": False, "source_url": quote_plus(source_url), "top_pin_id": None, "top_pin_ids": None }, "context": {} } encoded = quote_plus(json.dumps(payload).replace(" ", "")) encoded = ( encoded.replace("%2520", "%20") .replace("%252F", "%2F") .replace("%253F", "%3F") .replace("%252520", "%2520") .replace("%253D", "%3D") .replace("%2526", "%26") ) epoch = int(time.time() * 1000) url = ( f"{self.BASE_URL}/resource/BaseSearchResource/get/" f"?source_url={quote_plus(source_url)}&data={encoded}&_={epoch}" ) headers = self._rotate_headers() headers["X-Pinterest-Source-Url"] = source_url try: response = self._request_with_backoff(url, headers, current_proxy) except Exception as e: logging.error(f"Request failed for '{query}': {e}") self.errors.append(str(e)) return [] if response is None: logging.error(f"All retries exhausted for '{query}'") return [] self._jitter_sleep() if response.status_code != 200: logging.warning(f"Description search failed for '{query}': {response.status_code}") return [] try: data = SearchResponse(**response.json()) except Exception as e: logging.error(f"Failed to parse response for '{query}': {e}") self.errors.append(str(e)) return [] if not data.resource_response.data: return [] results = data.resource_response.data.results descriptions = [] for r in results: try: desc = r.description or "" if isinstance(desc, str) and desc.strip(): descriptions.append(desc.strip()) except Exception: continue return descriptions def fetch_related_keywords(keyword: str, proxy_list: list = None, max_results: int = 10, hl: str = "en", gl: str = "US") -> list: import re as _re import html as _html session = CurlSession(impersonate=random.choice(IMPERSONATE_PROFILES)) proxy = None if proxy_list: proxy = {"http": proxy_list[0], "https": proxy_list[0]} endpoint = "https://www.google.com/complete/search" params = { "cp": "1", "client": "gws-wiz", "xssi": "t", "gs_pcrt": "undefined", "hl": hl, "gl": gl, "authuser": "0", "dpr": "1", "q": keyword, } try: resp = session.get(endpoint, params=params, proxies=proxy, timeout=15) except Exception as e: logging.error(f"Related kw request failed for '{keyword}': {e}") try: session.close() except Exception: pass return [] finally: try: session.close() except Exception: pass if resp.status_code != 200: return [] raw = resp.text if raw.startswith(")]}'"): raw = raw[4:].lstrip("\n") try: top = json.loads(raw) except Exception: return [] if not isinstance(top, list) or len(top) == 0: return [] raw_suggestions = top[0] if isinstance(top[0], list) else [] results = [] for item in raw_suggestions: if not isinstance(item, list) or len(item) == 0: continue text, _ = item[0], item[1:] if not isinstance(text, str): continue text = _re.sub(r'<[^>]+>', '', text) text = _html.unescape(text) text = text.strip() if text: results.append(text) if len(results) >= max_results: break return results if __name__ == "__main__": keyword = "loki" p = Pinterest(sleep_time=(1.5, 3.0)) print(f"\nSearching for: {keyword}") results = p.search(keyword, 5) print(f"Found {len(results)} images") for r in results: print(f" - {r['title']}: {r['image_url'][:60]}...")