| 1 | """LinkedIn post search via ScrapeCreators API. |
| 2 | |
| 3 | Searches public LinkedIn posts by keyword using the ScrapeCreators |
| 4 | /v1/linkedin/search/posts endpoint, which uses Google-indexed LinkedIn |
| 5 | content to bypass auth requirements. |
| 6 | |
| 7 | Requires SCRAPECREATORS_API_KEY environment variable. |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import re |
| 13 | from typing import Any, Dict, List |
| 14 | |
| 15 | from . import http, log |
| 16 | |
| 17 | SC_BASE = "https://api.scrapecreators.com/v1/linkedin" |
| 18 | |
| 19 | DEPTH_CONFIG: dict[str, dict[str, Any]] = { |
| 20 | "quick": {"date_posted": "last-week", "max_results": 10}, |
| 21 | "default": {"date_posted": "last-month", "max_results": 20}, |
| 22 | "deep": {"date_posted": "last-month", "max_results": 30}, |
| 23 | } |
| 24 | |
| 25 | |
| 26 | def _log(msg: str) -> None: |
| 27 | log.source_log("LinkedIn", msg, tty_only=False) |
| 28 | |
| 29 | |
| 30 | def search_linkedin( |
| 31 | topic: str, |
| 32 | from_date: str, |
| 33 | to_date: str, |
| 34 | depth: str = "default", |
| 35 | token: str = "", |
| 36 | ) -> Dict[str, Any]: |
| 37 | """Search LinkedIn posts via ScrapeCreators API. |
| 38 | |
| 39 | Args: |
| 40 | topic: Search query / topic string. |
| 41 | from_date: Window start date (YYYY-MM-DD) — used for depth mapping. |
| 42 | to_date: Window end date (YYYY-MM-DD). |
| 43 | depth: Retrieval profile — 'quick', 'default', or 'deep'. |
| 44 | token: ScrapeCreators API key. |
| 45 | |
| 46 | Returns: |
| 47 | Dict with a 'posts' list of raw post dicts. |
| 48 | """ |
| 49 | if not token: |
| 50 | _log("No SCRAPECREATORS_API_KEY — skipping") |
| 51 | return {"posts": []} |
| 52 | |
| 53 | cfg = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 54 | date_posted = cfg["date_posted"] |
| 55 | |
| 56 | _log(f"Searching for '{topic}' (date_posted={date_posted})") |
| 57 | |
| 58 | try: |
| 59 | response = http.get( |
| 60 | f"{SC_BASE}/search/posts", |
| 61 | params={"query": topic, "date_posted": date_posted}, |
| 62 | headers=http.scrapecreators_headers(token), |
| 63 | timeout=30, |
| 64 | retries=2, |
| 65 | ) |
| 66 | except http.HTTPError as exc: |
| 67 | _log(f"Search failed (HTTP {exc.status_code}): {exc}") |
| 68 | return {"posts": [], "error": str(exc)} |
| 69 | except Exception as exc: |
| 70 | _log(f"Search failed: {type(exc).__name__}: {exc}") |
| 71 | return {"posts": [], "error": str(exc)} |
| 72 | |
| 73 | posts = _extract_posts(response) |
| 74 | max_results = cfg["max_results"] |
| 75 | posts = posts[:max_results] |
| 76 | _log(f"Found {len(posts)} posts") |
| 77 | return {"posts": posts} |
| 78 | |
| 79 | |
| 80 | def _extract_posts(response: Any) -> List[Dict[str, Any]]: |
| 81 | """Extract the posts list from various possible response shapes.""" |
| 82 | if not isinstance(response, dict): |
| 83 | return [] |
| 84 | for key in ("posts", "items", "data", "results"): |
| 85 | val = response.get(key) |
| 86 | if isinstance(val, list): |
| 87 | return val |
| 88 | return [] |
| 89 | |
| 90 | |
| 91 | def _parse_date(raw: Any) -> str | None: |
| 92 | """Extract a YYYY-MM-DD string from various date formats.""" |
| 93 | if not raw: |
| 94 | return None |
| 95 | s = str(raw).strip() |
| 96 | m = re.search(r"(\d{4}-\d{2}-\d{2})", s) |
| 97 | if m: |
| 98 | return m.group(1) |
| 99 | return None |
| 100 | |
| 101 | |
| 102 | def _int_field(post: dict[str, Any], *keys: str) -> int: |
| 103 | """Return the first present integer field from a post dict.""" |
| 104 | for key in keys: |
| 105 | val = post.get(key) |
| 106 | if val is not None: |
| 107 | try: |
| 108 | return int(val) |
| 109 | except (TypeError, ValueError): |
| 110 | pass |
| 111 | return 0 |
| 112 | |
| 113 | |
| 114 | def _is_article(url: str) -> bool: |
| 115 | """A LinkedIn long-form article (Pulse) lives under a /pulse/ URL. |
| 116 | |
| 117 | Articles are higher-signal than ordinary posts — someone who wrote a |
| 118 | full article on a topic is a stronger source than someone who dashed off |
| 119 | a status update. |
| 120 | """ |
| 121 | return "/pulse/" in (url or "").lower() |
| 122 | |
| 123 | |
| 124 | # Relevance hints: articles outrank ordinary posts at rerank time. |
| 125 | _ARTICLE_RELEVANCE = 0.9 |
| 126 | _POST_RELEVANCE = 0.5 |
| 127 | |
| 128 | |
| 129 | def parse_linkedin_response( |
| 130 | result: Dict[str, Any], |
| 131 | from_date: str | None = None, |
| 132 | to_date: str | None = None, |
| 133 | ) -> List[Dict[str, Any]]: |
| 134 | """Parse ScrapeCreators LinkedIn response into engine-compatible item dicts. |
| 135 | |
| 136 | Each returned dict must be normalizable by normalize._normalize_linkedin. |
| 137 | |
| 138 | If from_date/to_date are given, applies the same hard date-range filter |
| 139 | used by instagram.search_and_enrich: drop items outside the window, but |
| 140 | fall back to keeping everything if the filter would otherwise empty the |
| 141 | result (SC doesn't always return a usable date per post). |
| 142 | """ |
| 143 | posts = result.get("posts") or [] |
| 144 | items: List[Dict[str, Any]] = [] |
| 145 | |
| 146 | for i, post in enumerate(posts): |
| 147 | if not isinstance(post, dict): |
| 148 | continue |
| 149 | |
| 150 | # The live ScrapeCreators post object carries the body in `description` |
| 151 | # and the timestamp in `datePublished`. The other keys are tolerated |
| 152 | # fallbacks for shape drift / alternate endpoints. |
| 153 | text = str( |
| 154 | post.get("description") |
| 155 | or post.get("text") |
| 156 | or post.get("content") |
| 157 | or post.get("body") |
| 158 | or "" |
| 159 | ).strip() |
| 160 | if not text: |
| 161 | continue |
| 162 | |
| 163 | author_raw = ( |
| 164 | post.get("author") |
| 165 | or post.get("authorName") |
| 166 | or post.get("author_name") |
| 167 | or "" |
| 168 | ) |
| 169 | author_url = "" |
| 170 | if isinstance(author_raw, dict): |
| 171 | author = str( |
| 172 | author_raw.get("name") or author_raw.get("full_name") or "" |
| 173 | ).strip() |
| 174 | author_url = str(author_raw.get("url") or author_raw.get("link") or "").strip() |
| 175 | else: |
| 176 | author = str(author_raw).strip() |
| 177 | |
| 178 | url = str( |
| 179 | post.get("url") or post.get("postUrl") or post.get("post_url") or "" |
| 180 | ).strip() |
| 181 | |
| 182 | post_id = str( |
| 183 | post.get("urn") or post.get("id") or post.get("postId") or f"LI{i + 1}" |
| 184 | ) |
| 185 | |
| 186 | date_raw = ( |
| 187 | post.get("datePublished") |
| 188 | or post.get("date") |
| 189 | or post.get("postedAt") |
| 190 | or post.get("posted_at") |
| 191 | or post.get("createdAt") |
| 192 | or post.get("created_at") |
| 193 | ) |
| 194 | date = _parse_date(date_raw) |
| 195 | |
| 196 | likes = _int_field(post, "likes", "likesCount", "likes_count", "numLikes", "likeCount") |
| 197 | comments = _int_field(post, "comments", "commentsCount", "comments_count", "numComments", "commentCount") |
| 198 | reposts = _int_field(post, "reposts", "repostsCount", "shares", "shareCount", "reshares") |
| 199 | |
| 200 | is_article = _is_article(url) |
| 201 | items.append({ |
| 202 | "id": post_id, |
| 203 | "text": text, |
| 204 | "url": url, |
| 205 | "author": author, |
| 206 | "author_url": author_url, |
| 207 | "date": date, |
| 208 | "engagement": { |
| 209 | "likes": likes, |
| 210 | "comments": comments, |
| 211 | "reposts": reposts, |
| 212 | }, |
| 213 | "relevance": _ARTICLE_RELEVANCE if is_article else _POST_RELEVANCE, |
| 214 | "is_article": is_article, |
| 215 | }) |
| 216 | |
| 217 | if from_date and to_date: |
| 218 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 219 | out_of_range = len(items) - len(in_range) |
| 220 | if in_range: |
| 221 | items = in_range |
| 222 | if out_of_range: |
| 223 | _log(f"Filtered {out_of_range} posts outside date range") |
| 224 | elif items: |
| 225 | _log(f"No posts within date range, keeping all {len(items)}") |
| 226 | |
| 227 | return items |
| 228 | |
| 229 | |
| 230 | # --- Article enrichment --------------------------------------------------- |
| 231 | # |
| 232 | # LinkedIn articles (Pulse long-form) never appear in /search/posts results — |
| 233 | # every search hit is a /posts/ status update. Articles live only on the |
| 234 | # author's profile, under `articles[]`. To honor "an article is high signal" |
| 235 | # we run a bounded enrichment lane: when a returned post's author name matches |
| 236 | # the topic (i.e. this is a person topic and we already hold their profile |
| 237 | # URL), make ONE profile call and surface their articles as high-signal items. |
| 238 | |
| 239 | |
| 240 | def _normalize_name(s: str) -> str: |
| 241 | """Lowercase, strip punctuation, collapse whitespace — for name matching.""" |
| 242 | return re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip() |
| 243 | |
| 244 | |
| 245 | def _token_run(needle: List[str], haystack: List[str]) -> bool: |
| 246 | """True if `needle` appears as a contiguous run of whole tokens in `haystack`. |
| 247 | |
| 248 | Token-level (not substring) so "ai" never matches inside "daisuke" — matching |
| 249 | is on word boundaries. Equality is the n == len(haystack) case. |
| 250 | """ |
| 251 | n = len(needle) |
| 252 | if n == 0 or n > len(haystack): |
| 253 | return False |
| 254 | return any(haystack[i : i + n] == needle for i in range(len(haystack) - n + 1)) |
| 255 | |
| 256 | |
| 257 | def _best_author_match(items: List[Dict[str, Any]], topic: str) -> str: |
| 258 | """Return the profile URL of the post author whose name matches the topic. |
| 259 | |
| 260 | Person-topic detection without a global predicate: when a returned post's |
| 261 | author has a multi-word name that the topic clearly refers to, treat the |
| 262 | topic as being about that person and return their profile URL. Matching is |
| 263 | on whole-token runs (the author's full name appears in the topic, or vice |
| 264 | versa), and the topic itself must be at least two tokens — so single-word |
| 265 | keyword topics ("AI", "Tesla") and short phrases never enrich, and a topic |
| 266 | token can't accidentally match inside an unrelated author's name. |
| 267 | """ |
| 268 | topic_tokens = _normalize_name(topic).split() |
| 269 | if len(topic_tokens) < 2: |
| 270 | return "" |
| 271 | for item in items: |
| 272 | name_tokens = _normalize_name(item.get("author", "")).split() |
| 273 | url = (item.get("author_url") or "").strip() |
| 274 | if not url or len(name_tokens) < 2: |
| 275 | continue |
| 276 | if _token_run(name_tokens, topic_tokens) or _token_run(topic_tokens, name_tokens): |
| 277 | return url |
| 278 | return "" |
| 279 | |
| 280 | |
| 281 | def search_profile(profile_url: str, token: str) -> Dict[str, Any]: |
| 282 | """Fetch a LinkedIn profile (incl. `articles[]`) via ScrapeCreators.""" |
| 283 | if not token or not profile_url: |
| 284 | return {} |
| 285 | try: |
| 286 | response = http.get( |
| 287 | f"{SC_BASE}/profile", |
| 288 | params={"url": profile_url}, |
| 289 | headers=http.scrapecreators_headers(token), |
| 290 | timeout=30, |
| 291 | retries=2, |
| 292 | ) |
| 293 | except http.HTTPError as exc: |
| 294 | _log(f"Profile fetch failed (HTTP {exc.status_code}): {exc}") |
| 295 | return {} |
| 296 | except Exception as exc: |
| 297 | _log(f"Profile fetch failed: {type(exc).__name__}: {exc}") |
| 298 | return {} |
| 299 | return response if isinstance(response, dict) else {} |
| 300 | |
| 301 | |
| 302 | def parse_profile_articles( |
| 303 | profile: Dict[str, Any], |
| 304 | from_date: str | None = None, |
| 305 | to_date: str | None = None, |
| 306 | ) -> List[Dict[str, Any]]: |
| 307 | """Map a profile's `articles[]` into high-signal engine item dicts.""" |
| 308 | articles = profile.get("articles") or [] |
| 309 | author = str(profile.get("name") or "").strip() |
| 310 | items: List[Dict[str, Any]] = [] |
| 311 | |
| 312 | for i, art in enumerate(articles): |
| 313 | if not isinstance(art, dict): |
| 314 | continue |
| 315 | headline = str(art.get("headline") or art.get("title") or "").strip() |
| 316 | if not headline: |
| 317 | continue |
| 318 | url = str(art.get("url") or art.get("link") or "").strip() |
| 319 | date = _parse_date(art.get("datePublished") or art.get("date")) |
| 320 | items.append({ |
| 321 | "id": str(art.get("id") or f"LIA{i + 1}"), |
| 322 | "text": headline, |
| 323 | "url": url, |
| 324 | "author": author, |
| 325 | "date": date, |
| 326 | "engagement": {}, |
| 327 | "relevance": _ARTICLE_RELEVANCE, |
| 328 | "is_article": True, |
| 329 | }) |
| 330 | |
| 331 | if from_date and to_date: |
| 332 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 333 | if in_range: |
| 334 | items = in_range |
| 335 | return items |
| 336 | |
| 337 | |
| 338 | def enrich_articles( |
| 339 | items: List[Dict[str, Any]], |
| 340 | topic: str, |
| 341 | token: str, |
| 342 | from_date: str | None = None, |
| 343 | to_date: str | None = None, |
| 344 | ) -> List[Dict[str, Any]]: |
| 345 | """Surface a person's LinkedIn articles as high-signal items. |
| 346 | |
| 347 | Bounded: fires only on person topics (a returned post author matches the |
| 348 | topic) and makes at most ONE profile API call. No-ops gracefully when |
| 349 | there's no match, no token, no profile, or no articles. |
| 350 | """ |
| 351 | if not token: |
| 352 | return [] |
| 353 | profile_url = _best_author_match(items, topic) |
| 354 | if not profile_url: |
| 355 | return [] |
| 356 | _log(f"Person topic — enriching articles from {profile_url}") |
| 357 | profile = search_profile(profile_url, token) |
| 358 | if not profile: |
| 359 | return [] |
| 360 | articles = parse_profile_articles(profile, from_date=from_date, to_date=to_date) |
| 361 | if articles: |
| 362 | _log(f"Found {len(articles)} article(s)") |
| 363 | return articles |
| 364 |