diff --git a/scripts/lib/http.py b/scripts/lib/http.py index 5d4898d..1b09d76 100644 --- a/scripts/lib/http.py +++ b/scripts/lib/http.py @@ -166,6 +166,14 @@ def post_raw(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, st return request("POST", url, headers=headers, json_data=json_data, raw=True, **kwargs) +def scrapecreators_headers(token: str) -> Dict[str, str]: + """Build ScrapeCreators request headers (x-api-key + JSON content type).""" + return { + "x-api-key": token, + "Content-Type": "application/json", + } + + def get_reddit_json(path: str, timeout: int = DEFAULT_TIMEOUT, retries: int = MAX_RETRIES) -> Dict[str, Any]: """Fetch Reddit thread JSON. diff --git a/scripts/lib/instagram.py b/scripts/lib/instagram.py index d378e14..8ae6357 100644 --- a/scripts/lib/instagram.py +++ b/scripts/lib/instagram.py @@ -112,14 +112,6 @@ def _log(msg: str): log.source_log("Instagram", msg) -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _parse_date(item: Dict[str, Any]) -> Optional[str]: """Parse date from ScrapeCreators Instagram item to YYYY-MM-DD. @@ -249,7 +241,7 @@ def _user_reels( from urllib.parse import urlencode params = urlencode({"handle": handle}) url = f"{reels_url}?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -260,7 +252,7 @@ def _user_reels( resp = _requests.get( reels_url, params={"handle": handle}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -307,7 +299,7 @@ def search_instagram( from urllib.parse import urlencode params = urlencode({"query": core_topic}) url = f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -318,7 +310,7 @@ def search_instagram( resp = _requests.get( f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search", params={"query": core_topic}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -403,7 +395,7 @@ def fetch_captions( resp = _requests.get( f"{SCRAPECREATORS_BASE}/v2/instagram/media/transcript", params={"url": url}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=15, ) if resp.status_code == 200: diff --git a/scripts/lib/pinterest.py b/scripts/lib/pinterest.py index ec4aad9..7d56d33 100644 --- a/scripts/lib/pinterest.py +++ b/scripts/lib/pinterest.py @@ -49,14 +49,6 @@ def _log(msg: str): log.source_log("Pinterest", msg) -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _parse_items(raw_items: List[Dict[str, Any]], core_topic: str) -> List[Dict[str, Any]]: """Parse raw Pinterest items into normalized dicts. @@ -154,7 +146,7 @@ def search_pinterest( from urllib.parse import urlencode params = urlencode({"keyword": core_topic}) url = f"{SCRAPECREATORS_BASE}/search?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -165,7 +157,7 @@ def search_pinterest( resp = _requests.get( f"{SCRAPECREATORS_BASE}/search", params={"keyword": core_topic}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() diff --git a/scripts/lib/reddit.py b/scripts/lib/reddit.py index b815dc0..66e1dc0 100644 --- a/scripts/lib/reddit.py +++ b/scripts/lib/reddit.py @@ -69,14 +69,6 @@ def _log(msg: str): log.source_log("Reddit", msg, tty_only=False) -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _extract_core_subject(topic: str) -> str: """Extract core subject from verbose query. @@ -335,7 +327,7 @@ def _global_search( try: data = http.get( f"{SCRAPECREATORS_BASE}/search", - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), params={"query": query, "sort": sort, "timeframe": timeframe}, timeout=30, retries=2, @@ -373,7 +365,7 @@ def _subreddit_search( try: data = http.get( f"{SCRAPECREATORS_BASE}/subreddit/search", - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), params={ "subreddit": subreddit, "query": query, @@ -405,7 +397,7 @@ def fetch_post_comments( try: data = http.get( f"{SCRAPECREATORS_BASE}/post/comments", - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), params={"url": url}, timeout=30, retries=2, diff --git a/scripts/lib/threads.py b/scripts/lib/threads.py index 61c6723..fbc2460 100644 --- a/scripts/lib/threads.py +++ b/scripts/lib/threads.py @@ -28,14 +28,6 @@ def _log(msg: str): log.source_log("Threads", msg) -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _extract_core_subject(topic: str) -> str: """Extract core subject from verbose query for Threads search.""" from .query import extract_core_subject @@ -170,7 +162,7 @@ def search_threads( from urllib.parse import urlencode params = urlencode({"keyword": core_topic}) url = f"{SCRAPECREATORS_BASE}/search?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -181,7 +173,7 @@ def search_threads( resp = _requests.get( f"{SCRAPECREATORS_BASE}/search", params={"keyword": core_topic}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() diff --git a/scripts/lib/tiktok.py b/scripts/lib/tiktok.py index 64faa21..1ff70fb 100644 --- a/scripts/lib/tiktok.py +++ b/scripts/lib/tiktok.py @@ -109,14 +109,6 @@ def _log(msg: str): log.source_log("TikTok", msg) -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _parse_date(item: Dict[str, Any]) -> Optional[str]: """Parse date from ScrapeCreators TikTok item to YYYY-MM-DD.""" ts = item.get("create_time") @@ -227,7 +219,7 @@ def _hashtag_search( from urllib.parse import urlencode params = urlencode({"hashtag": hashtag}) url = f"{SCRAPECREATORS_BASE}/search/hashtag?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -238,7 +230,7 @@ def _hashtag_search( resp = _requests.get( f"{SCRAPECREATORS_BASE}/search/hashtag", params={"hashtag": hashtag}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -274,7 +266,7 @@ def _profile_videos( from urllib.parse import urlencode params = urlencode({"handle": handle, "sort_by": "latest"}) url = f"{profile_url}?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -285,7 +277,7 @@ def _profile_videos( resp = _requests.get( profile_url, params={"handle": handle, "sort_by": "latest"}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -332,7 +324,7 @@ def search_tiktok( from urllib.parse import urlencode params = urlencode({"query": core_topic, "sort_by": "relevance"}) url = f"{SCRAPECREATORS_BASE}/search/keyword?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as e: @@ -343,7 +335,7 @@ def search_tiktok( resp = _requests.get( f"{SCRAPECREATORS_BASE}/search/keyword", params={"query": core_topic, "sort_by": "relevance"}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -433,7 +425,7 @@ def fetch_captions( resp = _requests.get( f"{SCRAPECREATORS_BASE}/video/transcript", params={"url": url}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=15, ) if resp.status_code == 200: diff --git a/scripts/lib/youtube_yt.py b/scripts/lib/youtube_yt.py index ac26a6e..0356c6c 100644 --- a/scripts/lib/youtube_yt.py +++ b/scripts/lib/youtube_yt.py @@ -655,14 +655,6 @@ except ImportError: _requests = None -def _sc_headers(token: str) -> Dict[str, str]: - """Build ScrapeCreators request headers.""" - return { - "x-api-key": token, - "Content-Type": "application/json", - } - - def _total_engagement(item: Dict[str, Any]) -> int: """Combined engagement score for ranking which videos to enrich.""" eng = item.get("engagement", {}) @@ -745,7 +737,7 @@ def _fetch_video_comments( from urllib.parse import urlencode params = urlencode({"id": video_id}) url = f"{SCRAPECREATORS_YT_BASE}/video/comments?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as exc: @@ -756,7 +748,7 @@ def _fetch_video_comments( resp = _requests.get( f"{SCRAPECREATORS_YT_BASE}/video/comments", params={"id": video_id}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -906,7 +898,7 @@ def _sc_youtube_search(keyword: str, token: str) -> List[Dict[str, Any]]: from urllib.parse import urlencode params = urlencode({"keyword": keyword}) url = f"{SCRAPECREATORS_YT_BASE}/search?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) return data.get("videos", data.get("data", data.get("items", []))) @@ -918,7 +910,7 @@ def _sc_youtube_search(keyword: str, token: str) -> List[Dict[str, Any]]: resp = _requests.get( f"{SCRAPECREATORS_YT_BASE}/search", params={"keyword": keyword}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) resp.raise_for_status() @@ -944,7 +936,7 @@ def _sc_fetch_transcript(video_id: str, token: str) -> Optional[str]: from urllib.parse import urlencode params = urlencode({"id": video_id}) url = f"{SCRAPECREATORS_YT_BASE}/video/transcript?{params}" - headers = _sc_headers(token) + headers = http.scrapecreators_headers(token) headers["User-Agent"] = http.USER_AGENT data = http.get(url, headers=headers, timeout=30, retries=2) except Exception as exc: @@ -955,7 +947,7 @@ def _sc_fetch_transcript(video_id: str, token: str) -> Optional[str]: resp = _requests.get( f"{SCRAPECREATORS_YT_BASE}/video/transcript", params={"id": video_id}, - headers=_sc_headers(token), + headers=http.scrapecreators_headers(token), timeout=30, ) if resp.status_code != 200: