Improve Asteria headline board quality

Author: sametbasbug

news_pipeline/news_pipeline/cli/commands/heartbeat_prepare_one.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 import re
 from typing import Any
+from urllib.parse import urlparse
 
 import typer
 from rapidfuzz.fuzz import token_set_ratio
@@ -77,6 +78,7 @@
 }
 BLOCKED_BOARD_TERMS = {
     "celebrity",
+    "concert",
     "rod stewart",
     "ratbag",
     "opinion",
@@ -88,6 +90,68 @@
     "eurovision entry",
     "music museum",
     "father ted",
+    "rapper",
+    "red card controversy",
+    "football chiefs",
+    "training grounds",
+}
+SPONSORED_URL_MARKERS = {
+    "/sponsored-content/",
+    "/brandstudio/",
+    "/paid-content/",
+    "/partner-content/",
+}
+UNREADABLE_PRIMARY_HOSTS = {
+    "politico.com",
+    "www.politico.com",
+}
+TURKISH_READER_SIGNAL_TERMS = {
+    "tariff",
+    "tariffs",
+    "trade",
+    "sanctions",
+    "inflation",
+    "market",
+    "markets",
+    "energy",
+    "oil",
+    "gas",
+    "defense",
+    "defence",
+    "security",
+    "nato",
+    "eu",
+    "europe",
+    "ukraine",
+    "russia",
+    "china",
+    "trump",
+    "ai",
+    "artificial intelligence",
+    "chip",
+    "chips",
+    "climate",
+    "heatwave",
+    "election",
+    "migration",
+    "central bank",
+    "interest rates",
+    "turkey",
+    "türkiye",
+}
+LOW_PUBLIC_VALUE_ODDITY_TERMS = {
+    "snake",
+    "snakes",
+    "cobra",
+    "cobras",
+    "venomous",
+    "escaped",
+    "rapper",
+    "concert",
+    "kanye",
+    "football chiefs",
+    "red card",
+    "infantino",
 }
 
 
@@ -164,6 +228,24 @@ def _source_name(item: Any) -> str:
     return item.draft_sources[0].name if item.draft_sources else "-"
 
 
+def _source_url(item: Any) -> str:
+    return str(item.draft_sources[0].url) if item.draft_sources else ""
+
+
+def _source_host(item: Any) -> str:
+    host = urlparse(_source_url(item)).netloc.lower()
+    return host[4:] if host.startswith("www.") else host
+
+
+def _is_sponsored_source(item: Any) -> bool:
+    url = _source_url(item).lower()
+    return any(marker in url for marker in SPONSORED_URL_MARKERS)
+
+
+def _has_known_unreadable_primary_source(item: Any) -> bool:
+    return _source_host(item) in UNREADABLE_PRIMARY_HOSTS
+
+
 def _frontmatter_value(text: str, key: str) -> str:
     match = re.search(rf"(?m)^{re.escape(key)}:\s*[\"']?(.*?)[\"']?\s*$", text)
     return match.group(1).strip() if match else ""
@@ -329,6 +411,32 @@ def _headline_has_term(headline: str, term: str) -> bool:
     return term in headline
 
 
+def _public_value_text(root: Path, item: Any) -> str:
+    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
+    article = normalized_store.load(item.normalized_id)
+    parts = [
+        item.draft_title or "",
+        item.draft_description or "",
+        " ".join(item.draft_tags or []),
+        _source_name(item),
+    ]
+    if article:
+        parts.extend([article.title or "", article.summary or "", article.content_snippet or "", " ".join(article.tags or [])])
+    return _normalized(" ".join(parts))
+
+
+def _has_turkish_reader_signal(text: str) -> bool:
+    return any(_headline_has_term(text, term) for term in TURKISH_READER_SIGNAL_TERMS)
+
+
+def _has_low_public_value_oddity(text: str) -> bool:
+    return any(_headline_has_term(text, term) for term in LOW_PUBLIC_VALUE_ODDITY_TERMS)
+
+
+def _has_board_hard_veto(reasons: list[str]) -> bool:
+    return any(reason.startswith("public_value_penalty:") for reason in reasons)
+
+
 def _board_score(
     root: Path,
     item: Any,
@@ -338,6 +446,7 @@ def _board_score(
     recent_posts: list[dict[str, str]] | None = None,
 ) -> tuple[float, list[str]]:
     headline = _normalized(_headline_text(root, item))
+    public_value_text = _public_value_text(root, item)
     category = item.draft_category or ""
     source = _normalized(_source_name(item))
     score = float(item.editorial_priority)
@@ -371,6 +480,7 @@ def _board_score(
             repeated = "+".join(sorted(repeat_counts))
             reasons.append(f"recency_penalty:company_repeat:{repeated}:{strongest_repeat}")
     family_hits = _item_topic_family_hits(root, item)
+    has_recent_family_repeat = False
     if family_hits:
         recent_family_counts = recent_live_topic_family_counts(
             root / "src/content/equinoxHaber",
@@ -382,6 +492,7 @@ def _board_score(
             if recent_family_counts[family] >= RECENT_TOPIC_FAMILY_PENALTY_THRESHOLD
         }
         if repeat_counts:
+            has_recent_family_repeat = True
             strongest_repeat = max(repeat_counts.values())
             family_penalty = min(RECENT_TOPIC_FAMILY_PENALTY_MAX, strongest_repeat * RECENT_TOPIC_FAMILY_PENALTY_PER_ITEM)
             score -= family_penalty
@@ -402,6 +513,15 @@ def _board_score(
             score += 0.035
             reasons.append(f"signal_boost:{term}")
             break
+    if _has_turkish_reader_signal(public_value_text) and not localized_crime and not incidental_security and not has_recent_family_repeat:
+        score += 0.045
+        reasons.append("signal_boost:turkish_reader_interest")
+    if _has_low_public_value_oddity(public_value_text):
+        oddity_penalty = 0.18
+        if _has_turkish_reader_signal(public_value_text):
+            oddity_penalty = 0.10
+        score -= oddity_penalty
+        reasons.append("public_value_penalty:oddity_or_entertainment")
     if science_pressure and category == "Bilim":
         score -= SCIENCE_RECENT_PENALTY
         reasons.append("recency_penalty:science_saturation")
@@ -424,6 +544,10 @@ def _passes_basic_board_filter(root: Path, item: Any, max_source_age_hours: int,
     live_urls = live_urls or set()
     if any(str(source.url) in live_urls for source in item.draft_sources):
         return False, "source already published"
+    if _is_sponsored_source(item):
+        return False, "sponsored or paid content"
+    if _has_known_unreadable_primary_source(item):
+        return False, "known unreadable primary source"
     fresh, stale_reason = _source_is_fresh(root, item, max_source_age_hours)
     if not fresh:
         return False, stale_reason
@@ -479,6 +603,7 @@ def _select_headline_board(root: Path, items: list[Any], limit: int, max_source_
         eligible.append((item, board_score, reasons))
 
     eligible.sort(key=lambda row: row[1], reverse=True)
+    score_map = {item.queue_id: (score, reasons) for item, score, reasons in eligible}
     selected: list[Any] = []
     selected_ids: set[str] = set()
     selected_headlines: list[str] = []
@@ -489,6 +614,9 @@ def _select_headline_board(root: Path, items: list[Any], limit: int, max_source_
     def add(item: Any) -> bool:
         if item.queue_id in selected_ids:
             return False
+        _, board_reasons = score_map.get(item.queue_id, (round(float(item.editorial_priority), 3), []))
+        if _has_board_hard_veto(board_reasons):
+            return False
         headline = _normalized(_headline_text(root, item))
         if any(token_set_ratio(headline, existing) >= 88 for existing in selected_headlines):
             return False
@@ -531,7 +659,6 @@ def add(item: Any) -> bool:
             break
         add(item)
 
-    score_map = {item.queue_id: (score, reasons) for item, score, reasons in eligible}
     selected.sort(key=lambda item: score_map.get(item.queue_id, (round(float(item.editorial_priority), 3), []))[0], reverse=True)
     diagnostics = {
         "eligibleCount": len(eligible),

news_pipeline/news_pipeline/cli/commands/queue_source_text.py

@@ -6,7 +6,17 @@
 import typer
 
 from news_pipeline.extractors.article_text import ArticleDetails, fetch_article_details
+from news_pipeline.models.article import NormalizedArticle
 from news_pipeline.queue.service import QueueService
+from news_pipeline.storage.json_store import JsonStore
+
+
+def _metadata_fallback_text(article: NormalizedArticle | None, *, max_chars: int) -> str:
+    if article is None:
+        return ""
+    parts = [article.summary or "", article.content_snippet or ""]
+    text = " ".join(part.strip() for part in parts if part and part.strip())
+    return text[:max_chars].strip()
 
 
 def queue_source_text_command(
@@ -35,6 +45,13 @@ def queue_source_text_command(
 
     source = item.draft_sources[source_index]
     details = fetch_article_details(str(source.url), max_paragraphs=max_paragraphs, max_chars=max_chars)
+    normalized = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle).load(item.normalized_id)
+    extraction_status = "full_text" if details.snippet else "empty"
+    text = details.snippet
+    if not text:
+        text = _metadata_fallback_text(normalized, max_chars=max_chars)
+        if text:
+            extraction_status = "metadata_fallback"
     payload = {
         "queueId": item.queue_id,
         "title": item.draft_title,
@@ -44,8 +61,9 @@ def queue_source_text_command(
             "url": str(source.url),
         },
         "publishedAt": details.published_at.isoformat() if details.published_at else None,
-        "textChars": len(details.snippet),
-        "text": details.snippet,
+        "extractionStatus": extraction_status,
+        "textChars": len(text),
+        "text": text,
     }
     if json_output:
         print(json.dumps(payload, ensure_ascii=False, indent=2))

news_pipeline/tests/test_safety_gates.py

@@ -671,6 +671,68 @@ def test_tv_show_source_format_is_excluded_from_publish_and_board(tmp_path: Path
     assert skipped[0]["reason"] == "excluded source format (podcast/liveblog)"
 
 
+def test_prepare_board_excludes_known_unreadable_primary_source(tmp_path: Path) -> None:
+    article = _article(
+        "politico-com-cloudflare",
+        url="https://www.politico.com/news/2026/07/08/trump-spanish-exports-trade-embargo-00990123",
+    )
+    article.source_id = "politico-eu"
+    article.source_name = "Politico Europe"
+    article.title = "US officials compiling menu of Spanish goods as Trump weighs embargo"
+    item = _item(
+        "politico-com-cloudflare",
+        normalized_id=article.id,
+        url=str(article.canonical_url),
+        priority=0.74,
+    )
+    item.draft_sources = [DraftSource(name="Politico Europe", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, skipped, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    assert skipped[0]["reason"] == "known unreadable primary source"
+
+
+def test_prepare_board_excludes_sponsored_content_url(tmp_path: Path) -> None:
+    article = _article(
+        "sponsored-ai",
+        url="https://www.politico.eu/sponsored-content/europes-ai-moment-four-imperatives-for-business-leaders/",
+    )
+    article.source_id = "politico-eu"
+    article.source_name = "Politico Europe"
+    article.title = "Europe's AI moment: Four imperatives for business leaders"
+    item = _item("sponsored-ai", normalized_id=article.id, url=str(article.canonical_url), priority=0.72)
+    item.draft_sources = [DraftSource(name="Politico Europe", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, skipped, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    assert skipped[0]["reason"] == "sponsored or paid content"
+
+
+def test_prepare_board_penalizes_low_public_value_oddity_below_board_floor(tmp_path: Path) -> None:
+    article = _article("snake-flood", url="https://example.org/world/snakes-flood")
+    article.source_id = "guardian-world"
+    article.source_name = "The Guardian World"
+    article.title = "Venomous snakes escape breeding farms in southern China during flooding"
+    article.summary = "Hundreds of cobras and other snakes escaped during flooding in Guangxi."
+    item = _item("snake-flood", normalized_id=article.id, url=str(article.canonical_url), priority=0.687)
+    item.draft_title = article.title
+    item.draft_description = article.summary
+    item.draft_category = "Bilim"
+    item.draft_sources = [DraftSource(name="The Guardian World", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, _, meta = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    score, reasons = meta["scores"][item.queue_id]
+    assert score < heartbeat_prepare_one.MIN_CATEGORY_TARGET_SCORE
+    assert "public_value_penalty:oddity_or_entertainment" in reasons
+
+
 def test_publish_one_retries_next_candidate_after_duplicate_publish_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.chdir(tmp_path)
     first_article = _article("dup-first", url="https://example.org/demo/dup-first")
@@ -1080,6 +1142,27 @@ def test_queue_approve_refuses_rejected_item_without_explicit_recovery(tmp_path:
         queue_approve_command(item.queue_id)
 
 
+def test_queue_source_text_falls_back_to_normalized_context_when_fetch_is_empty(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    article = _article("source-text-fallback", url="https://example.org/demo/source-text-fallback")
+    article.summary = "RSS summary explains the trade and tariff stakes for readers."
+    article.content_snippet = "Normalized snippet keeps enough context when the live article blocks extraction."
+    item = _item("source-text-fallback", normalized_id=article.id, url=str(article.canonical_url))
+    _save_runtime(tmp_path, item, article)
+    monkeypatch.setattr(queue_source_text, "fetch_article_details", lambda *_, **__: queue_source_text.ArticleDetails())
+
+    queue_source_text.queue_source_text_command(item.queue_id, json_output=True)
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["extractionStatus"] == "metadata_fallback"
+    assert payload["textChars"] > 0
+    assert "RSS summary" in payload["text"]
+
+
 
 def test_body_english_gate_allows_turkish_body_with_english_source_name() -> None:
     body = "\n\n".join(