sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: consolidate _sc_headers into http.scrapecreators_headers (#209)

Browse Source 
Six source modules each defined an identical 8-line _sc_headers(token)
function returning {"x-api-key": token, "Content-Type": "application/json"}.
Moved it to http.scrapecreators_headers() and migrated all 33 call sites.

Affected files: reddit.py, threads.py, tiktok.py, instagram.py, pinterest.py,
youtube_yt.py. Zero per-source variation, zero behavior change.

Net: -40 lines. 1022 tests pass (15 pre-existing failures unchanged).
Live smoke test: reddit search returns 12 threads with full engagement.
This commit is contained in:
Ilia Alshanetsky
2026-04-14 07:43:56 -04:00
committed by GitHub
parent e395c1d57f
commit 9dd3f21476
7 changed files with 33 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/lib/http.py
+8
View File
@@ -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.
scripts/lib/instagram.py
+5 -13
View File
@@ -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:
scripts/lib/pinterest.py
+2 -10
View File
@@ -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()
scripts/lib/reddit.py
+3 -11
View File
@@ -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,
scripts/lib/threads.py
+2 -10
View File
@@ -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()
scripts/lib/tiktok.py
+7 -15
View File
@@ -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:
scripts/lib/youtube_yt.py
+6 -14
View File
@@ -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: