Task 5: demand-aware ingestion funnel

zeeguu/api/endpoints/topics.py

@@ -43,6 +43,15 @@ def subscribe_to_topic_with_id():
     user = User.find_by_id(flask.g.user_id)
     TopicSubscription.find_or_create(db_session, user, topic_object)
     db_session.commit()
+
+    # Promotion-on-join (Task 5): give this newly-demanded bucket an immediate
+    # head start so the reader sees their topic without waiting for the next
+    # crawl. Runs in the background; no-ops if the bucket already has inventory.
+    from zeeguu.api.utils.background import run_in_background
+    from zeeguu.core.content_retriever.backfill import maybe_backfill_bucket
+
+    run_in_background(maybe_backfill_bucket, user.id, topic_id)
+
     return "OK"
 
 

zeeguu/api/endpoints/user.py

@@ -58,6 +58,15 @@ def learned_language_set(language_code):
     user = User.find_by_id(flask.g.user_id)
     user.set_learned_language(language_code, session=zeeguu.core.model.db.session)
     zeeguu.core.model.db.session.commit()
+
+    # Promotion-on-join (Task 5): a reader who just switched to this language
+    # shouldn't wait for the next crawl. Backfill a few articles across the
+    # language (no specific topic) in the background; no-ops if already stocked.
+    from zeeguu.api.utils.background import run_in_background
+    from zeeguu.core.content_retriever.backfill import maybe_backfill_bucket
+
+    run_in_background(maybe_backfill_bucket, user.id)
+
     return "OK"
 
 

zeeguu/core/content_retriever/article_downloader.py

@@ -285,8 +285,60 @@ def extract_article_image(np_article):
         return ""
 
 
+def _apply_title_triage(feed, items, limit, funnel_budget):
+    """Prune a feed's candidate items to the best ones worth downloading.
+
+    See the demand-aware funnel (funnel.py, phase 3). Returns the (possibly
+    shorter, possibly reordered) list of feed items to process. Never raises:
+    on any trouble it returns the original list so the crawl proceeds unchanged.
+    """
+    from zeeguu.core.content_retriever.funnel import (
+        triage_keep_count,
+        select_titles_to_download,
+    )
+    from zeeguu.core.model import Topic
+
+    items_list = list(items)
+    try:
+        headroom = funnel_budget.language_simplification_headroom(feed.language_id)
+        keep = triage_keep_count(headroom, limit)
+
+        if keep <= 0:
+            log(
+                f"   ⏭ Skipping feed '{feed.title}' - no simplification budget left "
+                f"today for {feed.language.code}"
+            )
+            return []
+
+        if len(items_list) <= keep:
+            return items_list
+
+        demanded_ids = funnel_budget.demand.demanded_topic_ids(feed.language_id)
+        demand_topic_names = []
+        for topic_id in demanded_ids:
+            topic = Topic.find_by_id(topic_id)
+            if topic:
+                demand_topic_names.append(topic.title)
+
+        titles = [feed_item.get("title", "") for feed_item in items_list]
+        kept_indices = select_titles_to_download(
+            titles, feed.language.code, demand_topic_names, keep
+        )
+        log(
+            f"   Title triage: keeping {len(kept_indices)}/{len(items_list)} candidates "
+            f"for {feed.language.code} (headroom={headroom})"
+        )
+        return [items_list[i] for i in kept_indices]
+    except Exception as e:
+        # Triage is a best-effort optimization: on any trouble, fall back to the
+        # full candidate list so the crawl proceeds exactly as it would without
+        # the funnel, rather than dropping the whole feed.
+        log(f"   ⚠ Title triage errored ({e}); processing full feed")
+        return items_list
+
+
 def download_from_feed(
-    feed: Feed, session, crawl_report, limit=1000, save_in_elastic=True, simplification_provider=None, topic_simplification_counts=None
+    feed: Feed, session, crawl_report, limit=1000, save_in_elastic=True, simplification_provider=None, funnel_budget=None
 ):
     """
 
@@ -338,6 +390,14 @@ def download_from_feed(
 
     skipped_already_in_db = 0
 
+    # Pre-download title triage (Task 5, phase 3). Before paying any readability /
+    # download cost, let the funnel prune this feed's candidates: skip the feed
+    # entirely when its language is already satisfied for the day, or keep only
+    # the best-N titles when headroom is limited. Falls back to the full list on
+    # any problem, so this can only reduce waste, never drop must-have articles.
+    if funnel_budget is not None:
+        items = _apply_title_triage(feed, items, limit, funnel_budget)
+
     # Arm the per-article watchdog. signal.signal() only works on the main thread;
     # the crawler is single-threaded, but guard anyway so an unexpected non-crawler
     # caller degrades gracefully (the shell-level `timeout` backstop still applies).
@@ -419,7 +479,7 @@ def download_from_feed(
                 url,
                 crawl_report,
                 simplification_provider=simplification_provider,
-                topic_simplification_counts=topic_simplification_counts,
+                funnel_budget=funnel_budget,
             )
             # The article is fetched + saved; disarm now so the alarm can't fire
             # during the ES-indexing/bookkeeping below — a timeout there would
@@ -614,7 +674,7 @@ def get_todays_simplified_counts_by_language_topic(session, language_id):
     return {topic_id: count for topic_id, count in results}
 
 
-def download_feed_item(session, feed, feed_item, url, crawl_report, simplification_provider=None, topic_simplification_counts=None):
+def download_feed_item(session, feed, feed_item, url, crawl_report, simplification_provider=None, funnel_budget=None):
     import html
 
     title = html.unescape(feed_item["title"])
@@ -757,27 +817,23 @@ def download_feed_item(session, feed, feed_item, url, crawl_report, simplificati
         )
         _save_classifications(session, new_article, [("DISTURBING", "KEYWORD")])
 
-    # Check topic simplification cap - skip simplification if all topics are "full" for this language today
-    # topic_simplification_counts is keyed by (language_id, topic_id) tuple
-    # Note: article_topic_ids and article_topic_names were captured earlier before potential rollback
-    skip_simplification_due_to_cap = False
+    # Demand-aware funnel gate (Task 5): does this article earn a simplification?
+    # The funnel replaces the old static per-language cap with a per-(language,
+    # topic) quota derived from what active readers actually want, plus a
+    # pilot-light floor for dormant supported languages. See funnel.py.
+    # article_topic_ids / article_topic_names were captured earlier, before any
+    # potential rollback.
     language_id = feed.language_id
-    lang_code = feed.language.code
-    max_for_lang = get_max_simplified_for_language(lang_code)
-
-    if topic_simplification_counts is not None and article_topic_ids:
-        # Check if ANY topic still needs simplified articles today for this language
-        needs_simplification = False
-        for topic_id in article_topic_ids:
-            key = (language_id, topic_id)
-            current_count = topic_simplification_counts.get(key, 0)
-            if current_count < max_for_lang:
-                needs_simplification = True
-                break
-
-        if not needs_simplification:
-            skip_simplification_due_to_cap = True
-            log(f"   ⏭ Skipping simplification - daily cap ({max_for_lang}/topic) reached for: {article_topic_names}")
+    simplify_reason = None
+
+    if funnel_budget is not None:
+        should_simplify, simplify_reason = funnel_budget.should_simplify(
+            language_id, article_topic_ids
+        )
+        if not should_simplify:
+            log(
+                f"   ⏭ Skipping simplification ({simplify_reason}) for topics: {article_topic_names}"
+            )
             return new_article
 
     # Auto-create simplified versions and classify content
@@ -798,12 +854,11 @@ def download_feed_item(session, feed, feed_item, url, crawl_report, simplificati
             # here keeps that off the request path (mirrors the original above).
             for simplified in simplified_articles:
                 _cache_article_tokenization(simplified, session)
-            # Update topic simplification counts after successful simplification
-            # Key is (language_id, topic_id) to track per language
-            if topic_simplification_counts is not None:
-                for topic_id in article_topic_ids:
-                    key = (language_id, topic_id)
-                    topic_simplification_counts[key] = topic_simplification_counts.get(key, 0) + 1
+            # Book this simplification against the funnel budget so the running
+            # per-(language, topic) counts (and the floor's topic-diversity set)
+            # reflect it for the rest of the crawl.
+            if funnel_budget is not None:
+                funnel_budget.record(language_id, article_topic_ids, simplify_reason)
         else:
             log(
                 f"   No simplified versions created"

zeeguu/core/content_retriever/backfill.py

@@ -0,0 +1,153 @@
+"""Promotion-on-join backfill (Task 5, phase 5).
+
+When a reader opens up demand the last crawl didn't know about -- subscribes to a
+topic, or picks a new learned language -- the demand-aware funnel will start
+stocking that ``(language, topic)`` bucket, but only at the *next* crawl (up to
+an hour away, more for low-frequency languages). The pilot-light floor keeps the
+feed non-empty meanwhile, but a reader who just asked for a topic wants to see it
+*now*.
+
+Key insight: a dormant bucket is not empty of articles, only of *simplified*
+ones. The crawler still downloads and stores the raw article; it just skips the
+(expensive) simplification when the bucket has no demand. So the immediate
+backfill doesn't need to crawl anything -- it simplifies a few of the raw
+articles that are already sitting in the DB for this bucket. That's fast (LLM
+only, no readability/download) and targeted.
+
+Runs in a background thread (``run_in_background``) so it never blocks the
+request; conservative caps keep it cheap.
+"""
+
+from datetime import datetime, timedelta
+
+from sqlalchemy import or_
+from sqlalchemy.orm import aliased
+
+import zeeguu.core
+from zeeguu.logging import log
+
+
+def _not_broken(article_cls):
+    """Non-broken predicate: ``broken`` is 0 for healthy articles, but tolerate
+    legacy NULLs too (the column is nullable)."""
+    return or_(article_cls.broken == 0, article_cls.broken.is_(None))
+
+# If the bucket already has at least this many freshly-simplified articles, the
+# reader won't hit an empty feed and we skip the backfill entirely.
+BACKFILL_FRESH_DAYS = 3
+BACKFILL_MIN_FRESH = 3
+
+# How far back to look for raw articles to simplify, and how many to add.
+BACKFILL_LOOKBACK_DAYS = 7
+BACKFILL_TARGET = 3
+
+
+def fresh_simplified_count(session, language_id, topic_id, days=BACKFILL_FRESH_DAYS):
+    """Count freshly-simplified articles available for a bucket.
+
+    Simplified articles are the children (``parent_article_id`` set); topics live
+    on the parent. ``topic_id=None`` counts across the whole language (used when a
+    reader just picked a new learned language rather than a specific topic).
+    """
+    from zeeguu.core.model import Article, ArticleTopicMap
+
+    cutoff = datetime.now() - timedelta(days=days)
+    ParentArticle = aliased(Article)
+
+    q = (
+        session.query(Article.id)
+        .join(ParentArticle, Article.parent_article_id == ParentArticle.id)
+        .filter(ParentArticle.language_id == language_id)
+        # crawled_at is when the simplified child was created; published_time is
+        # inherited from the (backdatable) parent, so it can't measure freshness.
+        .filter(Article.crawled_at >= cutoff)
+        .filter(_not_broken(Article))
+    )
+    if topic_id is not None:
+        q = q.join(
+            ArticleTopicMap, ArticleTopicMap.article_id == ParentArticle.id
+        ).filter(ArticleTopicMap.topic_id == topic_id)
+    return q.count()
+
+
+def recent_unsimplified_articles(
+    session, language_id, topic_id, days=BACKFILL_LOOKBACK_DAYS, limit=BACKFILL_TARGET
+):
+    """Recent raw (original, non-broken) articles for a bucket that have no
+    simplified child yet -- the backfill's raw material, newest first."""
+    from zeeguu.core.model import Article, ArticleTopicMap
+
+    cutoff = datetime.now() - timedelta(days=days)
+    Child = aliased(Article)
+
+    q = (
+        session.query(Article)
+        .outerjoin(Child, Child.parent_article_id == Article.id)
+        .filter(Article.parent_article_id.is_(None))  # originals only
+        .filter(Child.id.is_(None))  # not yet simplified
+        .filter(Article.language_id == language_id)
+        # Recency = when we ingested it, not its (backdatable) publish date, so a
+        # recently-crawled but old-dated article is still eligible backfill material.
+        .filter(Article.crawled_at >= cutoff)
+        .filter(_not_broken(Article))
+    )
+    if topic_id is not None:
+        q = q.join(
+            ArticleTopicMap, ArticleTopicMap.article_id == Article.id
+        ).filter(ArticleTopicMap.topic_id == topic_id)
+
+    return q.order_by(Article.crawled_at.desc()).limit(limit).all()
+
+
+def maybe_backfill_bucket(user_id, topic_id=None, provider="deepseek"):
+    """Background entry point: give a just-joined reader's bucket a head start.
+
+    Re-queries everything by id (runs in its own thread + app context). No-ops
+    quietly if the bucket already has fresh inventory or there's nothing raw to
+    simplify -- the next crawl and the pilot-light floor cover those cases.
+    """
+    from zeeguu.core.model import User
+    from zeeguu.core.llm_services.simplification_and_classification import (
+        simplify_and_classify,
+    )
+
+    session = zeeguu.core.model.db.session
+
+    user = User.find_by_id(user_id)
+    if not user or not user.learned_language:
+        return
+    language = user.learned_language
+
+    fresh = fresh_simplified_count(session, language.id, topic_id)
+    if fresh >= BACKFILL_MIN_FRESH:
+        log(
+            f"[backfill] {language.code}/topic={topic_id}: {fresh} fresh simplified "
+            f"already available; skipping"
+        )
+        return
+
+    candidates = recent_unsimplified_articles(session, language.id, topic_id)
+    if not candidates:
+        log(
+            f"[backfill] {language.code}/topic={topic_id}: no raw articles to "
+            f"simplify; next crawl + floor will cover it"
+        )
+        return
+
+    simplified = 0
+    for article in candidates:
+        try:
+            children, _classifications = simplify_and_classify(
+                session, article, simplification_provider=provider
+            )
+            if children:
+                simplified += 1
+            session.commit()
+        except Exception as e:
+            session.rollback()
+            log(f"[backfill] simplification failed for article {article.id}: {e}")
+
+    log(
+        f"[backfill] {language.code}/topic={topic_id}: simplified {simplified} "
+        f"article(s) (had {fresh} fresh, targeted {BACKFILL_TARGET})"
+    )