feat: add Digg AI 1000 as an opt-in source (#370)
* feat(digg): add Digg AI 1000 source module with cluster search and post enrichment - search_digg shells out to digg-pp-cli with --since 30d --agent - parse_digg_response normalizes clusters to last30days dict shape - enrich_with_top_posts attaches top-ranked X posts to top-K clusters - shutil.which gate plus subproc.run_with_timeout discipline matches bird_x.py / youtube_yt.py patterns 25 unit tests cover parse, age window, relevance, binary-missing fallback, timeout recovery, and partial enrichment failures. * feat(digg): wire Digg source into pipeline, normalize, signals, and render pipeline.py: - Import digg, add to MOCK_AVAILABLE_SOURCES, gate via shutil.which - Dispatch case calls search_digg + parse_digg_response, runs enrich_with_top_posts at default/deep depth - Mock fixture includes one enriched cluster + one bare cluster normalize.py: - _normalize_digg maps cluster dicts to SourceItem with container='Digg AI 1000' and metadata.posts pass-through signals.py: - SOURCE_QUALITY['digg'] = 0.85 (top tier alongside YouTube, reflecting Digg's curatorial layer) - ENGAGEMENT_WEIGHTS['digg'] balances postCount, uniqueAuthors, and the rank_score derived from Digg's curatorial position render.py: - SOURCE_LABELS['digg'] = 'Digg AI 1000' - _FOOTER_SOURCES adds '⛏️ Digg AI 1000' line after GitHub - ENGAGEMENT_DISPLAY mirrors footer keys - New _digg_posts_for + _format_digg_quote helpers emit inline '@handle via Digg AI 1000' quotes for clusters with attached X posts; both compact and full-dump renderers call them * feat(digg): polish per-item engagement display and progress label - ENGAGEMENT_DISPLAY for digg uses 'posts' / 'auth' to match the codebase abbreviation convention (HN: 'pts'/'cmt', X: 'rt'/'re') - Footer item word changes from 'story' to 'cluster' to dodge the pre-existing naive plural in _footer_line_for_source ('storys') and to match Digg's actual data model - ui.py SOURCE_COMPLETION_META adds digg with correct 'cluster'/ 'clusters' plural so 'Research complete' shows 'Digg: N clusters' * feat(digg): document Digg AI 1000 source in skill, README, and changelog - planner.py SOURCE_CAPABILITIES adds digg with discussion/social/link capabilities so the planner offers it through the standard fanout - SKILL.md ACTIVE_SOURCES_LIST gate includes 'which digg-pp-cli' check and the source list / available-sources line names digg as opt-in - README.md Sources table adds the Digg AI 1000 row with the activation gate so first-time readers see what they get - CHANGELOG.md Unreleased section calls out the source addition * fix(digg): enrich post-dedupe so brief survivors carry inline quotes Pipeline dispatch was attaching X posts to the top-3 items returned by search, but dedupe later picked different survivors when multiple clusters compared similar (common for trending topics). The brief ended up showing clusters with no posts attached even though enrichment ran successfully on positions 0-2. Move enrichment to _finalize_items_by_source. The new digg.enrich_source_items helper reads metadata['clusterUrlId'] and writes metadata['posts'] in place on the SourceItems that actually survive dedupe. Verified live on 'openclaw': 2 surviving clusters, both now carry real X-post quotes from @sama and @jeremyphoward attributed 'via Digg AI 1000'. Adds 3 unit tests covering survivor enrichment, non-digg skip, and clusterUrlId fallback to item_id. * test(digg): relax live off-topic test to check shape, not emptiness Digg's live search uses fuzzy/popularity fallback, so an impossible token can still return some loosely-related clusters. The contract the pipeline depends on is shape (results is always a list); token-overlap relevance handles the noise downstream. --------- Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com>
This commit is contained in:
Matt Van Horn
@@ -0,0 +1,413 @@
|
||||
"""Digg AI 1000 source for last30days.
|
||||
|
||||
Shells out to ``digg-pp-cli`` (read-only, no auth required) to surface
|
||||
clustered stories curated from ~1000 high-signal AI accounts on X. Each
|
||||
cluster carries a published TLDR, a curatorial rank, and a list of X
|
||||
posts that can be fetched as inline quotes.
|
||||
|
||||
Activation gate: this source is only available when ``digg-pp-cli`` is
|
||||
on PATH. ``pipeline.available_sources`` checks ``shutil.which`` before
|
||||
including ``digg`` in the source list. The functions below also detect
|
||||
the missing-binary case as a defensive fallback.
|
||||
|
||||
Primary path: ``digg-pp-cli search <topic> --since 30d --agent --limit N``.
|
||||
Optional enrichment: ``digg-pp-cli posts <clusterUrlId> --agent --by rank
|
||||
--limit M`` for the top K clusters in default/deep depth, attaching the
|
||||
top-ranked X posts to each cluster's ``posts`` field.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from . import log, subproc
|
||||
from .relevance import token_overlap_relevance
|
||||
|
||||
|
||||
CLI_BIN = "digg-pp-cli"
|
||||
|
||||
# Per-depth knobs.
|
||||
DEPTH_CONFIG = {
|
||||
"quick": 8,
|
||||
"default": 20,
|
||||
"deep": 40,
|
||||
}
|
||||
|
||||
# How many top-ranked clusters get post enrichment, per depth. Quick mode
|
||||
# skips enrichment to keep latency low (clusters already carry a TLDR).
|
||||
ENRICH_CONFIG = {
|
||||
"quick": 0,
|
||||
"default": 3,
|
||||
"deep": 5,
|
||||
}
|
||||
|
||||
# X posts pulled per enriched cluster.
|
||||
POSTS_PER_CLUSTER = 3
|
||||
|
||||
SEARCH_TIMEOUT = 30
|
||||
POSTS_TIMEOUT = 15
|
||||
|
||||
|
||||
def _log(msg: str) -> None:
|
||||
log.source_log("Digg", msg)
|
||||
|
||||
|
||||
def _is_available() -> bool:
|
||||
"""True when the digg-pp-cli binary is on PATH."""
|
||||
return shutil.which(CLI_BIN) is not None
|
||||
|
||||
|
||||
def _today() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _parse_first_post_age(age: Optional[str], today: Optional[datetime] = None) -> Optional[str]:
|
||||
"""Convert a digg firstPostAge token (e.g. '5d', '17d', '5h', '1w', '1m')
|
||||
into a YYYY-MM-DD string. Returns None when the value is outside the
|
||||
last-30-day window or cannot be parsed.
|
||||
|
||||
Digg uses minutes-symbol-collision for 'months' (per agent-context:
|
||||
'Nh, Nd, Nw, Nm (e.g. 30d, 1w, 12h, 1m)'), so 'Nm' is months ~30 days.
|
||||
"""
|
||||
if not age or not isinstance(age, str):
|
||||
return None
|
||||
age = age.strip().lower()
|
||||
if len(age) < 2:
|
||||
return None
|
||||
unit = age[-1]
|
||||
try:
|
||||
amount = int(age[:-1])
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
if amount < 0:
|
||||
return None
|
||||
|
||||
base = today or _today()
|
||||
|
||||
if unit == "h":
|
||||
delta = timedelta(hours=amount)
|
||||
elif unit == "d":
|
||||
delta = timedelta(days=amount)
|
||||
elif unit == "w":
|
||||
delta = timedelta(weeks=amount)
|
||||
elif unit == "m":
|
||||
delta = timedelta(days=amount * 30)
|
||||
else:
|
||||
return None
|
||||
|
||||
if delta > timedelta(days=30):
|
||||
return None
|
||||
|
||||
point = base - delta
|
||||
return point.date().isoformat()
|
||||
|
||||
|
||||
def _build_search_args(query: str, limit: int) -> List[str]:
|
||||
return [
|
||||
CLI_BIN,
|
||||
"search",
|
||||
query,
|
||||
"--since",
|
||||
"30d",
|
||||
"--agent",
|
||||
"--limit",
|
||||
str(limit),
|
||||
]
|
||||
|
||||
|
||||
def _build_posts_args(cluster_url_id: str, posts_per: int) -> List[str]:
|
||||
return [
|
||||
CLI_BIN,
|
||||
"posts",
|
||||
cluster_url_id,
|
||||
"--agent",
|
||||
"--by",
|
||||
"rank",
|
||||
"--limit",
|
||||
str(posts_per),
|
||||
]
|
||||
|
||||
|
||||
def _run_cli(cmd: List[str], timeout: int) -> Dict[str, Any]:
|
||||
"""Invoke digg-pp-cli and parse the JSON envelope.
|
||||
|
||||
Returns ``{"results": [...]}`` on success, ``{"results": [], "error": "..."}``
|
||||
on failure. Never raises; the pipeline relies on shape consistency.
|
||||
"""
|
||||
if not _is_available():
|
||||
return {"results": [], "error": f"{CLI_BIN} not on PATH"}
|
||||
try:
|
||||
result = subproc.run_with_timeout(cmd, timeout=timeout)
|
||||
except subproc.SubprocTimeout as exc:
|
||||
_log(f"Timeout: {exc}")
|
||||
return {"results": [], "error": str(exc)}
|
||||
except FileNotFoundError as exc:
|
||||
_log(f"Binary missing: {exc}")
|
||||
return {"results": [], "error": str(exc)}
|
||||
except OSError as exc:
|
||||
_log(f"Spawn failed: {exc}")
|
||||
return {"results": [], "error": str(exc)}
|
||||
|
||||
if result.returncode != 0:
|
||||
snippet = (result.stderr or "").strip().splitlines()[:1]
|
||||
first = snippet[0] if snippet else f"exit {result.returncode}"
|
||||
_log(f"CLI exit {result.returncode}: {first}")
|
||||
return {"results": [], "error": first}
|
||||
|
||||
stdout = result.stdout or ""
|
||||
if not stdout.strip():
|
||||
return {"results": []}
|
||||
try:
|
||||
data = json.loads(stdout)
|
||||
except json.JSONDecodeError as exc:
|
||||
_log(f"JSON decode failed: {exc}")
|
||||
return {"results": [], "error": f"json decode: {exc}"}
|
||||
|
||||
if not isinstance(data, dict):
|
||||
return {"results": []}
|
||||
results = data.get("results")
|
||||
if not isinstance(results, list):
|
||||
return {"results": []}
|
||||
return data
|
||||
|
||||
|
||||
def search_digg(
|
||||
topic: str,
|
||||
from_date: str,
|
||||
to_date: str,
|
||||
depth: str = "default",
|
||||
) -> Dict[str, Any]:
|
||||
"""Search Digg AI 1000 clusters via digg-pp-cli.
|
||||
|
||||
Args:
|
||||
topic: search query.
|
||||
from_date: YYYY-MM-DD start (advisory; --since 30d is the actual filter).
|
||||
to_date: YYYY-MM-DD end (advisory; same).
|
||||
depth: 'quick' | 'default' | 'deep'.
|
||||
|
||||
Returns:
|
||||
Dict with ``results`` list. On failure, ``results`` is empty and an
|
||||
``error`` key carries a one-line description.
|
||||
"""
|
||||
limit = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
|
||||
if not topic or not topic.strip():
|
||||
return {"results": []}
|
||||
cmd = _build_search_args(topic, limit)
|
||||
_log(f"search '{topic}' (limit={limit}, since=30d)")
|
||||
response = _run_cli(cmd, timeout=SEARCH_TIMEOUT)
|
||||
n = len(response.get("results") or [])
|
||||
_log(f"found {n} clusters")
|
||||
return response
|
||||
|
||||
|
||||
def _build_url(cluster_url_id: str) -> str:
|
||||
return f"https://di.gg/ai/{cluster_url_id}"
|
||||
|
||||
|
||||
def _rank_score(rank: Optional[int]) -> float:
|
||||
"""Convert Digg rank (lower is better, top 50 are notable) into a
|
||||
positive engagement-style signal in [0, 50]. Anything off the top-50
|
||||
leaderboard contributes 0.
|
||||
"""
|
||||
if rank is None:
|
||||
return 0.0
|
||||
try:
|
||||
r = int(rank)
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
if r < 1 or r > 50:
|
||||
return 0.0
|
||||
return float(51 - r)
|
||||
|
||||
|
||||
def parse_digg_response(
|
||||
response: Dict[str, Any],
|
||||
query: str = "",
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Parse a digg search envelope into normalized item dicts.
|
||||
|
||||
Args:
|
||||
response: payload from ``search_digg``.
|
||||
query: original search query, used for token-overlap relevance.
|
||||
|
||||
Returns:
|
||||
List of dicts ready for ``normalize._normalize_digg``.
|
||||
"""
|
||||
raw = response.get("results") if isinstance(response, dict) else None
|
||||
if not isinstance(raw, list):
|
||||
return []
|
||||
|
||||
items: List[Dict[str, Any]] = []
|
||||
for i, cluster in enumerate(raw):
|
||||
if not isinstance(cluster, dict):
|
||||
continue
|
||||
cluster_url_id = cluster.get("clusterUrlId")
|
||||
if not cluster_url_id:
|
||||
continue
|
||||
|
||||
title = str(cluster.get("title") or "").strip()
|
||||
tldr = str(cluster.get("tldr") or "").strip()
|
||||
rank = cluster.get("rank")
|
||||
post_count = cluster.get("postCount") or 0
|
||||
unique_authors = cluster.get("uniqueAuthors") or 0
|
||||
first_post_age = cluster.get("firstPostAge")
|
||||
date_str = _parse_first_post_age(first_post_age)
|
||||
if date_str is None and first_post_age:
|
||||
# firstPostAge present but outside 30d -> drop; last30days contract.
|
||||
continue
|
||||
|
||||
rank_decay = max(0.3, 1.0 - (i * 0.02))
|
||||
if query:
|
||||
content_score = token_overlap_relevance(query, f"{title} {tldr}".strip())
|
||||
else:
|
||||
content_score = 0.5
|
||||
rank_boost = min(0.2, _rank_score(rank) / 250.0)
|
||||
relevance = min(1.0, 0.55 * rank_decay + 0.35 * content_score + rank_boost)
|
||||
|
||||
items.append(
|
||||
{
|
||||
"id": str(cluster_url_id),
|
||||
"title": title or f"Digg cluster {i + 1}",
|
||||
"url": _build_url(str(cluster_url_id)),
|
||||
"tldr": tldr,
|
||||
"author": "",
|
||||
"date": date_str,
|
||||
"engagement": {
|
||||
"postCount": int(post_count) if isinstance(post_count, (int, float)) else 0,
|
||||
"uniqueAuthors": int(unique_authors) if isinstance(unique_authors, (int, float)) else 0,
|
||||
"rank": int(rank) if isinstance(rank, (int, float)) else None,
|
||||
"rank_score": _rank_score(rank),
|
||||
},
|
||||
"first_post_age": first_post_age,
|
||||
"posts": [],
|
||||
"relevance": round(relevance, 2),
|
||||
"why_relevant": (
|
||||
f"Digg AI 1000 cluster (rank {rank}, {post_count} posts, {unique_authors} authors)"
|
||||
if rank is not None
|
||||
else f"Digg AI 1000 cluster ({post_count} posts, {unique_authors} authors)"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def _parse_post(raw_post: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
"""Reduce a digg post payload into the small dict render uses.
|
||||
|
||||
We deliberately keep this minimal: an inline quote needs the author
|
||||
handle, the body, the post type, and the X URL.
|
||||
"""
|
||||
if not isinstance(raw_post, dict):
|
||||
return None
|
||||
body = str(raw_post.get("body") or "").strip()
|
||||
if not body:
|
||||
return None
|
||||
author = raw_post.get("author") or {}
|
||||
if not isinstance(author, dict):
|
||||
author = {}
|
||||
username = str(author.get("username") or "").strip()
|
||||
if not username:
|
||||
return None
|
||||
x_url = str(raw_post.get("xUrl") or "").strip()
|
||||
if not x_url:
|
||||
return None
|
||||
return {
|
||||
"username": username,
|
||||
"display_name": str(author.get("display_name") or "").strip() or username,
|
||||
"category": str(author.get("category") or "").strip(),
|
||||
"rank": author.get("rank"),
|
||||
"body": body,
|
||||
"post_type": str(raw_post.get("post_type") or "tweet").strip(),
|
||||
"x_url": x_url,
|
||||
"posted_at": raw_post.get("posted_at"),
|
||||
}
|
||||
|
||||
|
||||
def fetch_top_posts(cluster_url_id: str, posts_per: int = POSTS_PER_CLUSTER) -> List[Dict[str, Any]]:
|
||||
"""Fetch top-ranked X posts attached to a cluster.
|
||||
|
||||
Returns an empty list on any failure (timeout, missing cluster, JSON
|
||||
error). Never raises.
|
||||
"""
|
||||
if posts_per <= 0:
|
||||
return []
|
||||
cmd = _build_posts_args(cluster_url_id, posts_per)
|
||||
response = _run_cli(cmd, timeout=POSTS_TIMEOUT)
|
||||
raw = response.get("results") or []
|
||||
out: List[Dict[str, Any]] = []
|
||||
for entry in raw:
|
||||
post = _parse_post(entry)
|
||||
if post is not None:
|
||||
out.append(post)
|
||||
return out
|
||||
|
||||
|
||||
def enrich_with_top_posts(
|
||||
items: List[Dict[str, Any]],
|
||||
top_k: int = 3,
|
||||
posts_per: int = POSTS_PER_CLUSTER,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Attach top X posts to the first ``top_k`` clusters by Digg rank order.
|
||||
|
||||
Mutates and returns the same list. Items that already have posts, or
|
||||
whose ``postCount`` is 0, are skipped.
|
||||
"""
|
||||
if top_k <= 0 or posts_per <= 0:
|
||||
return items
|
||||
enriched = 0
|
||||
for item in items:
|
||||
if enriched >= top_k:
|
||||
break
|
||||
if item.get("posts"):
|
||||
continue
|
||||
engagement = item.get("engagement") or {}
|
||||
if not engagement.get("postCount"):
|
||||
continue
|
||||
cluster_url_id = item.get("id")
|
||||
if not cluster_url_id:
|
||||
continue
|
||||
posts = fetch_top_posts(str(cluster_url_id), posts_per=posts_per)
|
||||
item["posts"] = posts
|
||||
enriched += 1
|
||||
if enriched:
|
||||
_log(f"enriched {enriched} clusters with X posts")
|
||||
return items
|
||||
|
||||
|
||||
def enrich_source_items(items: list, top_k: int = 3, posts_per: int = POSTS_PER_CLUSTER) -> list:
|
||||
"""Attach top X posts to the first ``top_k`` SourceItems that survived dedupe.
|
||||
|
||||
Reads ``metadata['clusterUrlId']`` and writes ``metadata['posts']`` in
|
||||
place. Skips items that already carry a non-empty ``metadata['posts']``,
|
||||
items whose engagement ``postCount`` is 0, and items whose source is not
|
||||
'digg'. Designed to run from `_finalize_items_by_source` so enrichment
|
||||
is spent on the items the brief actually shows.
|
||||
"""
|
||||
if top_k <= 0 or posts_per <= 0:
|
||||
return items
|
||||
enriched = 0
|
||||
for item in items:
|
||||
if enriched >= top_k:
|
||||
break
|
||||
if getattr(item, "source", None) != "digg":
|
||||
continue
|
||||
metadata = getattr(item, "metadata", None) or {}
|
||||
if metadata.get("posts"):
|
||||
continue
|
||||
engagement = getattr(item, "engagement", None) or {}
|
||||
if not engagement.get("postCount"):
|
||||
continue
|
||||
cluster_url_id = metadata.get("clusterUrlId") or item.item_id
|
||||
if not cluster_url_id:
|
||||
continue
|
||||
posts = fetch_top_posts(str(cluster_url_id), posts_per=posts_per)
|
||||
if posts:
|
||||
metadata["posts"] = posts
|
||||
enriched += 1
|
||||
if enriched:
|
||||
_log(f"post-dedupe enriched {enriched} clusters with X posts")
|
||||
return items
|
||||
@@ -49,6 +49,7 @@ def normalize_source_items(
|
||||
"xquik": _normalize_x,
|
||||
"pinterest": _normalize_pinterest,
|
||||
"polymarket": _normalize_polymarket,
|
||||
"digg": _normalize_digg,
|
||||
"grounding": _normalize_grounding,
|
||||
"xiaohongshu": _normalize_grounding,
|
||||
"github": _normalize_github,
|
||||
@@ -399,6 +400,53 @@ def _normalize_microblog(
|
||||
)
|
||||
|
||||
|
||||
def _normalize_digg(
|
||||
source: str,
|
||||
item: dict[str, Any],
|
||||
index: int,
|
||||
from_date: str,
|
||||
to_date: str,
|
||||
) -> schema.SourceItem:
|
||||
"""Normalizer for Digg AI 1000 clusters.
|
||||
|
||||
Each cluster is one item. The TLDR carries the most useful body for
|
||||
rerank and synthesis. Top-ranked X posts attached at search time are
|
||||
passed through under metadata['posts'] so render can emit them as
|
||||
inline 'via Digg AI 1000' quotes.
|
||||
"""
|
||||
title = str(item.get("title") or "").strip()
|
||||
tldr = str(item.get("tldr") or "").strip()
|
||||
body = "\n\n".join(part for part in [title, tldr] if part)
|
||||
posts = item.get("posts") or []
|
||||
if not isinstance(posts, list):
|
||||
posts = []
|
||||
cluster_url_id = str(item.get("id") or f"DG{index + 1}")
|
||||
return _source_item(
|
||||
item_id=cluster_url_id,
|
||||
source=source,
|
||||
title=title or f"Digg cluster {index + 1}",
|
||||
body=body,
|
||||
url=str(item.get("url") or f"https://di.gg/ai/{cluster_url_id}"),
|
||||
author="",
|
||||
container="Digg AI 1000",
|
||||
published_at=item.get("date"),
|
||||
date_confidence=_date_confidence(item, from_date, to_date, default="high"),
|
||||
engagement=item.get("engagement") or {},
|
||||
relevance_hint=item.get("relevance", 0.5),
|
||||
why_relevant=str(item.get("why_relevant") or ""),
|
||||
snippet=tldr[:400],
|
||||
metadata={
|
||||
"clusterUrlId": cluster_url_id,
|
||||
"tldr": tldr,
|
||||
"rank": (item.get("engagement") or {}).get("rank"),
|
||||
"uniqueAuthors": (item.get("engagement") or {}).get("uniqueAuthors"),
|
||||
"postCount": (item.get("engagement") or {}).get("postCount"),
|
||||
"firstPostAge": item.get("first_post_age"),
|
||||
"posts": posts,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _normalize_polymarket(
|
||||
source: str,
|
||||
item: dict[str, Any],
|
||||
|
||||
@@ -15,6 +15,7 @@ from . import (
|
||||
bluesky,
|
||||
dates,
|
||||
dedupe,
|
||||
digg,
|
||||
entity_extract,
|
||||
env,
|
||||
github,
|
||||
@@ -79,6 +80,7 @@ MOCK_AVAILABLE_SOURCES = [
|
||||
"github",
|
||||
"perplexity",
|
||||
"xquik",
|
||||
"digg",
|
||||
]
|
||||
|
||||
|
||||
@@ -106,6 +108,8 @@ def available_sources(config: dict[str, Any], requested_sources: list[str] | Non
|
||||
available.extend(["hackernews", "polymarket"])
|
||||
if config.get("GITHUB_TOKEN") or which("gh"):
|
||||
available.append("github")
|
||||
if which("digg-pp-cli"):
|
||||
available.append("digg")
|
||||
if env.is_bluesky_available(config):
|
||||
available.append("bluesky")
|
||||
if env.is_truthsocial_available(config):
|
||||
@@ -531,6 +535,12 @@ def _finalize_items_by_source(
|
||||
keywords = config.get("_polymarket_keywords") if isinstance(config, dict) else None
|
||||
if keywords:
|
||||
items = polymarket.filter_items_against_keywords(items, keywords)
|
||||
if source == "digg" and items:
|
||||
# Pull top-ranked X posts only for the survivors that will appear
|
||||
# in the brief. Spending the enrichment budget here (rather than
|
||||
# at retrieval time) keeps the inline 'via Digg AI 1000' quotes
|
||||
# paired with the clusters dedupe actually kept.
|
||||
digg.enrich_source_items(items, top_k=3)
|
||||
finalized[source] = items
|
||||
return finalized
|
||||
|
||||
@@ -966,6 +976,13 @@ def _retrieve_stream(
|
||||
if source == "hackernews":
|
||||
result = hackernews.search_hackernews(subquery.search_query, from_date, to_date, depth=depth)
|
||||
return hackernews.parse_hackernews_response(result, query=subquery.search_query), {}
|
||||
if source == "digg":
|
||||
result = digg.search_digg(subquery.search_query, from_date, to_date, depth=depth)
|
||||
items = digg.parse_digg_response(result, query=subquery.search_query)
|
||||
# Enrichment with attached X posts is deferred to
|
||||
# _finalize_items_by_source so it runs on the items that actually
|
||||
# survive dedupe rather than on top-K of the raw fanout.
|
||||
return items, {}
|
||||
if source == "bluesky":
|
||||
result = bluesky.search_bluesky(subquery.search_query, from_date, to_date, depth=depth, config=config)
|
||||
return bluesky.parse_bluesky_response(result), {}
|
||||
@@ -1058,6 +1075,45 @@ def _mock_stream_results(source: str, subquery: schema.SubQuery) -> tuple[list[d
|
||||
"why_relevant": "Brave web search",
|
||||
}
|
||||
],
|
||||
"digg": [
|
||||
{
|
||||
"id": "mock1abc",
|
||||
"title": f"Digg AI 1000 cluster about {subquery.search_query}",
|
||||
"url": "https://di.gg/ai/mock1abc",
|
||||
"tldr": f"Curated cluster summarizing recent {subquery.search_query} discussion across the AI 1000.",
|
||||
"author": "",
|
||||
"date": dates.get_date_range(3)[0],
|
||||
"engagement": {"postCount": 8, "uniqueAuthors": 5, "rank": 2, "rank_score": 49.0},
|
||||
"first_post_age": "3d",
|
||||
"posts": [
|
||||
{
|
||||
"username": "exampledev",
|
||||
"display_name": "Example Dev",
|
||||
"category": "Engineer",
|
||||
"rank": 142,
|
||||
"body": f"Quote from the AI 1000 about {subquery.search_query}.",
|
||||
"post_type": "tweet",
|
||||
"x_url": "https://x.com/exampledev/status/1",
|
||||
"posted_at": dates.get_date_range(3)[0],
|
||||
},
|
||||
],
|
||||
"relevance": 0.84,
|
||||
"why_relevant": "Mock Digg cluster",
|
||||
},
|
||||
{
|
||||
"id": "mock2def",
|
||||
"title": f"Second Digg cluster on {subquery.search_query}",
|
||||
"url": "https://di.gg/ai/mock2def",
|
||||
"tldr": f"Another angle on {subquery.search_query}.",
|
||||
"author": "",
|
||||
"date": dates.get_date_range(8)[0],
|
||||
"engagement": {"postCount": 3, "uniqueAuthors": 2, "rank": 18, "rank_score": 33.0},
|
||||
"first_post_age": "8d",
|
||||
"posts": [],
|
||||
"relevance": 0.71,
|
||||
"why_relevant": "Mock Digg cluster",
|
||||
},
|
||||
],
|
||||
}
|
||||
if source == "grounding":
|
||||
return payloads.get(source, []), {
|
||||
|
||||
@@ -67,6 +67,7 @@ SOURCE_CAPABILITIES = {
|
||||
"bluesky": {"discussion", "social"},
|
||||
"truthsocial": {"discussion", "social"},
|
||||
"polymarket": {"market"},
|
||||
"digg": {"discussion", "social", "link"},
|
||||
"xiaohongshu": {"video", "video_shortform", "social"},
|
||||
"github": {"discussion", "link"},
|
||||
"grounding": {"web", "reference", "link"},
|
||||
|
||||
@@ -53,6 +53,7 @@ SOURCE_LABELS = {
|
||||
"xiaohongshu": "Xiaohongshu",
|
||||
"x": "X",
|
||||
"github": "GitHub",
|
||||
"digg": "Digg AI 1000",
|
||||
"perplexity": "Perplexity",
|
||||
}
|
||||
|
||||
@@ -826,7 +827,7 @@ def render_full(report: schema.Report) -> str:
|
||||
lines.append("## All Items by Source")
|
||||
lines.append("")
|
||||
source_order = ["reddit", "x", "youtube", "tiktok", "instagram", "threads", "pinterest",
|
||||
"hackernews", "bluesky", "truthsocial", "polymarket", "grounding", "xiaohongshu", "github", "perplexity"]
|
||||
"hackernews", "bluesky", "truthsocial", "polymarket", "grounding", "xiaohongshu", "github", "digg", "perplexity"]
|
||||
for source in source_order:
|
||||
items = report.items_by_source.get(source, [])
|
||||
if not items:
|
||||
@@ -852,6 +853,9 @@ def render_full(report: schema.Report) -> str:
|
||||
tc_score = tc.get("score", "")
|
||||
attribution = _comment_attribution(item.source, tc.get("author"))
|
||||
lines.append(f" Top comment {attribution} ({tc_score} {vote_label}): {excerpt}")
|
||||
# Digg AI 1000: inline X-post quotes attached to the cluster.
|
||||
for post in _digg_posts_for(item, limit=3):
|
||||
lines.append(f" > {_format_digg_quote(post)}")
|
||||
# Comment insights for Reddit
|
||||
insights = item.metadata.get("comment_insights", [])
|
||||
if insights:
|
||||
@@ -973,6 +977,8 @@ def _render_candidate(candidate: schema.Candidate, prefix: str) -> list[str]:
|
||||
source = primary.source if primary else None
|
||||
attribution = _comment_attribution(source, tc.get("author"))
|
||||
lines.append(f" - {attribution} ({score} {vote_label}): {_truncate(excerpt.strip(), 240)}")
|
||||
for post in _digg_posts_for(primary):
|
||||
lines.append(f" - {_format_digg_quote(post)}")
|
||||
insight = _comment_insight(primary)
|
||||
if insight:
|
||||
lines.append(f" - Insight: {_truncate(insight, 220)}")
|
||||
@@ -1223,6 +1229,7 @@ _FOOTER_SOURCES: list[tuple[str, str, str, str, list[tuple[str, str]]]] = [
|
||||
("bluesky", "🦋", "Bluesky", "post", [("likes", "likes"), ("reposts", "reposts")]),
|
||||
("truthsocial", "🇺🇸", "Truth Social", "post", [("likes", "likes"), ("reposts", "reposts")]),
|
||||
("github", "🐙", "GitHub", "item", [("reactions", "reactions"), ("comments", "comments")]),
|
||||
("digg", "⛏️", "Digg AI 1000", "cluster", [("postCount", "posts"), ("uniqueAuthors", "authors")]),
|
||||
]
|
||||
|
||||
|
||||
@@ -1480,6 +1487,7 @@ ENGAGEMENT_DISPLAY: dict[str, list[tuple[str, str]]] = {
|
||||
"polymarket": [],
|
||||
"github": [("reactions", "react"), ("comments", "cmt")],
|
||||
"perplexity": [("citations", "cite")],
|
||||
"digg": [("postCount", "posts"), ("uniqueAuthors", "auth")],
|
||||
}
|
||||
|
||||
|
||||
@@ -1676,6 +1684,39 @@ def _comment_insight(item: schema.SourceItem | None) -> str | None:
|
||||
return str(insights[0]).strip() or None
|
||||
|
||||
|
||||
def _digg_posts_for(item: schema.SourceItem | None, limit: int = 2) -> list[dict]:
|
||||
"""Return up to `limit` parsed Digg posts attached as enrichment to a cluster.
|
||||
|
||||
Returns an empty list for non-digg sources or clusters without enrichment.
|
||||
"""
|
||||
if not item or item.source != "digg":
|
||||
return []
|
||||
posts = item.metadata.get("posts") or []
|
||||
if not isinstance(posts, list):
|
||||
return []
|
||||
out: list[dict] = []
|
||||
for entry in posts:
|
||||
if isinstance(entry, dict) and entry.get("body") and entry.get("username"):
|
||||
out.append(entry)
|
||||
if len(out) >= limit:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _format_digg_quote(post: dict, body_limit: int = 200) -> str:
|
||||
"""Format a Digg-attached X post as an inline 'via Digg AI 1000' quote line."""
|
||||
handle = post.get("username") or ""
|
||||
x_url = post.get("x_url") or ""
|
||||
body = (post.get("body") or "").replace("\n", " ").strip()
|
||||
if len(body) > body_limit:
|
||||
body = body[: body_limit - 1].rstrip() + "…"
|
||||
if x_url and handle:
|
||||
return f"[@{handle}]({x_url}) via Digg AI 1000: {body}"
|
||||
if handle:
|
||||
return f"@{handle} via Digg AI 1000: {body}"
|
||||
return f"via Digg AI 1000: {body}"
|
||||
|
||||
|
||||
def _transcript_highlights(item: schema.SourceItem | None) -> list[str]:
|
||||
if not item or item.source != "youtube":
|
||||
return []
|
||||
|
||||
@@ -12,6 +12,7 @@ SOURCE_QUALITY = {
|
||||
"xiaohongshu": 0.7,
|
||||
"hackernews": 0.8,
|
||||
"youtube": 0.85,
|
||||
"digg": 0.85,
|
||||
"reddit": 0.6,
|
||||
"x": 0.68,
|
||||
"bluesky": 0.66,
|
||||
@@ -95,6 +96,7 @@ ENGAGEMENT_WEIGHTS: dict[str, list[tuple[str, float]]] = {
|
||||
"bluesky": [("likes", 0.40), ("reposts", 0.30), ("replies", 0.20), ("quotes", 0.10)],
|
||||
"truthsocial": [("likes", 0.45), ("reposts", 0.30), ("replies", 0.25)],
|
||||
"polymarket": [("volume", 0.60), ("liquidity", 0.40)],
|
||||
"digg": [("postCount", 0.40), ("uniqueAuthors", 0.30), ("rank_score", 0.30)],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -124,6 +124,7 @@ SOURCE_COMPLETION_ORDER = [
|
||||
"polymarket",
|
||||
"grounding",
|
||||
"xiaohongshu",
|
||||
"digg",
|
||||
]
|
||||
|
||||
SOURCE_COMPLETION_META = {
|
||||
@@ -138,6 +139,7 @@ SOURCE_COMPLETION_META = {
|
||||
"polymarket": ("Polymarket", "market", "markets", Colors.GREEN),
|
||||
"grounding": ("Web", "result", "results", Colors.GREEN),
|
||||
"xiaohongshu": ("Xiaohongshu", "post", "posts", Colors.RED),
|
||||
"digg": ("Digg", "cluster", "clusters", Colors.YELLOW),
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user