| 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 datetime |
| 13 | import hashlib |
| 14 | import re |
| 15 | import time |
| 16 | from typing import Any, Dict, List |
| 17 | |
| 18 | from . import http, log |
| 19 | |
| 20 | SC_BASE = "https://api.scrapecreators.com/v1/linkedin" |
| 21 | |
| 22 | DEPTH_CONFIG: dict[str, dict[str, Any]] = { |
| 23 | "quick": {"date_posted": "last-week", "max_results": 10}, |
| 24 | "default": {"date_posted": "last-month", "max_results": 20}, |
| 25 | "deep": {"date_posted": "last-month", "max_results": 30}, |
| 26 | } |
| 27 | |
| 28 | # The upstream endpoint only accepts these coarse buckets. Anything else is a |
| 29 | # hard 400, so an arbitrary day count has to be widened to the smallest bucket |
| 30 | # that still covers it; the caller's real from_date/to_date window is then |
| 31 | # enforced downstream by normalize.filter_by_date_range(). |
| 32 | DATE_POSTED_BUCKETS: tuple[tuple[int, str], ...] = ( |
| 33 | (1, "last-day"), |
| 34 | (7, "last-week"), |
| 35 | (31, "last-month"), |
| 36 | (366, "last-year"), |
| 37 | ) |
| 38 | |
| 39 | # This endpoint returns HTTP 404 with {"message": "No posts found"} for a query |
| 40 | # that simply matched nothing. That is an empty result, not a failure: treating |
| 41 | # it as an error marks a whole run partial and hides a legitimate null behind |
| 42 | # what looks like an outage. |
| 43 | EMPTY_RESULT_STATUS = 404 |
| 44 | |
| 45 | # Each cursor page returns ~10 posts; this bounds a runaway pagination loop. |
| 46 | MAX_PAGES = 10 |
| 47 | |
| 48 | # Consecutive all-duplicate pages tolerated before a bucket is abandoned. The |
| 49 | # covering bucket's first page routinely repeats the narrow bucket's top hits, |
| 50 | # so breaking on the first zero-add page would skip the older tail the second |
| 51 | # bucket exists to fetch. |
| 52 | MAX_EMPTY_ADD_PAGES = 3 |
| 53 | |
| 54 | # Whole-call wall-clock budget. Pagination bounds the request COUNT, not time: |
| 55 | # 2 buckets x MAX_PAGES pages x (30s timeout + retry) is ~10 minutes against a |
| 56 | # slow-but-alive endpoint, with no error to attribute the stall to. Each request |
| 57 | # is additionally clamped to the remaining budget so the bound actually holds. |
| 58 | SEARCH_BUDGET_SECONDS = 120.0 |
| 59 | PAGE_TIMEOUT = 30.0 |
| 60 | MIN_PAGE_TIMEOUT = 5.0 |
| 61 | |
| 62 | # Endpoint-scoped failures. These are properties of the credential or the |
| 63 | # account, not of one bucket, so retrying the next bucket only wastes a call |
| 64 | # and — for 429 — deepens the rate limit already being hit. |
| 65 | FATAL_STATUS_CODES = frozenset({401, 403, 429}) |
| 66 | |
| 67 | |
| 68 | def _log(msg: str) -> None: |
| 69 | log.source_log("LinkedIn", msg, tty_only=False) |
| 70 | |
| 71 | |
| 72 | def _coerce_date(raw: Any) -> datetime.date | None: |
| 73 | """Parse a YYYY-MM-DD date, tolerating a full ISO timestamp.""" |
| 74 | try: |
| 75 | return datetime.date.fromisoformat(str(raw)[:10]) |
| 76 | except (TypeError, ValueError): |
| 77 | return None |
| 78 | |
| 79 | |
| 80 | def _buckets_for_window( |
| 81 | from_date: str, |
| 82 | to_date: str, |
| 83 | fallback: str, |
| 84 | today: datetime.date | None = None, |
| 85 | ) -> List[str]: |
| 86 | """Buckets to query for a window, narrowest first. |
| 87 | |
| 88 | Two properties of the upstream endpoint drive this: |
| 89 | |
| 90 | 1. Buckets are relative to NOW, not to the requested window. So the bucket |
| 91 | has to be sized by how OLD the requested window is (today - from_date), |
| 92 | never by its span. Sizing by span would send a 31-day window from two |
| 93 | years ago to 'last-month', which cannot contain a single matching post. |
| 94 | 2. Results inside a bucket are relevance-ranked, not recency-ranked, so a |
| 95 | widened bucket does NOT contain the narrower one's yield: 'last-year' |
| 96 | for a 90-day window returned 30 posts spread across a full year, only 4 |
| 97 | inside the window. So query the narrow bucket for dense recent coverage |
| 98 | AND the covering bucket for the older tail, then union them. |
| 99 | |
| 100 | The caller's exact window is still enforced downstream by |
| 101 | filter_by_date_range(); these buckets only have to COVER it. |
| 102 | """ |
| 103 | start = _coerce_date(from_date) |
| 104 | end = _coerce_date(to_date) |
| 105 | if start is None or end is None: |
| 106 | _log(f"Unparseable window ({from_date!r}..{to_date!r}), using {fallback}") |
| 107 | return [fallback] |
| 108 | if end < start: |
| 109 | _log(f"Inverted window ({from_date}..{to_date}), using {fallback}") |
| 110 | return [fallback] |
| 111 | |
| 112 | today = today or datetime.date.today() |
| 113 | # Age of the OLDEST requested post, which is what the bucket must reach |
| 114 | # back to. +1 because the window is inclusive of from_date. |
| 115 | age_days = (today - start).days + 1 |
| 116 | if age_days < 0: |
| 117 | _log(f"Window starts in the future ({from_date}), using {fallback}") |
| 118 | return [fallback] |
| 119 | |
| 120 | widest_limit, widest_bucket = DATE_POSTED_BUCKETS[-1] |
| 121 | covering = widest_bucket |
| 122 | for limit, bucket in DATE_POSTED_BUCKETS: |
| 123 | if age_days <= limit: |
| 124 | covering = bucket |
| 125 | break |
| 126 | else: |
| 127 | _log( |
| 128 | f"Window reaches back {age_days}d but the widest bucket is " |
| 129 | f"{widest_bucket} ({widest_limit}d) — older results are unreachable" |
| 130 | ) |
| 131 | |
| 132 | buckets = [covering] |
| 133 | # Backfill the dense recent block that the wider bucket skips over. |
| 134 | for _, bucket in DATE_POSTED_BUCKETS: |
| 135 | if bucket == covering: |
| 136 | break |
| 137 | if bucket not in buckets: |
| 138 | buckets.insert(-1, bucket) |
| 139 | # Only the immediately-narrower bucket is worth the extra calls. |
| 140 | return buckets[-2:] |
| 141 | |
| 142 | |
| 143 | def _dedupe_key(post: Dict[str, Any]) -> str: |
| 144 | """Stable identity for a post. |
| 145 | |
| 146 | Falls back to a content fingerprint when no identifier is present. Without |
| 147 | it, keyless posts are never deduped, so the union returns them once per |
| 148 | bucket AND a repeating cursor never trips the all-duplicates guard. |
| 149 | """ |
| 150 | for field in ("url", "postUrl", "post_url", "urn", "id", "postId"): |
| 151 | val = post.get(field) |
| 152 | if val: |
| 153 | return str(val) |
| 154 | body = str(post.get("description") or post.get("text") or "") |
| 155 | author = str(post.get("author") or "") |
| 156 | if not body and not author: |
| 157 | return "" |
| 158 | return "sha1:" + hashlib.sha1(f"{author}\x00{body}".encode()).hexdigest() |
| 159 | |
| 160 | |
| 161 | def search_linkedin( |
| 162 | topic: str, |
| 163 | from_date: str, |
| 164 | to_date: str, |
| 165 | depth: str = "default", |
| 166 | token: str = "", |
| 167 | ) -> Dict[str, Any]: |
| 168 | """Search LinkedIn posts via ScrapeCreators API. |
| 169 | |
| 170 | Args: |
| 171 | topic: Search query / topic string. |
| 172 | from_date: Window start date (YYYY-MM-DD) — sets the date_posted bucket. |
| 173 | to_date: Window end date (YYYY-MM-DD). |
| 174 | depth: Retrieval profile — 'quick', 'default', or 'deep'. |
| 175 | token: ScrapeCreators API key. |
| 176 | |
| 177 | Returns: |
| 178 | Dict with a 'posts' list of raw post dicts. |
| 179 | """ |
| 180 | if not token: |
| 181 | _log("No SCRAPECREATORS_API_KEY — skipping") |
| 182 | return {"posts": []} |
| 183 | |
| 184 | cfg = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 185 | buckets = _buckets_for_window(from_date, to_date, cfg["date_posted"]) |
| 186 | max_results = cfg["max_results"] |
| 187 | query = topic.strip() |
| 188 | if not query: |
| 189 | _log("Empty query — skipping") |
| 190 | return {"posts": []} |
| 191 | |
| 192 | # max_results bounds the RETURNED collection, not each retrieval lane, so |
| 193 | # split it across buckets rather than granting each the full budget. |
| 194 | quota = max(1, -(-max_results // len(buckets))) |
| 195 | deadline = time.monotonic() + SEARCH_BUDGET_SECONDS |
| 196 | |
| 197 | _log( |
| 198 | f"Searching for '{query}' (date_posted={','.join(buckets)}, " |
| 199 | f"max_results={max_results}, quota/window={quota})" |
| 200 | ) |
| 201 | |
| 202 | posts: List[Dict[str, Any]] = [] |
| 203 | seen: set[str] = set() |
| 204 | errors: List[str] = [] |
| 205 | fatal = False |
| 206 | |
| 207 | for date_posted in buckets: |
| 208 | if fatal or len(posts) >= max_results: |
| 209 | break |
| 210 | |
| 211 | bucket_count = 0 |
| 212 | pages = 0 |
| 213 | empty_adds = 0 |
| 214 | cursor: Any = None |
| 215 | prev_cursor: Any = None |
| 216 | stop = "pages exhausted" |
| 217 | |
| 218 | for _ in range(MAX_PAGES): |
| 219 | remaining = deadline - time.monotonic() |
| 220 | if remaining <= 0: |
| 221 | stop = "wall-clock budget exceeded" |
| 222 | errors.append(f"{date_posted}: search budget of {SEARCH_BUDGET_SECONDS}s exceeded") |
| 223 | break |
| 224 | |
| 225 | # Clamp the request to what is left of the budget, and drop the |
| 226 | # retry when there is not room for one. Checking the deadline only |
| 227 | # BETWEEN requests does not bound the call: a request starting a |
| 228 | # second before the deadline could still run its full |
| 229 | # timeout-times-retries budget past it. |
| 230 | page_timeout = max(MIN_PAGE_TIMEOUT, min(PAGE_TIMEOUT, remaining)) |
| 231 | page_retries = 2 if remaining > (page_timeout * 2 + 2) else 1 |
| 232 | |
| 233 | params: Dict[str, Any] = {"query": query, "date_posted": date_posted} |
| 234 | if cursor: |
| 235 | params["cursor"] = cursor |
| 236 | try: |
| 237 | response = http.get( |
| 238 | f"{SC_BASE}/search/posts", |
| 239 | params=params, |
| 240 | headers=http.scrapecreators_headers(token), |
| 241 | timeout=page_timeout, |
| 242 | retries=page_retries, |
| 243 | ) |
| 244 | except http.HTTPError as exc: |
| 245 | if exc.status_code == EMPTY_RESULT_STATUS: |
| 246 | # "No posts found" — a legitimate empty result, not a |
| 247 | # failure. Recording it as an error would mark the whole |
| 248 | # run partial and hide a real null behind a fake outage. |
| 249 | stop = "no posts found" |
| 250 | break |
| 251 | _log(f"Search failed (HTTP {exc.status_code}, {date_posted}): {exc}") |
| 252 | errors.append(f"{date_posted}: {exc}") |
| 253 | stop = f"HTTP {exc.status_code}" |
| 254 | if exc.status_code in FATAL_STATUS_CODES: |
| 255 | # Credential- or account-scoped: the next bucket would fail |
| 256 | # identically, and retrying a 429 deepens the rate limit. |
| 257 | fatal = True |
| 258 | break |
| 259 | except Exception as exc: |
| 260 | _log(f"Search failed ({date_posted}): {type(exc).__name__}: {exc}") |
| 261 | errors.append(f"{date_posted}: {type(exc).__name__}: {exc}") |
| 262 | stop = type(exc).__name__ |
| 263 | fatal = True |
| 264 | break |
| 265 | |
| 266 | pages += 1 |
| 267 | page = _extract_posts(response) |
| 268 | if not page: |
| 269 | stop = "empty page" |
| 270 | break |
| 271 | |
| 272 | added = 0 |
| 273 | for post in page: |
| 274 | if len(posts) >= max_results: |
| 275 | break |
| 276 | key = _dedupe_key(post) |
| 277 | if key and key in seen: |
| 278 | continue |
| 279 | if key: |
| 280 | seen.add(key) |
| 281 | posts.append(post) |
| 282 | added += 1 |
| 283 | bucket_count += 1 |
| 284 | |
| 285 | if len(posts) >= max_results: |
| 286 | stop = "max_results reached" |
| 287 | break |
| 288 | if bucket_count >= quota: |
| 289 | stop = "window quota reached" |
| 290 | break |
| 291 | |
| 292 | # An all-duplicate page is expected where buckets overlap; only a |
| 293 | # run of them means the cursor has stopped yielding anything new. |
| 294 | empty_adds = empty_adds + 1 if added == 0 else 0 |
| 295 | if empty_adds >= MAX_EMPTY_ADD_PAGES: |
| 296 | stop = f"{empty_adds} consecutive all-duplicate pages" |
| 297 | break |
| 298 | |
| 299 | raw_cursor = response.get("cursor") if isinstance(response, dict) else None |
| 300 | cursor = raw_cursor if isinstance(raw_cursor, (str, int)) else None |
| 301 | if not cursor: |
| 302 | stop = "no cursor" |
| 303 | break |
| 304 | if cursor == prev_cursor: |
| 305 | stop = "cursor stopped advancing" |
| 306 | break |
| 307 | prev_cursor = cursor |
| 308 | |
| 309 | _log(f" {date_posted}: {bucket_count} posts over {pages} page(s), stopped: {stop}") |
| 310 | |
| 311 | if not posts and errors: |
| 312 | return {"posts": [], "error": errors[0]} |
| 313 | |
| 314 | result: Dict[str, Any] = {"posts": posts} |
| 315 | if errors: |
| 316 | # A window that failed after another succeeded must not read as a |
| 317 | # complete result — downstream would treat thin coverage as a finding |
| 318 | # about LinkedIn rather than about the request that never landed. |
| 319 | result["partial"] = True |
| 320 | result["error"] = errors[0] |
| 321 | _log(f"PARTIAL — {len(errors)} window/page failure(s); first: {errors[0]}") |
| 322 | |
| 323 | _log(f"Found {len(posts)} posts across {len(buckets)} window(s)") |
| 324 | return result |
| 325 | |
| 326 | |
| 327 | def _extract_posts(response: Any) -> List[Dict[str, Any]]: |
| 328 | """Extract the posts list from various possible response shapes.""" |
| 329 | if not isinstance(response, dict): |
| 330 | return [] |
| 331 | for key in ("posts", "items", "data", "results"): |
| 332 | val = response.get(key) |
| 333 | if isinstance(val, list): |
| 334 | return val |
| 335 | return [] |
| 336 | |
| 337 | |
| 338 | def _parse_date(raw: Any) -> str | None: |
| 339 | """Extract a YYYY-MM-DD string from various date formats.""" |
| 340 | if not raw: |
| 341 | return None |
| 342 | s = str(raw).strip() |
| 343 | m = re.search(r"(\d{4}-\d{2}-\d{2})", s) |
| 344 | if m: |
| 345 | return m.group(1) |
| 346 | return None |
| 347 | |
| 348 | |
| 349 | def _int_field(post: dict[str, Any], *keys: str) -> int: |
| 350 | """Return the first present integer field from a post dict.""" |
| 351 | for key in keys: |
| 352 | val = post.get(key) |
| 353 | if val is not None: |
| 354 | try: |
| 355 | return int(val) |
| 356 | except (TypeError, ValueError): |
| 357 | pass |
| 358 | return 0 |
| 359 | |
| 360 | |
| 361 | def _is_article(url: str) -> bool: |
| 362 | """A LinkedIn long-form article (Pulse) lives under a /pulse/ URL. |
| 363 | |
| 364 | Articles are higher-signal than ordinary posts — someone who wrote a |
| 365 | full article on a topic is a stronger source than someone who dashed off |
| 366 | a status update. |
| 367 | """ |
| 368 | return "/pulse/" in (url or "").lower() |
| 369 | |
| 370 | |
| 371 | # Relevance hints: articles outrank ordinary posts at rerank time. |
| 372 | _ARTICLE_RELEVANCE = 0.9 |
| 373 | _POST_RELEVANCE = 0.5 |
| 374 | |
| 375 | |
| 376 | def parse_linkedin_response( |
| 377 | result: Dict[str, Any], |
| 378 | from_date: str | None = None, |
| 379 | to_date: str | None = None, |
| 380 | ) -> List[Dict[str, Any]]: |
| 381 | """Parse ScrapeCreators LinkedIn response into engine-compatible item dicts. |
| 382 | |
| 383 | Each returned dict must be normalizable by normalize._normalize_linkedin. |
| 384 | |
| 385 | If from_date/to_date are given, applies the same hard date-range filter |
| 386 | used by instagram.search_and_enrich: drop items outside the window, but |
| 387 | fall back to keeping everything if the filter would otherwise empty the |
| 388 | result (SC doesn't always return a usable date per post). |
| 389 | """ |
| 390 | posts = result.get("posts") or [] |
| 391 | items: List[Dict[str, Any]] = [] |
| 392 | |
| 393 | for i, post in enumerate(posts): |
| 394 | if not isinstance(post, dict): |
| 395 | continue |
| 396 | |
| 397 | # The live ScrapeCreators post object carries the body in `description` |
| 398 | # and the timestamp in `datePublished`. The other keys are tolerated |
| 399 | # fallbacks for shape drift / alternate endpoints. |
| 400 | text = str( |
| 401 | post.get("description") |
| 402 | or post.get("text") |
| 403 | or post.get("content") |
| 404 | or post.get("body") |
| 405 | or "" |
| 406 | ).strip() |
| 407 | if not text: |
| 408 | continue |
| 409 | |
| 410 | author_raw = ( |
| 411 | post.get("author") |
| 412 | or post.get("authorName") |
| 413 | or post.get("author_name") |
| 414 | or "" |
| 415 | ) |
| 416 | author_url = "" |
| 417 | if isinstance(author_raw, dict): |
| 418 | author = str( |
| 419 | author_raw.get("name") or author_raw.get("full_name") or "" |
| 420 | ).strip() |
| 421 | author_url = str(author_raw.get("url") or author_raw.get("link") or "").strip() |
| 422 | else: |
| 423 | author = str(author_raw).strip() |
| 424 | |
| 425 | url = str( |
| 426 | post.get("url") or post.get("postUrl") or post.get("post_url") or "" |
| 427 | ).strip() |
| 428 | |
| 429 | post_id = str( |
| 430 | post.get("urn") or post.get("id") or post.get("postId") or f"LI{i + 1}" |
| 431 | ) |
| 432 | |
| 433 | date_raw = ( |
| 434 | post.get("datePublished") |
| 435 | or post.get("date") |
| 436 | or post.get("postedAt") |
| 437 | or post.get("posted_at") |
| 438 | or post.get("createdAt") |
| 439 | or post.get("created_at") |
| 440 | ) |
| 441 | date = _parse_date(date_raw) |
| 442 | |
| 443 | likes = _int_field(post, "likes", "likesCount", "likes_count", "numLikes", "likeCount") |
| 444 | comments = _int_field(post, "comments", "commentsCount", "comments_count", "numComments", "commentCount") |
| 445 | reposts = _int_field(post, "reposts", "repostsCount", "shares", "shareCount", "reshares") |
| 446 | |
| 447 | is_article = _is_article(url) |
| 448 | items.append({ |
| 449 | "id": post_id, |
| 450 | "text": text, |
| 451 | "url": url, |
| 452 | "author": author, |
| 453 | "author_url": author_url, |
| 454 | "date": date, |
| 455 | "engagement": { |
| 456 | "likes": likes, |
| 457 | "comments": comments, |
| 458 | "reposts": reposts, |
| 459 | }, |
| 460 | "relevance": _ARTICLE_RELEVANCE if is_article else _POST_RELEVANCE, |
| 461 | "is_article": is_article, |
| 462 | }) |
| 463 | |
| 464 | if from_date and to_date: |
| 465 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 466 | out_of_range = len(items) - len(in_range) |
| 467 | if in_range: |
| 468 | items = in_range |
| 469 | if out_of_range: |
| 470 | _log(f"Filtered {out_of_range} posts outside date range") |
| 471 | elif items: |
| 472 | _log(f"No posts within date range, keeping all {len(items)}") |
| 473 | |
| 474 | return items |
| 475 | |
| 476 | |
| 477 | # --- Article enrichment --------------------------------------------------- |
| 478 | # |
| 479 | # LinkedIn articles (Pulse long-form) never appear in /search/posts results — |
| 480 | # every search hit is a /posts/ status update. Articles live only on the |
| 481 | # author's profile, under `articles[]`. To honor "an article is high signal" |
| 482 | # we run a bounded enrichment lane: when a returned post's author name matches |
| 483 | # the topic (i.e. this is a person topic and we already hold their profile |
| 484 | # URL), make ONE profile call and surface their articles as high-signal items. |
| 485 | |
| 486 | |
| 487 | def _normalize_name(s: str) -> str: |
| 488 | """Lowercase, strip punctuation, collapse whitespace — for name matching.""" |
| 489 | return re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip() |
| 490 | |
| 491 | |
| 492 | def _token_run(needle: List[str], haystack: List[str]) -> bool: |
| 493 | """True if `needle` appears as a contiguous run of whole tokens in `haystack`. |
| 494 | |
| 495 | Token-level (not substring) so "ai" never matches inside "daisuke" — matching |
| 496 | is on word boundaries. Equality is the n == len(haystack) case. |
| 497 | """ |
| 498 | n = len(needle) |
| 499 | if n == 0 or n > len(haystack): |
| 500 | return False |
| 501 | return any(haystack[i : i + n] == needle for i in range(len(haystack) - n + 1)) |
| 502 | |
| 503 | |
| 504 | def _best_author_match(items: List[Dict[str, Any]], topic: str) -> str: |
| 505 | """Return the profile URL of the post author whose name matches the topic. |
| 506 | |
| 507 | Person-topic detection without a global predicate: when a returned post's |
| 508 | author has a multi-word name that the topic clearly refers to, treat the |
| 509 | topic as being about that person and return their profile URL. Matching is |
| 510 | on whole-token runs (the author's full name appears in the topic, or vice |
| 511 | versa), and the topic itself must be at least two tokens — so single-word |
| 512 | keyword topics ("AI", "Tesla") and short phrases never enrich, and a topic |
| 513 | token can't accidentally match inside an unrelated author's name. |
| 514 | """ |
| 515 | topic_tokens = _normalize_name(topic).split() |
| 516 | if len(topic_tokens) < 2: |
| 517 | return "" |
| 518 | for item in items: |
| 519 | name_tokens = _normalize_name(item.get("author", "")).split() |
| 520 | url = (item.get("author_url") or "").strip() |
| 521 | if not url or len(name_tokens) < 2: |
| 522 | continue |
| 523 | if _token_run(name_tokens, topic_tokens) or _token_run(topic_tokens, name_tokens): |
| 524 | return url |
| 525 | return "" |
| 526 | |
| 527 | |
| 528 | def search_profile(profile_url: str, token: str) -> Dict[str, Any]: |
| 529 | """Fetch a LinkedIn profile (incl. `articles[]`) via ScrapeCreators.""" |
| 530 | if not token or not profile_url: |
| 531 | return {} |
| 532 | try: |
| 533 | response = http.get( |
| 534 | f"{SC_BASE}/profile", |
| 535 | params={"url": profile_url}, |
| 536 | headers=http.scrapecreators_headers(token), |
| 537 | timeout=30, |
| 538 | retries=2, |
| 539 | ) |
| 540 | except http.HTTPError as exc: |
| 541 | _log(f"Profile fetch failed (HTTP {exc.status_code}): {exc}") |
| 542 | return {} |
| 543 | except Exception as exc: |
| 544 | _log(f"Profile fetch failed: {type(exc).__name__}: {exc}") |
| 545 | return {} |
| 546 | return response if isinstance(response, dict) else {} |
| 547 | |
| 548 | |
| 549 | def parse_profile_articles( |
| 550 | profile: Dict[str, Any], |
| 551 | from_date: str | None = None, |
| 552 | to_date: str | None = None, |
| 553 | ) -> List[Dict[str, Any]]: |
| 554 | """Map a profile's `articles[]` into high-signal engine item dicts.""" |
| 555 | articles = profile.get("articles") or [] |
| 556 | author = str(profile.get("name") or "").strip() |
| 557 | items: List[Dict[str, Any]] = [] |
| 558 | |
| 559 | for i, art in enumerate(articles): |
| 560 | if not isinstance(art, dict): |
| 561 | continue |
| 562 | headline = str(art.get("headline") or art.get("title") or "").strip() |
| 563 | if not headline: |
| 564 | continue |
| 565 | url = str(art.get("url") or art.get("link") or "").strip() |
| 566 | date = _parse_date(art.get("datePublished") or art.get("date")) |
| 567 | items.append({ |
| 568 | "id": str(art.get("id") or f"LIA{i + 1}"), |
| 569 | "text": headline, |
| 570 | "url": url, |
| 571 | "author": author, |
| 572 | "date": date, |
| 573 | "engagement": {}, |
| 574 | "relevance": _ARTICLE_RELEVANCE, |
| 575 | "is_article": True, |
| 576 | }) |
| 577 | |
| 578 | if from_date and to_date: |
| 579 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 580 | if in_range: |
| 581 | items = in_range |
| 582 | return items |
| 583 | |
| 584 | |
| 585 | def enrich_articles( |
| 586 | items: List[Dict[str, Any]], |
| 587 | topic: str, |
| 588 | token: str, |
| 589 | from_date: str | None = None, |
| 590 | to_date: str | None = None, |
| 591 | ) -> List[Dict[str, Any]]: |
| 592 | """Surface a person's LinkedIn articles as high-signal items. |
| 593 | |
| 594 | Bounded: fires only on person topics (a returned post author matches the |
| 595 | topic) and makes at most ONE profile API call. No-ops gracefully when |
| 596 | there's no match, no token, no profile, or no articles. |
| 597 | """ |
| 598 | if not token: |
| 599 | return [] |
| 600 | profile_url = _best_author_match(items, topic) |
| 601 | if not profile_url: |
| 602 | return [] |
| 603 | _log(f"Person topic — enriching articles from {profile_url}") |
| 604 | profile = search_profile(profile_url, token) |
| 605 | if not profile: |
| 606 | return [] |
| 607 | articles = parse_profile_articles(profile, from_date=from_date, to_date=to_date) |
| 608 | if articles: |
| 609 | _log(f"Found {len(articles)} article(s)") |
| 610 | return articles |
| 611 |