返回 last30days-skill
normalize.py
根目录 / skills / last30days / scripts / lib / normalize.py
1 """Normalization of source-specific payloads into the v3 generic item model."""
2
3 from __future__ import annotations
4
5 from typing import Any
6 from urllib.parse import urlparse
7
8 from . import dates, schema
9
10
11 def filter_by_date_range(
12 items: list[schema.SourceItem],
13 from_date: str,
14 to_date: str,
15 require_date: bool = False,
16 ) -> list[schema.SourceItem]:
17 """Keep only items within the requested window."""
18 filtered: list[schema.SourceItem] = []
19 for item in items:
20 if not item.published_at:
21 if not require_date:
22 filtered.append(item)
23 continue
24 if item.published_at < from_date or item.published_at > to_date:
25 continue
26 filtered.append(item)
27 return filtered
28
29
30 def normalize_source_items(
31 source: str,
32 items: list[dict[str, Any]],
33 from_date: str,
34 to_date: str,
35 freshness_mode: str = "balanced_recent",
36 ) -> list[schema.SourceItem]:
37 """Normalize raw source items, filter by date range, with evergreen fallback for how_to queries."""
38 source = source.lower()
39 normalizers = {
40 "reddit": _normalize_reddit,
41 "x": _normalize_x,
42 "youtube": _normalize_youtube,
43 "tiktok": lambda s, i, idx, fd, td: _normalize_shortform_video(
44 s, i, idx, fd, td, "TK", "TikTok post"
45 ),
46 "instagram": lambda s, i, idx, fd, td: _normalize_shortform_video(
47 s, i, idx, fd, td, "IG", "Instagram reel"
48 ),
49 "hackernews": _normalize_hackernews,
50 "stocktwits": _normalize_stocktwits,
51 "dripstack": _normalize_dripstack,
52 "bluesky": lambda s, i, idx, fd, td: _normalize_microblog(
53 s, i, idx, fd, td, "BS", "Bluesky post"
54 ),
55 "truthsocial": lambda s, i, idx, fd, td: _normalize_microblog(
56 s, i, idx, fd, td, "TS", "Truth Social post"
57 ),
58 "threads": lambda s, i, idx, fd, td: _normalize_microblog(
59 s, i, idx, fd, td, "TH", "Threads post"
60 ),
61 "telegram": lambda s, i, idx, fd, td: _normalize_microblog(
62 s, i, idx, fd, td, "TG", "Telegram post"
63 ),
64 "xquik": _normalize_x,
65 "pinterest": _normalize_pinterest,
66 "polymarket": _normalize_polymarket,
67 "digg": _normalize_digg,
68 "arxiv": _normalize_arxiv,
69 "techmeme": _normalize_techmeme,
70 "trustpilot": _normalize_trustpilot,
71 "amazon": _normalize_amazon,
72 "meta_ads": _normalize_meta_ads,
73 "grounding": _normalize_grounding,
74 "xiaohongshu": _normalize_grounding,
75 "github": _normalize_github,
76 "perplexity": _normalize_grounding,
77 "jobs": _normalize_jobs,
78 "linkedin": _normalize_linkedin,
79 }
80 normalizer = normalizers.get(source)
81 if normalizer is None:
82 raise ValueError(f"Unsupported source: {source}")
83 normalized = [
84 normalizer(source, item, index, from_date, to_date)
85 for index, item in enumerate(items)
86 ]
87 if source == "arxiv":
88 # The adapter owns arXiv's 365-day recency contract. Applying the
89 # report window again here drops relevant papers the adapter accepted.
90 return normalized
91 if source == "jobs":
92 # A careers board is a snapshot of CURRENTLY OPEN roles. An open posting
93 # is current evidence regardless of when it was posted, so date-windowing
94 # it drops still-open roles (the "Founding Research Scientist, Human
95 # Simulation" miss: 26 open roles filtered to 3 by a 30-day window).
96 # Keep the full board; recency is annotated, not used to drop.
97 return normalized
98 require_date = source == "grounding"
99 filtered = filter_by_date_range(
100 normalized, from_date, to_date, require_date=require_date
101 )
102 if filtered:
103 return filtered
104 # YouTube search already keeps out-of-window videos when fewer than 3
105 # are recent, then pays for transcripts. A second hard date filter here
106 # dropped those transcribed items to zero (#1043). Keep transcript-backed
107 # retrieved items instead of paying for transcripts that never appear in
108 # the brief. Metadata-only / caption-failed videos are not that rescue.
109 if source == "youtube" and normalized:
110 transcribed = [
111 item for item in normalized if str(item.snippet or "").strip()
112 ]
113 if transcribed:
114 if require_date:
115 dated = [item for item in transcribed if item.published_at]
116 return dated or transcribed
117 return transcribed
118 if freshness_mode == "evergreen_ok" and source == "youtube":
119 if require_date:
120 return [item for item in normalized if item.published_at]
121 return normalized
122 return filtered
123
124
125 def _remap_comments(
126 raw: list[Any],
127 score_keys: tuple[str, ...],
128 excerpt_keys: tuple[str, ...],
129 *,
130 preserve_absent_score: bool = False,
131 ) -> list[dict[str, Any]]:
132 """Normalize comments from any source into the shared Reddit-compatible shape.
133
134 Downstream code (signals._top_comment_score, render._top_comments_list,
135 entity_extract, rerank) all expect `score` and `excerpt`. This helper maps
136 per-source field names (YT: likes/text, TikTok: digg_count/text) onto that
137 shape while preserving author/date/url passthrough.
138
139 Sources that distinguish an absent vote from a measured zero can opt into
140 preserving the absent value as ``None``.
141 """
142 out: list[dict[str, Any]] = []
143 for raw_c in raw:
144 if not isinstance(raw_c, dict):
145 continue
146 score = _first_present(
147 raw_c,
148 score_keys,
149 default=None if preserve_absent_score else 0,
150 )
151 excerpt = _first_present(raw_c, excerpt_keys, default="")
152 if score is None and preserve_absent_score:
153 normalized_score = None
154 else:
155 try:
156 normalized_score = int(score or 0)
157 except (TypeError, ValueError):
158 normalized_score = 0
159 entry: dict[str, Any] = {
160 "score": normalized_score,
161 "excerpt": str(excerpt or "")[:400],
162 "author": str(raw_c.get("author") or ""),
163 "date": str(raw_c.get("date") or ""),
164 }
165 if raw_c.get("url"):
166 entry["url"] = str(raw_c["url"])
167 out.append(entry)
168 return out
169
170
171 def _first_present(d: dict[str, Any], keys: tuple[str, ...], default: Any) -> Any:
172 for key in keys:
173 if key in d and d[key] not in (None, ""):
174 return d[key]
175 return default
176
177
178 def _join_comment_excerpts(
179 top_comments: list[Any],
180 key: str,
181 limit: int = 3,
182 ) -> str:
183 """Space-join the `key` field from the first `limit` dict-shaped comments."""
184 return " ".join(
185 str(comment.get(key) or "").strip()
186 for comment in top_comments[:limit]
187 if isinstance(comment, dict)
188 )
189
190
191 def _domain_from_url(url: str) -> str | None:
192 if not url:
193 return None
194 domain = urlparse(url).netloc.strip().lower()
195 return domain or None
196
197
198 def _date_confidence(
199 item: dict[str, Any], from_date: str, to_date: str, default: str = "low"
200 ) -> str:
201 if item.get("date_confidence"):
202 return str(item["date_confidence"])
203 date_value = item.get("date")
204 if not date_value:
205 return default
206 return dates.get_date_confidence(str(date_value), from_date, to_date)
207
208
209 def _source_item(
210 *,
211 item_id: str,
212 source: str,
213 title: str,
214 body: str,
215 url: str,
216 published_at: str | None,
217 date_confidence: str,
218 relevance_hint: float,
219 why_relevant: str,
220 author: str | None = None,
221 container: str | None = None,
222 engagement: dict[str, float | int] | None = None,
223 snippet: str = "",
224 metadata: dict[str, Any] | None = None,
225 ) -> schema.SourceItem:
226 return schema.SourceItem(
227 item_id=item_id,
228 source=source,
229 title=title.strip() or body.strip()[:160] or item_id,
230 body=body.strip(),
231 url=url.strip(),
232 author=(author or "").strip() or None,
233 container=(container or "").strip() or None,
234 published_at=published_at,
235 date_confidence=date_confidence,
236 engagement=engagement or {},
237 relevance_hint=max(0.0, min(1.0, float(relevance_hint or 0.0))),
238 why_relevant=why_relevant.strip(),
239 snippet=snippet.strip(),
240 metadata=metadata or {},
241 )
242
243
244 def _normalize_stocktwits(
245 source: str,
246 item: dict[str, Any],
247 index: int,
248 from_date: str,
249 to_date: str,
250 ) -> schema.SourceItem:
251 meta = item.get("metadata") or {}
252 return _source_item(
253 item_id=str(item.get("id") or f"ST{index + 1}"),
254 source=source,
255 title=str(item.get("title") or ""),
256 body=str(item.get("snippet") or ""),
257 url=str(item.get("url") or ""),
258 author=str(item.get("author") or "") or None,
259 container=str(meta.get("symbol") or "") or None,
260 published_at=item.get("date"),
261 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
262 engagement=item.get("engagement") or {},
263 relevance_hint=item.get("relevance", 0.7),
264 why_relevant=str(item.get("why_relevant") or ""),
265 snippet=str(item.get("snippet") or "")[:400],
266 metadata=meta, # carries sentiment + symbol-level bull/bear aggregate
267 )
268
269
270 def _normalize_dripstack(
271 source: str,
272 item: dict[str, Any],
273 index: int,
274 from_date: str,
275 to_date: str,
276 ) -> schema.SourceItem:
277 """Normalizer for DripStack newsletter search results.
278
279 DripStack returns article metadata from paid financial newsletters.
280 No engagement signal — ranking relies on DripStack's own relevanceScore
281 (0-100, normalized to 0-1) plus recency. The publication name serves as
282 author/attribution (e.g. "SemiAnalysis", "Bloomberg").
283 """
284 meta = item.get("metadata") or {}
285 return _source_item(
286 item_id=str(item.get("id") or f"DS{index + 1}"),
287 source=source,
288 title=str(item.get("title") or ""),
289 body=str(item.get("body") or "")
290 or str(item.get("snippet") or "")
291 or str(item.get("title") or ""),
292 url=str(item.get("url") or ""),
293 author=str(item.get("author") or "") or None,
294 container=str(meta.get("publication_slug") or "") or None,
295 published_at=item.get("date"),
296 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
297 engagement={},
298 relevance_hint=item.get("relevance", 0.5),
299 why_relevant=str(item.get("why_relevant") or ""),
300 snippet=str(item.get("snippet") or "")[:400],
301 metadata={
302 **meta,
303 "publication_slug": meta.get("publication_slug"),
304 },
305 )
306
307
308 def _normalize_reddit(
309 source: str,
310 item: dict[str, Any],
311 index: int,
312 from_date: str,
313 to_date: str,
314 ) -> schema.SourceItem:
315 top_comments = item.get("top_comments") or []
316 comment_text = _join_comment_excerpts(top_comments, "excerpt")
317 body = "\n".join(
318 part
319 for part in [
320 str(item.get("title") or "").strip(),
321 str(item.get("selftext") or "").strip(),
322 comment_text,
323 ]
324 if part
325 )
326 return _source_item(
327 item_id=str(item.get("id") or f"R{index + 1}"),
328 source=source,
329 title=str(item.get("title") or ""),
330 body=body,
331 url=str(item.get("url") or ""),
332 author=None,
333 container=str(item.get("subreddit") or ""),
334 published_at=item.get("date"),
335 date_confidence=_date_confidence(item, from_date, to_date),
336 engagement=item.get("engagement") or {},
337 relevance_hint=item.get("relevance", 0.5),
338 why_relevant=str(item.get("why_relevant") or ""),
339 snippet=comment_text or str(item.get("selftext") or "")[:400],
340 metadata={
341 "top_comments": top_comments,
342 "comment_insights": item.get("comment_insights") or [],
343 },
344 )
345
346
347 def _normalize_x(
348 source: str,
349 item: dict[str, Any],
350 index: int,
351 from_date: str,
352 to_date: str,
353 ) -> schema.SourceItem:
354 text = str(item.get("text") or "").strip()
355 mentioned = item.get("mentioned_handles") or []
356 return _source_item(
357 item_id=str(item.get("id") or f"X{index + 1}"),
358 source=source,
359 title=text[:140] or f"X post {index + 1}",
360 body=text,
361 url=str(item.get("url") or ""),
362 author=str(item.get("author_handle") or "").lstrip("@"),
363 published_at=item.get("date"),
364 date_confidence=_date_confidence(item, from_date, to_date),
365 engagement=item.get("engagement") or {},
366 relevance_hint=item.get("relevance", 0.5),
367 why_relevant=str(item.get("why_relevant") or ""),
368 metadata={"mentioned_handles": list(mentioned)} if mentioned else {},
369 )
370
371
372 def _normalize_jobs(
373 source: str,
374 item: dict[str, Any],
375 index: int,
376 from_date: str,
377 to_date: str,
378 ) -> schema.SourceItem:
379 description = str(item.get("description") or item.get("snippet") or "").strip()
380 title = str(item.get("title") or "").strip()
381 department = str(item.get("department") or "").strip()
382 location = str(item.get("location") or "").strip()
383 body = "\n".join(
384 part for part in [title, department, location, description] if part
385 )
386 provider = str(item.get("provider") or "").strip()
387 return _source_item(
388 item_id=str(item.get("id") or f"J{index + 1}"),
389 source=source,
390 title=title or f"Job posting {index + 1}",
391 body=body,
392 url=str(item.get("url") or ""),
393 author=provider or None,
394 container=department or None,
395 published_at=item.get("date"),
396 date_confidence=_date_confidence(item, from_date, to_date),
397 engagement={"open_roles": 1},
398 relevance_hint=item.get("relevance", 0.65),
399 why_relevant=str(item.get("why_relevant") or "Public job posting"),
400 snippet=description[:500],
401 metadata={
402 "provider": provider,
403 "department": department,
404 "departments": item.get("departments")
405 or ([department] if department else []),
406 "location": location,
407 "offices": item.get("offices") or [],
408 "board_token": item.get("board_token") or "",
409 "source_url": item.get("source_url") or "",
410 "source_domain": item.get("source_domain")
411 or _domain_from_url(str(item.get("url") or ""))
412 or "",
413 },
414 )
415
416
417 def _normalize_youtube(
418 source: str,
419 item: dict[str, Any],
420 index: int,
421 from_date: str,
422 to_date: str,
423 ) -> schema.SourceItem:
424 transcript = str(item.get("transcript_snippet") or "").strip()
425 description = str(item.get("description") or "").strip()
426 title = str(item.get("title") or "").strip()
427 highlights = item.get("transcript_highlights") or []
428 metadata: dict[str, Any] = {}
429 if highlights:
430 metadata["transcript_highlights"] = highlights
431 if item.get("captions_disabled"):
432 # Surfaced for quality_nudge: uploader disabled captions, so this
433 # video should be subtracted from the degraded-transcript-ratio
434 # denominator (it was never going to produce a transcript).
435 metadata["captions_disabled"] = True
436 metadata["top_comments"] = _remap_comments(
437 item.get("top_comments") or [],
438 score_keys=("score", "likes"),
439 excerpt_keys=("excerpt", "text"),
440 )
441 return _source_item(
442 item_id=str(item.get("video_id") or item.get("id") or f"YT{index + 1}"),
443 source=source,
444 title=title,
445 body="\n".join(part for part in [title, description, transcript] if part),
446 url=str(item.get("url") or ""),
447 author=str(item.get("channel_name") or ""),
448 published_at=item.get("date"),
449 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
450 engagement=item.get("engagement") or {},
451 relevance_hint=item.get("relevance", 0.5),
452 why_relevant=str(item.get("why_relevant") or ""),
453 snippet=transcript,
454 metadata=metadata,
455 )
456
457
458 def _normalize_shortform_video(
459 source: str,
460 item: dict[str, Any],
461 index: int,
462 from_date: str,
463 to_date: str,
464 id_prefix: str,
465 default_title: str,
466 ) -> schema.SourceItem:
467 """Shared normalizer for TikTok and Instagram (identical structure)."""
468 caption = str(item.get("caption_snippet") or "").strip()
469 text = str(item.get("text") or "").strip()
470 return _source_item(
471 item_id=str(item.get("id") or f"{id_prefix}{index + 1}"),
472 source=source,
473 title=text[:140] or caption[:140] or f"{default_title} {index + 1}",
474 body="\n".join(part for part in [text, caption] if part),
475 url=str(item.get("url") or ""),
476 author=str(item.get("author_name") or ""),
477 published_at=item.get("date"),
478 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
479 engagement=item.get("engagement") or {},
480 relevance_hint=item.get("relevance", 0.5),
481 why_relevant=str(item.get("why_relevant") or ""),
482 snippet=caption,
483 metadata={
484 "hashtags": item.get("hashtags") or [],
485 "top_comments": _remap_comments(
486 item.get("top_comments") or [],
487 # Instagram comments use comment_like_count as the vote field
488 # (ScrapeCreators /v2/instagram/post/comments); digg_count/likes
489 # kept for shape compatibility.
490 score_keys=("score", "comment_like_count", "digg_count", "likes"),
491 excerpt_keys=("excerpt", "text"),
492 ),
493 },
494 )
495
496
497 def _normalize_pinterest(
498 source: str,
499 item: dict[str, Any],
500 index: int,
501 from_date: str,
502 to_date: str,
503 ) -> schema.SourceItem:
504 """Normalizer for Pinterest pins (visual content with descriptions).
505
506 Saves are the primary engagement signal, analogous to likes/upvotes.
507 """
508 description = str(item.get("description") or "").strip()
509 return _source_item(
510 item_id=str(item.get("pin_id") or item.get("id") or f"PI{index + 1}"),
511 source=source,
512 title=description[:140] or f"Pinterest pin {index + 1}",
513 body=description,
514 url=str(item.get("url") or ""),
515 author=str(item.get("author") or ""),
516 container=str(item.get("board") or ""),
517 published_at=item.get("date"),
518 date_confidence=_date_confidence(item, from_date, to_date, default="low"),
519 engagement=item.get("engagement") or {},
520 relevance_hint=item.get("relevance", 0.5),
521 why_relevant=str(item.get("why_relevant") or ""),
522 snippet=description[:400],
523 )
524
525
526 def _normalize_hackernews(
527 source: str,
528 item: dict[str, Any],
529 index: int,
530 from_date: str,
531 to_date: str,
532 ) -> schema.SourceItem:
533 # HN comments arrive as {author, text, points}; downstream code keys on
534 # score/excerpt, so remap here exactly as the YouTube and TikTok normalisers
535 # do. Without this the per-source floor in render._top_comments_list reads a
536 # `score` that is never present and rejects every HN comment.
537 top_comments = _remap_comments(
538 item.get("top_comments") or [],
539 score_keys=("points", "score"),
540 excerpt_keys=("text", "excerpt"),
541 preserve_absent_score=True,
542 )
543 comment_text = _join_comment_excerpts(top_comments, "excerpt")
544 title = str(item.get("title") or "").strip()
545 body = "\n".join(
546 part
547 for part in [title, str(item.get("text") or "").strip(), comment_text]
548 if part
549 )
550 return _source_item(
551 item_id=str(item.get("id") or f"HN{index + 1}"),
552 source=source,
553 title=title or f"HN story {index + 1}",
554 body=body,
555 url=str(item.get("url") or item.get("hn_url") or ""),
556 author=str(item.get("author") or ""),
557 container="Hacker News",
558 published_at=item.get("date"),
559 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
560 engagement=item.get("engagement") or {},
561 relevance_hint=item.get("relevance", 0.5),
562 why_relevant=str(item.get("why_relevant") or ""),
563 snippet=comment_text,
564 metadata={
565 "hn_url": item.get("hn_url"),
566 "top_comments": top_comments,
567 "comment_insights": item.get("comment_insights") or [],
568 },
569 )
570
571
572 def _normalize_microblog(
573 source: str,
574 item: dict[str, Any],
575 index: int,
576 from_date: str,
577 to_date: str,
578 id_prefix: str,
579 default_title: str,
580 ) -> schema.SourceItem:
581 """Shared normalizer for Bluesky and Truth Social (identical structure)."""
582 text = str(item.get("text") or "").strip()
583 return _source_item(
584 item_id=str(item.get("id") or f"{id_prefix}{index + 1}"),
585 source=source,
586 title=text[:140] or f"{default_title} {index + 1}",
587 body=text,
588 url=str(item.get("url") or ""),
589 author=str(item.get("handle") or item.get("author_handle") or "").lstrip("@"),
590 published_at=item.get("date"),
591 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
592 engagement=item.get("engagement") or {},
593 relevance_hint=item.get("relevance", 0.5),
594 why_relevant=str(item.get("why_relevant") or ""),
595 metadata={"display_name": item.get("display_name")},
596 )
597
598
599 def _normalize_digg(
600 source: str,
601 item: dict[str, Any],
602 index: int,
603 from_date: str,
604 to_date: str,
605 ) -> schema.SourceItem:
606 """Normalizer for Digg AI 1000 clusters.
607
608 Each cluster is one item. The TLDR carries the most useful body for
609 rerank and synthesis. Top-ranked X posts attached at search time are
610 passed through under metadata['posts'] so render can emit them as
611 inline 'via Digg' quotes.
612 """
613 title = str(item.get("title") or "").strip()
614 tldr = str(item.get("tldr") or "").strip()
615 body = "\n\n".join(part for part in [title, tldr] if part)
616 posts = item.get("posts") or []
617 if not isinstance(posts, list):
618 posts = []
619 cluster_url_id = str(item.get("id") or f"DG{index + 1}")
620 return _source_item(
621 item_id=cluster_url_id,
622 source=source,
623 title=title or f"Digg cluster {index + 1}",
624 body=body,
625 url=str(item.get("url") or f"https://di.gg/ai/{cluster_url_id}"),
626 author="",
627 container="Digg",
628 published_at=item.get("date"),
629 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
630 engagement=item.get("engagement") or {},
631 relevance_hint=item.get("relevance", 0.5),
632 why_relevant=str(item.get("why_relevant") or ""),
633 snippet=tldr[:400],
634 metadata={
635 "clusterUrlId": cluster_url_id,
636 "tldr": tldr,
637 "rank": (item.get("engagement") or {}).get("rank"),
638 "uniqueAuthors": (item.get("engagement") or {}).get("uniqueAuthors"),
639 "postCount": (item.get("engagement") or {}).get("postCount"),
640 "firstPostAge": item.get("first_post_age"),
641 "posts": posts,
642 },
643 )
644
645
646 def _normalize_arxiv(
647 source: str,
648 item: dict[str, Any],
649 index: int,
650 from_date: str,
651 to_date: str,
652 ) -> schema.SourceItem:
653 """Normalizer for arXiv papers.
654
655 The abstract (summary) is the body that feeds rerank and synthesis. arXiv
656 has no engagement signal, so engagement is empty and ranking leans on
657 relevance and recency.
658 """
659 title = str(item.get("title") or "").strip()
660 summary = str(item.get("summary") or "").strip()
661 body = "\n\n".join(part for part in [title, summary] if part)
662 authors = item.get("authors") or []
663 if not isinstance(authors, list):
664 authors = []
665 paper_id = str(item.get("id") or f"AX{index + 1}")
666 return _source_item(
667 item_id=paper_id,
668 source=source,
669 title=title or f"arXiv paper {index + 1}",
670 body=body,
671 url=str(item.get("url") or ""),
672 author=str(item.get("author") or "") or None,
673 container="arXiv",
674 published_at=item.get("date"),
675 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
676 engagement={},
677 relevance_hint=item.get("relevance", 0.5),
678 why_relevant=str(item.get("why_relevant") or ""),
679 snippet=summary[:400],
680 metadata={
681 "authors": authors,
682 "summary": summary,
683 },
684 )
685
686
687 def _normalize_techmeme(
688 source: str,
689 item: dict[str, Any],
690 index: int,
691 from_date: str,
692 to_date: str,
693 ) -> schema.SourceItem:
694 """Normalizer for Techmeme headlines.
695
696 The headline is both title and body (Techmeme carries no abstract). The
697 publication is the container/author. No engagement signal in the search
698 shape, so ranking leans on relevance and recency.
699 """
700 title = str(item.get("title") or "").strip()
701 source_name = str(item.get("source_name") or "").strip()
702 return _source_item(
703 item_id=str(item.get("id") or f"TM{index + 1}"),
704 source=source,
705 title=title or f"Techmeme headline {index + 1}",
706 body=title,
707 url=str(item.get("url") or ""),
708 author=source_name or None,
709 container=source_name or "Techmeme",
710 published_at=item.get("date"),
711 date_confidence=_date_confidence(item, from_date, to_date, default="low"),
712 engagement={},
713 relevance_hint=item.get("relevance", 0.5),
714 why_relevant=str(item.get("why_relevant") or ""),
715 snippet=title[:400],
716 metadata={
717 "publication": source_name,
718 },
719 )
720
721
722 def _normalize_meta_ads(
723 source: str,
724 item: dict[str, Any],
725 index: int,
726 from_date: str,
727 to_date: str,
728 ) -> schema.SourceItem:
729 """Normalizer for Meta Ad Library creatives.
730
731 One item per distinct creative launched inside the window. Identity is the
732 Ad Library permalink, never the landing URL: many creatives for one product
733 share a landing page, and fusion merges by normalized URL, so keying on the
734 landing page would collapse a whole campaign into one candidate and lose
735 every transcript but the first.
736
737 Grounding-exempt on the Amazon precedent. Ad copy is written to sell, not
738 to name the brand -- "Stop making boring drinks" never repeats the topic --
739 and the advertiser page was already resolved by name before any of these
740 were created, so they are on-entity by construction.
741 """
742 body = str(item.get("text") or "").strip()
743 transcript = str(item.get("transcript") or "").strip()
744 advertiser = str(item.get("advertiser") or "").strip()
745 cta = str(item.get("cta") or "").strip()
746 landing = str(item.get("landing_url") or "").strip()
747 placements = [str(p) for p in (item.get("placements") or [])]
748 promo = str(item.get("promo_code") or "").strip()
749
750 # The spoken script is usually the sharper version of the pitch, so it
751 # leads the evidence snippet when present.
752 snippet_parts = [part for part in [body, transcript] if part]
753 # The adapter deliberately keeps creatives that launched inside the window
754 # and have since ended -- a one-week promo push is exactly the signal this
755 # source exists for -- so the wording has to follow the stored state rather
756 # than calling every creative active.
757 running = bool(item.get("is_active"))
758 ended_on = str(item.get("ended_on") or "").strip()
759 if running:
760 state_word = "Running paid creative"
761 elif ended_on:
762 state_word = f"Paid creative that ran until {ended_on}"
763 else:
764 state_word = "Paid creative that has since ended"
765 default_why = f"{state_word} from {advertiser}" if advertiser else state_word
766 context = " │ ".join(
767 part
768 for part in [
769 cta or "",
770 f"code {promo}" if promo else "",
771 landing,
772 ]
773 if part
774 )
775 return _source_item(
776 item_id=str(item.get("id") or f"MA{index + 1}"),
777 source=source,
778 title=str(item.get("title") or "").strip() or body[:140] or f"Meta ad {index + 1}",
779 body=body,
780 url=str(item.get("url") or "").strip(),
781 author=advertiser,
782 container="Meta Ad Library",
783 published_at=item.get("date"),
784 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
785 engagement={"variants": int(item.get("variants") or 1)},
786 relevance_hint=0.8,
787 why_relevant=str(item.get("why_relevant") or "") or default_why,
788 snippet=" ".join(snippet_parts)[:400],
789 metadata={
790 "grounding_exempt": True,
791 "advertiser": advertiser,
792 "page_id": str(item.get("page_id") or ""),
793 "is_active": bool(item.get("is_active")),
794 "ended_on": item.get("ended_on"),
795 "display_format": str(item.get("display_format") or ""),
796 "placements": placements,
797 "cta": cta,
798 "landing_url": landing,
799 "promo_code": promo,
800 "variants": int(item.get("variants") or 1),
801 "has_video": bool(item.get("has_video")),
802 "transcript_snippet": transcript[:400],
803 "ad_context": context,
804 },
805 )
806
807
808 def _normalize_trustpilot(
809 source: str,
810 item: dict[str, Any],
811 index: int,
812 from_date: str,
813 to_date: str,
814 ) -> schema.SourceItem:
815 """Normalizer for Trustpilot company sentiment.
816
817 One item per company. The AI summary (already balanced positive/negative)
818 is the body. TrustScore and review count are engagement and metadata.
819 """
820 title = str(item.get("title") or "").strip()
821 name = str(item.get("name") or "").strip()
822 summary = str(item.get("summary") or "").strip()
823 body = "\n\n".join(part for part in [title, summary] if part)
824 return _source_item(
825 item_id=str(item.get("id") or f"TP{index + 1}"),
826 source=source,
827 title=title
828 or (f"{name} on Trustpilot" if name else f"Trustpilot reviews {index + 1}"),
829 body=body,
830 url=str(item.get("url") or ""),
831 author=name or None,
832 container="Trustpilot",
833 published_at=item.get("date"),
834 date_confidence=_date_confidence(item, from_date, to_date, default="low"),
835 engagement=item.get("engagement") or {},
836 relevance_hint=item.get("relevance", 0.6),
837 why_relevant=str(item.get("why_relevant") or ""),
838 snippet=summary[:400],
839 metadata={
840 "name": name,
841 "trustScore": item.get("trustScore"),
842 "reviewCount": item.get("reviewCount"),
843 "aiSummary": summary,
844 },
845 )
846
847
848 def _normalize_amazon(
849 source: str,
850 item: dict[str, Any],
851 index: int,
852 from_date: str,
853 to_date: str,
854 ) -> schema.SourceItem:
855 """Normalizer for Amazon product-and-review signals.
856
857 One item per product. The aggregate rating is current-state evidence, so
858 the item is stamped with today's date on the Trustpilot precedent -- a
859 live 4.4-star average is a fact about now, not about whenever the
860 product launched.
861
862 Reviews arrive already in the shared score/excerpt comment shape (built
863 in the amazon adapter, deliberately not routed through _remap_comments,
864 which would strip the rating/date/verified keys this source needs), so
865 they pass straight through to metadata.
866 """
867 name = str(item.get("name") or "").strip()
868 brand = str(item.get("brand") or "").strip()
869 top_comments = item.get("top_comments") or []
870 comment_text = _join_comment_excerpts(top_comments, "excerpt")
871 rating = item.get("product_rating") if item.get("product_rating") is not None else item.get("rating")
872 ratings_total = item.get("product_rating_count") or item.get("num_ratings") or 0
873
874 headline = " ".join(
875 part for part in [
876 f"{rating}/5" if rating is not None else "",
877 f"({ratings_total:,} ratings)" if ratings_total else "",
878 ] if part
879 )
880 # The brand rides in its own field and is usually absent from the name,
881 # so prepend it -- unless the name already leads with it, which would
882 # otherwise read "Weber Weber Spirit E-325".
883 if brand and not name.lower().startswith(brand.lower()):
884 product_label = f"{brand} {name}".strip()
885 else:
886 product_label = name or brand
887 title = " - ".join(part for part in [product_label, headline] if part)
888 body = "\n".join(part for part in [title, comment_text] if part)
889
890 return _source_item(
891 item_id=str(item.get("asin") or f"AMZ{index + 1}"),
892 source=source,
893 title=title or f"Amazon product {index + 1}",
894 body=body,
895 url=str(item.get("url") or ""),
896 author=brand or None,
897 container="Amazon",
898 published_at=item.get("date"),
899 date_confidence=_date_confidence(item, from_date, to_date, default="low"),
900 engagement=item.get("engagement") or {"ratings": ratings_total},
901 relevance_hint=item.get("relevance", 0.6),
902 why_relevant=str(item.get("why_relevant") or ""),
903 snippet=comment_text[:400],
904 metadata={
905 "asin": str(item.get("asin") or ""),
906 "name": name,
907 "short_name": item.get("short_name") or "",
908 "brand": brand,
909 "rating": item.get("rating"),
910 "num_ratings": item.get("num_ratings") or 0,
911 "price": item.get("price"),
912 "currency": item.get("currency") or "",
913 "badge": item.get("badge") or "",
914 # Recorded, never used as a filter: the flag's distribution
915 # swings with keyword phrasing, so filtering can blank the lane.
916 "sponsored": bool(item.get("sponsored")),
917 "top_comments": top_comments,
918 "product_rating": item.get("product_rating"),
919 "product_rating_count": item.get("product_rating_count") or 0,
920 "star_distribution": item.get("star_distribution") or {},
921 # Relevant by construction: the adapter already gated products
922 # against the model-supplied keyword, and review text rarely
923 # names the product (KTD8).
924 "grounding_exempt": True,
925 },
926 )
927
928
929 def _normalize_polymarket(
930 source: str,
931 item: dict[str, Any],
932 index: int,
933 from_date: str,
934 to_date: str,
935 ) -> schema.SourceItem:
936 title = str(item.get("title") or "").strip()
937 question = str(item.get("question") or "").strip()
938 engagement = {
939 "volume": item.get("volume1mo") or item.get("volume24hr") or 0,
940 "liquidity": item.get("liquidity") or 0,
941 }
942 return _source_item(
943 item_id=str(item.get("event_id") or item.get("id") or f"PM{index + 1}"),
944 source=source,
945 title=title or question or f"Polymarket event {index + 1}",
946 body="\n".join(
947 part
948 for part in [title, question, str(item.get("price_movement") or "")]
949 if part
950 ),
951 url=str(item.get("url") or ""),
952 author=None,
953 container="Polymarket",
954 published_at=item.get("date"),
955 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
956 engagement=engagement,
957 relevance_hint=item.get("relevance", 0.5),
958 why_relevant=str(item.get("why_relevant") or ""),
959 snippet=str(item.get("price_movement") or ""),
960 metadata={
961 "event_id": item.get("event_id"),
962 "question": question,
963 "end_date": item.get("end_date"),
964 "outcome_prices": item.get("outcome_prices") or [],
965 "outcomes_remaining": item.get("outcomes_remaining"),
966 },
967 )
968
969
970 def _normalize_github(
971 source: str,
972 item: dict[str, Any],
973 index: int,
974 from_date: str,
975 to_date: str,
976 ) -> schema.SourceItem:
977 title = str(item.get("title") or "").strip()
978 snippet_text = str(item.get("snippet") or "").strip()
979 top_comments = item.get("metadata", {}).get("top_comments") or []
980 comment_text = _join_comment_excerpts(top_comments, "excerpt")
981 body = "\n".join(part for part in [title, snippet_text, comment_text] if part)
982 metadata = item.get("metadata") or {}
983 return _source_item(
984 item_id=str(item.get("id") or f"GH{index + 1}"),
985 source=source,
986 title=title or f"GitHub item {index + 1}",
987 body=body,
988 url=str(item.get("url") or ""),
989 author=str(item.get("author") or ""),
990 container=str(item.get("container") or ""),
991 published_at=item.get("date"),
992 date_confidence=_date_confidence(item, from_date, to_date, default="high"),
993 engagement=item.get("engagement") or {},
994 relevance_hint=item.get("relevance", 0.5),
995 why_relevant=str(item.get("why_relevant") or ""),
996 snippet=comment_text or snippet_text[:400],
997 metadata={
998 "top_comments": top_comments,
999 "labels": metadata.get("labels") or [],
1000 "state": metadata.get("state", ""),
1001 "is_pr": metadata.get("is_pr", False),
1002 },
1003 )
1004
1005
1006 def _normalize_grounding(
1007 source: str,
1008 item: dict[str, Any],
1009 index: int,
1010 from_date: str,
1011 to_date: str,
1012 ) -> schema.SourceItem:
1013 title = str(item.get("title") or "").strip()
1014 snippet = str(item.get("snippet") or "").strip()
1015 url = str(item.get("url") or "").strip()
1016 return _source_item(
1017 item_id=str(item.get("id") or f"W{index + 1}"),
1018 source=source,
1019 title=title or _domain_from_url(url) or f"Web result {index + 1}",
1020 body="\n".join(part for part in [title, snippet] if part),
1021 url=url,
1022 author=None,
1023 container=str(item.get("source_domain") or _domain_from_url(url) or ""),
1024 published_at=item.get("date"),
1025 date_confidence=_date_confidence(item, from_date, to_date),
1026 engagement=item.get("engagement") or {},
1027 relevance_hint=item.get("relevance", 0.5),
1028 why_relevant=str(item.get("why_relevant") or ""),
1029 snippet=snippet,
1030 metadata=item.get("metadata") or {},
1031 )
1032
1033
1034 def _normalize_linkedin(
1035 source: str,
1036 item: dict[str, Any],
1037 index: int,
1038 from_date: str,
1039 to_date: str,
1040 ) -> schema.SourceItem:
1041 """Normalizer for LinkedIn posts and articles via ScrapeCreators.
1042
1043 A LinkedIn article (Pulse long-form, under a /pulse/ URL) is treated as
1044 high signal: it ranks above ordinary posts. Detection is belt-and-suspenders
1045 — honor the parser's `is_article` flag, and re-derive from the URL so an
1046 article still ranks high even if the flag wasn't set upstream.
1047 """
1048 text = str(item.get("text") or "").strip()
1049 author = str(item.get("author") or "").strip()
1050 url = str(item.get("url") or "").strip()
1051 is_article = bool(item.get("is_article")) or "/pulse/" in url.lower()
1052 kind = "article" if is_article else "post"
1053 default_relevance = 0.9 if is_article else 0.5
1054 return _source_item(
1055 item_id=str(item.get("id") or f"LI{index + 1}"),
1056 source=source,
1057 title=text[:140] or f"LinkedIn {kind} {index + 1}",
1058 body=text,
1059 url=url,
1060 author=author,
1061 container="LinkedIn Article" if is_article else "LinkedIn",
1062 published_at=item.get("date"),
1063 date_confidence=_date_confidence(item, from_date, to_date, default="medium"),
1064 engagement=item.get("engagement") or {},
1065 relevance_hint=item.get("relevance", default_relevance),
1066 why_relevant=str(item.get("why_relevant") or ""),
1067 snippet=text[:200],
1068 metadata={"author_display": author, "is_article": is_article},
1069 )
1070
1070 lines PYTHON