From 56e186706cffe722bb0e884d11b7fd4d737be5dc Mon Sep 17 00:00:00 2001 From: Mircea Lungu Date: Wed, 15 Jul 2026 21:34:43 +0300 Subject: [PATCH 1/2] Task 5: demand-aware ingestion funnel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the crawler's expensive LLM simplification demand-driven instead of supply-driven. Previously simplification ran until a static per-language cap (50/day/topic for da/fr/de, 20 otherwise) and then stopped — so dormant languages were simplified at the same rate as busy ones and first-in-feed-order won the quota. New zeeguu/core/content_retriever/funnel.py: - compute_demand_surface(): the distinct (language, topic) buckets active readers (last_seen <= 30d) actually want, deduped across the cohort; a reader with no topic subscription lends demand to the whole language. - FunnelBudget: per-(language, topic) quota that scales 20->50 with reader count, plus a pilot-light floor (<=5 distinct topics/day) for dormant *supported* languages so a new learner / the kiosk never hits an empty feed. - Pre-download title triage: skip a feed entirely when its language is already satisfied for the day, or keep only the best-N titles (cheap LLM ranker with a safe first-N fallback) when headroom is limited. article_downloader.py + crawl.py thread a FunnelBudget instead of the old static-cap counts dict. Legacy download_from_feed callers pass no budget, so their behaviour is unchanged. No-topic articles are now simplified only when the language has all-topic readers (old code always simplified them). Promotion-on-join (backfill.py): subscribing to a topic or switching learned language fires a background backfill that simplifies a few EXISTING recent raw articles for the newly-demanded bucket (dormant buckets already hold raw, un-simplified articles — the crawl stores them, only skips the LLM), so the reader doesn't wait for the next crawl. No-ops if the bucket already has fresh inventory. Wired into the subscribe and learned-language endpoints. 31 tests in zeeguu/core/test/test_funnel.py (in-memory SQLite). Co-Authored-By: Claude Opus 4.8 --- zeeguu/api/endpoints/topics.py | 9 + zeeguu/api/endpoints/user.py | 9 + .../content_retriever/article_downloader.py | 106 ++++-- zeeguu/core/content_retriever/backfill.py | 149 ++++++++ zeeguu/core/content_retriever/funnel.py | 331 +++++++++++++++++ zeeguu/core/test/test_funnel.py | 347 ++++++++++++++++++ zeeguu/operations/crawler/crawl.py | 56 +-- 7 files changed, 953 insertions(+), 54 deletions(-) create mode 100644 zeeguu/core/content_retriever/backfill.py create mode 100644 zeeguu/core/content_retriever/funnel.py create mode 100644 zeeguu/core/test/test_funnel.py diff --git a/zeeguu/api/endpoints/topics.py b/zeeguu/api/endpoints/topics.py index 3ace93e7d..3d96d012f 100644 --- a/zeeguu/api/endpoints/topics.py +++ b/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" diff --git a/zeeguu/api/endpoints/user.py b/zeeguu/api/endpoints/user.py index ead0c9152..3f5365acd 100644 --- a/zeeguu/api/endpoints/user.py +++ b/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" diff --git a/zeeguu/core/content_retriever/article_downloader.py b/zeeguu/core/content_retriever/article_downloader.py index 40aebcf4b..a02d8ee5b 100644 --- a/zeeguu/core/content_retriever/article_downloader.py +++ b/zeeguu/core/content_retriever/article_downloader.py @@ -285,8 +285,53 @@ 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) + 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["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] + + 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 +383,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 +472,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 +667,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 +810,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 +847,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" diff --git a/zeeguu/core/content_retriever/backfill.py b/zeeguu/core/content_retriever/backfill.py new file mode 100644 index 000000000..328207441 --- /dev/null +++ b/zeeguu/core/content_retriever/backfill.py @@ -0,0 +1,149 @@ +"""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) + .filter(Article.published_time >= 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) + .filter(Article.published_time >= 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.published_time.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})" + ) diff --git a/zeeguu/core/content_retriever/funnel.py b/zeeguu/core/content_retriever/funnel.py new file mode 100644 index 000000000..957bfaa72 --- /dev/null +++ b/zeeguu/core/content_retriever/funnel.py @@ -0,0 +1,331 @@ +"""Demand-aware ingestion funnel (Task 5). + +The crawler used to simplify articles until a *static* per-language cap was hit +(50/day/topic for da/fr/de, 20 for the rest) and then skip the rest. That made +the expensive LLM step supply-driven: German was ~29% of the crawl while barely +anyone read it, and dormant languages were simplified at the same rate as busy +ones. + +This module makes the simplification budget *demand-driven*: + + * The **demand surface** is what active readers actually want -- the distinct + ``(language, topic)`` buckets across the active cohort, deduped (one FR + politics article stocks every FR-politics reader, so demand is per bucket, + not per user). A reader with no topic subscription reads the whole language, + so they raise demand for every topic in it. + + * A bucket with demand gets a quota that scales with its reader count, clamped + into a sane band. A bucket with no demand gets nothing from the demand path. + + * A **pilot-light floor** keeps a handful of distinct topics warm in every + *supported* language that currently has no active readers, so a brand-new + learner (or the Romanian kiosk) never opens an empty feed. The floor is + per-language and topic-diverse, not per-topic, so it stays cheap. + +``FunnelBudget`` is the single object the crawler threads through; it holds the +demand surface plus the day's running counts and answers one question per +article: *does this article earn a simplification?* +""" + +from collections import defaultdict +from datetime import datetime, timedelta + +# --- Tunables --------------------------------------------------------------- + +# A reader counts as "active" if seen within this many days. This is the one +# knob that defines the whole demand surface; see ``_active_reader_rows``. +ACTIVE_READER_DAYS = 30 + +# Per-(language, topic) daily simplification quota for a bucket that has demand. +# Scales with the number of active readers who want the bucket, clamped so a +# single reader still gets a usable feed and one popular bucket can't run away. +# readers 1-5 -> 20, 6-10 -> 30, 11-15 -> 40, 16+ -> 50 (matches the old da/fr +# top cap for high-demand buckets while starving nobody). +MIN_ACTIVE_QUOTA = 20 +MAX_ACTIVE_QUOTA = 50 +QUOTA_STEP = 10 +READERS_PER_QUOTA_STEP = 5 + +# Pilot light: a *supported* language with no active readers still gets a few +# simplified articles a day, spread across distinct topics (diverse, not +# per-topic) so a future learner's first feed isn't empty. ~5 / language / day. +LANGUAGE_FLOOR_TOPICS = 5 +FLOOR_PER_TOPIC = 1 + + +def _active_quota(readers): + """Demand quota for a bucket with ``readers`` active readers (readers >= 1).""" + steps = (readers - 1) // READERS_PER_QUOTA_STEP + return min(MAX_ACTIVE_QUOTA, MIN_ACTIVE_QUOTA + steps * QUOTA_STEP) + + +class DemandSurface: + """What active readers want, deduped across the cohort. + + ``subscribed_readers`` maps ``(language_id, topic_id)`` -> count of distinct + active readers subscribed to that topic in that language. ``unfiltered_readers`` + maps ``language_id`` -> count of distinct active readers with *no* topic + subscription; those readers read every topic in the language, so they lend + demand to all of its topics. + """ + + def __init__(self, subscribed_readers, unfiltered_readers): + self.subscribed_readers = dict(subscribed_readers) + self.unfiltered_readers = dict(unfiltered_readers) + + def readers_for(self, language_id, topic_id): + """Distinct active readers who want this bucket (subscribers + all-topic readers).""" + return self.subscribed_readers.get( + (language_id, topic_id), 0 + ) + self.unfiltered_readers.get(language_id, 0) + + def demanded_topic_ids(self, language_id): + """Topic ids explicitly subscribed to in this language. + + Empty when nobody subscribes to a *specific* topic here. That is + ambiguous on its own -- it can mean "dormant" or "all-topic readers + only" -- so pair it with ``unfiltered_readers`` / ``language_has_demand``. + """ + return { + topic for (lang, topic) in self.subscribed_readers if lang == language_id + } + + def language_has_demand(self, language_id): + """True if any active reader wants any topic in this language.""" + if self.unfiltered_readers.get(language_id, 0) > 0: + return True + return any(lang == language_id for (lang, _topic) in self.subscribed_readers) + + def summary(self): + """Short human-readable line for crawl logs.""" + langs = set(self.unfiltered_readers) | { + lang for (lang, _t) in self.subscribed_readers + } + return ( + f"{len(langs)} language(s) with demand, " + f"{len(self.subscribed_readers)} subscribed bucket(s), " + f"{sum(self.unfiltered_readers.values())} all-topic reader(s)" + ) + + +def _active_reader_rows(session, active_days): + """(user_id, learned_language_id) for every currently-active reader. + + "Active" is defined here and nowhere else, so switching the definition + (e.g. to readers who actually *opened* an article) is a one-function change. + """ + from zeeguu.core.model.user import User + + cutoff = datetime.now() - timedelta(days=active_days) + return ( + session.query(User.id, User.learned_language_id) + .filter(User.last_seen >= cutoff) + .filter(User.learned_language_id != None) # noqa: E711 (SQLAlchemy needs != None) + .all() + ) + + +def compute_demand_surface(session, active_days=ACTIVE_READER_DAYS): + """Build the demand surface from the active cohort's language + topic subscriptions.""" + from zeeguu.core.model.topic_subscription import TopicSubscription + + active_rows = _active_reader_rows(session, active_days) + if not active_rows: + return DemandSurface({}, {}) + + lang_by_user = {uid: lang for uid, lang in active_rows} + + subs_by_user = defaultdict(set) + subs = ( + session.query(TopicSubscription.user_id, TopicSubscription.topic_id) + .filter(TopicSubscription.user_id.in_(list(lang_by_user))) + .all() + ) + for uid, topic_id in subs: + subs_by_user[uid].add(topic_id) + + subscribed_readers = defaultdict(int) + unfiltered_readers = defaultdict(int) + for uid, lang in active_rows: + topics = subs_by_user.get(uid) + if not topics: + # No topic subscription -> reads the whole language. + unfiltered_readers[lang] += 1 + else: + for topic_id in topics: + subscribed_readers[(lang, topic_id)] += 1 + + return DemandSurface(subscribed_readers, unfiltered_readers) + + +class FunnelBudget: + """The day's simplification budget for one crawl process. + + Threaded through ``download_from_feed`` -> ``download_feed_item``; the gate is + a single ``should_simplify`` call per article, followed by ``record`` when the + article is actually simplified. Replaces the old ``topic_simplification_counts`` + dict + static ``get_max_simplified_for_language`` cap. + """ + + def __init__(self, demand_surface, todays_counts=None, floor_topics_used=None): + self.demand = demand_surface + # {(language_id, topic_id): simplifications recorded so far today} + self.counts = defaultdict(int, todays_counts or {}) + # {language_id: set(topic_id)} distinct topics already given a floor slot + self.floor_topics_used = defaultdict(set, floor_topics_used or {}) + + def quota_for(self, language_id, topic_id): + """Demand-path quota for a bucket (0 when it has no demand; floor is separate).""" + readers = self.demand.readers_for(language_id, topic_id) + return _active_quota(readers) if readers > 0 else 0 + + def should_simplify(self, language_id, article_topic_ids): + """Decide whether an article earns a simplification. + + Returns ``(bool, reason)`` where reason is ``"demand"``, ``"floor"`` or + ``"capped"``. An article with no topics can never match a bucket and is + never simplified through the funnel (parity with the old cap, which only + kicked in when there were topics to match). + """ + if not article_topic_ids: + # No topic means no bucket to match. Such an article is only wanted by + # readers who read the whole language (no subscription); serve them, + # otherwise skip (a topic-subscriber can never see it anyway). Old + # behaviour simplified every untagged article -- that was the waste. + if self.demand.unfiltered_readers.get(language_id, 0) > 0: + return True, "no-topic" + return False, "no-topics" + + # Demand path: simplify if ANY of the article's topics is under quota. + for topic_id in article_topic_ids: + if self.counts[(language_id, topic_id)] < self.quota_for( + language_id, topic_id + ): + return True, "demand" + + # Pilot-light floor: only for supported languages with no demand at all, + # and only enough to keep a handful of distinct topics warm. + if not self.demand.language_has_demand(language_id): + used = self.floor_topics_used[language_id] + for topic_id in article_topic_ids: + if topic_id in used: + if self.counts[(language_id, topic_id)] < FLOOR_PER_TOPIC: + return True, "floor" + elif len(used) < LANGUAGE_FLOOR_TOPICS: + return True, "floor" + + return False, "capped" + + def record(self, language_id, article_topic_ids, reason): + """Book a simplification against every topic on the article.""" + for topic_id in article_topic_ids: + self.counts[(language_id, topic_id)] += 1 + if reason == "floor": + self.floor_topics_used[language_id].add(topic_id) + + def language_simplification_headroom(self, language_id): + """How many more simplifications this language can still absorb today. + + Drives the pre-download triage (Task 5, phase 3): if a language is fully + satisfied we skip its feeds before paying any readability/download cost; + if it has limited room we download only the best few titles. + + Returns ``None`` for "effectively unbounded" -- when any reader reads the + whole language, every topic is wanted, so there is no small budget to + prune against and the caller should just honour its own ``limit``. + """ + demand = self.demand + if demand.unfiltered_readers.get(language_id, 0) > 0: + return None # all-topic readers -> every topic wanted + + if not demand.language_has_demand(language_id): + # Dormant supported language: only the pilot-light floor slots remain. + used = len(self.floor_topics_used.get(language_id, ())) + return max(0, LANGUAGE_FLOOR_TOPICS - used) + + # Specific topic subscriptions only: sum the remaining quota per bucket. + headroom = 0 + for topic_id in demand.demanded_topic_ids(language_id): + quota = self.quota_for(language_id, topic_id) + headroom += max(0, quota - self.counts[(language_id, topic_id)]) + return headroom + + +# --- Pre-download title triage (phase 3) ------------------------------------ + +# When a language has limited headroom we still download a bit more than the +# strict remaining count, because a downloaded article's topic won't always land +# in an under-quota bucket. This slack keeps buckets from under-filling. +TRIAGE_OVERSHOOT = 2 + + +def triage_keep_count(headroom, limit): + """How many feed items to actually download, given today's headroom + the + caller's per-feed ``limit``. ``headroom is None`` means unbounded -> ``limit``.""" + if headroom is None: + return limit + if headroom <= 0: + return 0 + return min(limit, headroom * TRIAGE_OVERSHOOT) + + +def select_titles_to_download(titles, language_code, demand_topic_names, keep_count, ranker=None): + """Pick the best ``keep_count`` of ``titles`` to download (best-N, not first-N). + + ``titles`` is the ordered list of candidate feed-item titles. Returns the + *indices* (into ``titles``) to keep, best first. A cheap LLM ranks them by + newsworthiness / learner-suitability / topic match / diversity; on ANY + failure (no key, timeout, unparseable output) we fall back to first-N, i.e. + the crawler's historical behaviour -- triage can only ever be an improvement, + never a regression. + """ + if keep_count <= 0: + return [] + if len(titles) <= keep_count: + return list(range(len(titles))) + + ranker = ranker or _rank_titles_with_llm + try: + indices = ranker(titles, language_code, demand_topic_names, keep_count) + # Keep only valid, in-range, de-duplicated indices, preserving order. + seen = set() + clean = [] + for i in indices: + if isinstance(i, int) and 0 <= i < len(titles) and i not in seen: + seen.add(i) + clean.append(i) + if clean: + return clean[:keep_count] + except Exception as e: + from zeeguu.logging import log + + log(f" ⚠ Title triage failed ({e}); falling back to first-{keep_count}") + return list(range(keep_count)) + + +def _rank_titles_with_llm(titles, language_code, demand_topic_names, keep_count): + """Ask the cheap LLM tier to pick the best ``keep_count`` title indices.""" + import json + import re + + from zeeguu.core.llm_services.llm_service import UnifiedLLMService + + numbered = "\n".join(f"{i}. {t}" for i, t in enumerate(titles)) + interests = ", ".join(demand_topic_names) if demand_topic_names else "any general-interest news" + prompt = ( + f"You curate a {language_code} news feed for language learners.\n" + f"Our readers are interested in: {interests}.\n" + f"From the numbered headlines below, choose the best {keep_count} to keep. " + f"Favour genuine, timely news that matches the interests, is suitable for " + f"language learning, and gives topic variety. Avoid clickbait, ads, and " + f"near-duplicates.\n" + f"Reply with ONLY a JSON array of the chosen numbers, best first, " + f"e.g. [3, 0, 7].\n\n" + f"{numbered}" + ) + + raw = UnifiedLLMService().generate_text(prompt, max_tokens=200, temperature=0.2) + match = re.search(r"\[[^\]]*\]", raw) + if not match: + raise ValueError(f"no JSON array in LLM reply: {raw!r}") + return [int(x) for x in json.loads(match.group(0))] diff --git a/zeeguu/core/test/test_funnel.py b/zeeguu/core/test/test_funnel.py new file mode 100644 index 000000000..9dca766cc --- /dev/null +++ b/zeeguu/core/test/test_funnel.py @@ -0,0 +1,347 @@ +"""Tests for the demand-aware ingestion funnel (Task 5). + +The quota / budget / floor logic is pure Python and tested without a DB. The +demand-surface query is tested against real User + TopicSubscription rows. +""" + +import datetime +from unittest import TestCase + +import zeeguu.core +from zeeguu.core.test.model_test_mixin import ModelTestMixIn +from zeeguu.core.test.rules.user_rule import UserRule +from zeeguu.core.test.rules.language_rule import LanguageRule +from zeeguu.core.test.rules.topic_rule import TopicRule + +from zeeguu.core.content_retriever.funnel import ( + DemandSurface, + FunnelBudget, + compute_demand_surface, + triage_keep_count, + select_titles_to_download, + _active_quota, + MIN_ACTIVE_QUOTA, + MAX_ACTIVE_QUOTA, + LANGUAGE_FLOOR_TOPICS, + TRIAGE_OVERSHOOT, +) + +db_session = zeeguu.core.model.db.session + + +class FunnelQuotaTest(TestCase): + """Pure quota-scaling logic; no DB needed.""" + + def test_single_reader_gets_min_quota(self): + self.assertEqual(_active_quota(1), MIN_ACTIVE_QUOTA) + + def test_quota_scales_up_with_readers(self): + self.assertEqual(_active_quota(1), 20) + self.assertEqual(_active_quota(5), 20) + self.assertEqual(_active_quota(6), 30) + self.assertEqual(_active_quota(11), 40) + self.assertEqual(_active_quota(16), 50) + + def test_quota_is_clamped_at_max(self): + self.assertEqual(_active_quota(1000), MAX_ACTIVE_QUOTA) + + +class DemandSurfaceLogicTest(TestCase): + """readers_for / language_has_demand; no DB needed.""" + + def setUp(self): + # Language 1: 2 readers subscribed to topic 10, plus 3 all-topic readers. + # Language 2: only subscribed readers. Language 3: nothing. + self.surface = DemandSurface( + subscribed_readers={(1, 10): 2, (2, 20): 1}, + unfiltered_readers={1: 3}, + ) + + def test_all_topic_readers_lend_demand_to_every_topic(self): + # Topic 10 in lang 1: 2 subscribers + 3 all-topic = 5 + self.assertEqual(self.surface.readers_for(1, 10), 5) + # An unsubscribed topic in lang 1 still has the 3 all-topic readers + self.assertEqual(self.surface.readers_for(1, 99), 3) + + def test_subscribed_only_language(self): + self.assertEqual(self.surface.readers_for(2, 20), 1) + self.assertEqual(self.surface.readers_for(2, 99), 0) + + def test_language_has_demand(self): + self.assertTrue(self.surface.language_has_demand(1)) + self.assertTrue(self.surface.language_has_demand(2)) + self.assertFalse(self.surface.language_has_demand(3)) + + +class FunnelBudgetGateTest(TestCase): + """should_simplify / record / floor diversity; no DB needed.""" + + def test_demand_bucket_simplifies_until_quota(self): + surface = DemandSurface({(1, 10): 1}, {}) # 1 reader -> quota 20 + budget = FunnelBudget(surface) + for _ in range(MIN_ACTIVE_QUOTA): + ok, reason = budget.should_simplify(1, [10]) + self.assertTrue(ok) + self.assertEqual(reason, "demand") + budget.record(1, [10], reason) + ok, reason = budget.should_simplify(1, [10]) + self.assertFalse(ok) + self.assertEqual(reason, "capped") + + def test_seeded_counts_are_respected(self): + surface = DemandSurface({(1, 10): 1}, {}) + budget = FunnelBudget(surface, todays_counts={(1, 10): MIN_ACTIVE_QUOTA}) + ok, reason = budget.should_simplify(1, [10]) + self.assertFalse(ok) + + def test_article_matches_if_any_topic_under_quota(self): + # Topic 10 full, topic 11 has demand and room -> simplify. + surface = DemandSurface({(1, 10): 1, (1, 11): 1}, {}) + budget = FunnelBudget(surface, todays_counts={(1, 10): MIN_ACTIVE_QUOTA}) + ok, reason = budget.should_simplify(1, [10, 11]) + self.assertTrue(ok) + self.assertEqual(reason, "demand") + + def test_no_topics_skipped_without_all_topic_readers(self): + # Topic-subscribers can never see an untagged article, so skip it. + budget = FunnelBudget(DemandSurface({(1, 10): 1}, {})) + ok, reason = budget.should_simplify(1, []) + self.assertFalse(ok) + self.assertEqual(reason, "no-topics") + + def test_no_topics_simplified_for_all_topic_readers(self): + # A reader with no subscription reads the whole language, untagged included. + budget = FunnelBudget(DemandSurface({}, {1: 2})) + ok, reason = budget.should_simplify(1, []) + self.assertTrue(ok) + self.assertEqual(reason, "no-topic") + + def test_dormant_language_gets_topic_diverse_floor(self): + # No demand anywhere -> language 1 is dormant; floor keeps a few distinct + # topics warm, one each, capped at LANGUAGE_FLOOR_TOPICS. + budget = FunnelBudget(DemandSurface({}, {})) + simplified_topics = [] + for topic_id in range(100, 100 + LANGUAGE_FLOOR_TOPICS + 3): + ok, reason = budget.should_simplify(1, [topic_id]) + if ok: + self.assertEqual(reason, "floor") + budget.record(1, [topic_id], reason) + simplified_topics.append(topic_id) + self.assertEqual(len(simplified_topics), LANGUAGE_FLOOR_TOPICS) + + def test_floor_does_not_refire_same_topic(self): + budget = FunnelBudget(DemandSurface({}, {})) + ok, reason = budget.should_simplify(1, [100]) + self.assertTrue(ok) + budget.record(1, [100], reason) + # Same topic again: floor is 1/topic, so no more. + ok, _ = budget.should_simplify(1, [100]) + self.assertFalse(ok) + + def test_language_with_demand_gets_no_floor(self): + # Lang 1 has demand on topic 10; a different, unsubscribed topic 99 with + # 0 readers gets neither demand nor floor (floor is only for dormant langs). + surface = DemandSurface({(1, 10): 1}, {}) + budget = FunnelBudget(surface) + ok, reason = budget.should_simplify(1, [99]) + self.assertFalse(ok) + self.assertEqual(reason, "capped") + + +class HeadroomTest(TestCase): + """language_simplification_headroom drives the pre-download triage; no DB.""" + + def test_all_topic_readers_mean_unbounded(self): + budget = FunnelBudget(DemandSurface({}, {1: 3})) + self.assertIsNone(budget.language_simplification_headroom(1)) + + def test_dormant_language_headroom_is_floor_slots_left(self): + budget = FunnelBudget(DemandSurface({}, {})) + self.assertEqual( + budget.language_simplification_headroom(1), LANGUAGE_FLOOR_TOPICS + ) + budget.record(1, [100], "floor") + self.assertEqual( + budget.language_simplification_headroom(1), LANGUAGE_FLOOR_TOPICS - 1 + ) + + def test_subscribed_language_headroom_sums_remaining_quota(self): + # Two demanded buckets, quota 20 each. One already half used. + surface = DemandSurface({(1, 10): 1, (1, 11): 1}, {}) + budget = FunnelBudget(surface, todays_counts={(1, 10): 5}) + self.assertEqual( + budget.language_simplification_headroom(1), + (MIN_ACTIVE_QUOTA - 5) + MIN_ACTIVE_QUOTA, + ) + + def test_fully_used_language_has_zero_headroom(self): + surface = DemandSurface({(1, 10): 1}, {}) + budget = FunnelBudget(surface, todays_counts={(1, 10): MIN_ACTIVE_QUOTA}) + self.assertEqual(budget.language_simplification_headroom(1), 0) + + +class TriageSelectionTest(TestCase): + """triage_keep_count + select_titles_to_download; no DB.""" + + def test_keep_count_unbounded_uses_limit(self): + self.assertEqual(triage_keep_count(None, 10), 10) + + def test_keep_count_zero_headroom_skips(self): + self.assertEqual(triage_keep_count(0, 10), 0) + + def test_keep_count_overshoots_but_respects_limit(self): + self.assertEqual(triage_keep_count(2, 10), 2 * TRIAGE_OVERSHOOT) + self.assertEqual(triage_keep_count(50, 10), 10) + + def test_keep_all_when_fewer_titles_than_keep(self): + titles = ["a", "b"] + self.assertEqual(select_titles_to_download(titles, "de", [], 5), [0, 1]) + + def test_ranker_choice_is_honoured(self): + titles = ["a", "b", "c", "d"] + + def ranker(ts, lang, interests, keep): + return [3, 1] + + self.assertEqual( + select_titles_to_download(titles, "de", [], 2, ranker=ranker), [3, 1] + ) + + def test_ranker_output_is_sanitized(self): + titles = ["a", "b", "c", "d"] + + def messy_ranker(ts, lang, interests, keep): + return [3, 3, 99, "x", 1] # dupes, out-of-range, wrong type + + self.assertEqual( + select_titles_to_download(titles, "de", [], 3, ranker=messy_ranker), [3, 1] + ) + + def test_ranker_failure_falls_back_to_first_n(self): + titles = ["a", "b", "c", "d"] + + def broken_ranker(ts, lang, interests, keep): + raise RuntimeError("LLM down") + + self.assertEqual( + select_titles_to_download(titles, "de", [], 2, ranker=broken_ranker), [0, 1] + ) + + def test_empty_ranker_output_falls_back_to_first_n(self): + titles = ["a", "b", "c", "d"] + self.assertEqual( + select_titles_to_download(titles, "de", [], 2, ranker=lambda *a: []), + [0, 1], + ) + + +class ComputeDemandSurfaceTest(ModelTestMixIn, TestCase): + """The demand-surface query against real User + TopicSubscription rows.""" + + def setUp(self): + super().setUp() + from zeeguu.core.model.topic_subscription import TopicSubscription + + self.lang = LanguageRule().de + self.topic_sports = TopicRule.get_or_create_topic(1) + self.topic_tech = TopicRule.get_or_create_topic(3) + + # Active subscribed reader: German, subscribed to Sports. + self.subscribed = UserRule().user + self.subscribed.learned_language = self.lang + self.subscribed.last_seen = datetime.datetime.now() + TopicSubscription.find_or_create(db_session, self.subscribed, self.topic_sports) + + # Active all-topic reader: German, no subscriptions. + self.unfiltered = UserRule().user + self.unfiltered.learned_language = self.lang + self.unfiltered.last_seen = datetime.datetime.now() + + # Dormant reader: German, but not seen in a long time -> excluded. + self.dormant = UserRule().user + self.dormant.learned_language = self.lang + self.dormant.last_seen = datetime.datetime.now() - datetime.timedelta(days=90) + TopicSubscription.find_or_create(db_session, self.dormant, self.topic_tech) + + db_session.commit() + + def test_active_subscribers_and_all_topic_readers_counted(self): + surface = compute_demand_surface(db_session) + # Sports: 1 subscriber + 1 all-topic reader = 2 + self.assertEqual(surface.readers_for(self.lang.id, self.topic_sports.id), 2) + # Tech: subscribed only by the dormant reader (excluded), but the + # all-topic active reader still lends demand = 1 + self.assertEqual(surface.readers_for(self.lang.id, self.topic_tech.id), 1) + self.assertTrue(surface.language_has_demand(self.lang.id)) + + def test_dormant_reader_excluded(self): + # If we shrink the active window below the dormant reader's staleness it + # stays excluded; the all-topic reader alone keeps tech demand at 1, not 2. + surface = compute_demand_surface(db_session) + self.assertEqual(surface.readers_for(self.lang.id, self.topic_tech.id), 1) + + +class BackfillInventoryTest(ModelTestMixIn, TestCase): + """fresh_simplified_count / recent_unsimplified_articles against real rows.""" + + def setUp(self): + super().setUp() + from zeeguu.core.test.rules.article_rule import ArticleRule + from zeeguu.core.model import ArticleTopicMap + from zeeguu.core.model.article_topic_map import TopicOriginType + + self.lang = LanguageRule().de + self.topic = TopicRule.get_or_create_topic(1) # Sports + + def make_original(days_old, broken=0): + art = ArticleRule().article + art.language = self.lang + art.parent_article_id = None + art.broken = broken + art.published_time = datetime.datetime.now() - datetime.timedelta(days=days_old) + db_session.add( + ArticleTopicMap(art, self.topic, TopicOriginType.HARDSET) + ) + db_session.commit() + return art + + # Two fresh raw originals with the topic, not yet simplified. + self.raw_recent = [make_original(1), make_original(2)] + # One old raw original (outside the lookback window). + self.raw_old = make_original(30) + + # One original that HAS a simplified child (so it's "fresh simplified"). + self.parent = make_original(1) + child = ArticleRule().article + child.language = self.lang + child.parent_article_id = self.parent.id + child.broken = 0 + child.published_time = datetime.datetime.now() + db_session.add(child) + db_session.commit() + + def test_fresh_simplified_count_counts_children_in_window(self): + from zeeguu.core.content_retriever.backfill import fresh_simplified_count + + self.assertEqual( + fresh_simplified_count(db_session, self.lang.id, self.topic.id), 1 + ) + + def test_recent_unsimplified_excludes_old_and_already_simplified(self): + from zeeguu.core.content_retriever.backfill import recent_unsimplified_articles + + found = recent_unsimplified_articles(db_session, self.lang.id, self.topic.id) + found_ids = {a.id for a in found} + # The two fresh raw originals are eligible... + for art in self.raw_recent: + self.assertIn(art.id, found_ids) + # ...but the old one and the already-simplified parent are not. + self.assertNotIn(self.raw_old.id, found_ids) + self.assertNotIn(self.parent.id, found_ids) + + def test_recent_unsimplified_respects_topic_filter(self): + from zeeguu.core.content_retriever.backfill import recent_unsimplified_articles + + other_topic = TopicRule.get_or_create_topic(7) # Politics — nothing tagged + found = recent_unsimplified_articles(db_session, self.lang.id, other_topic.id) + self.assertEqual(found, []) diff --git a/zeeguu/operations/crawler/crawl.py b/zeeguu/operations/crawler/crawl.py index bed7c5a66..d776db411 100755 --- a/zeeguu/operations/crawler/crawl.py +++ b/zeeguu/operations/crawler/crawl.py @@ -166,36 +166,42 @@ def crawl_round_robin(languages_to_crawl, articles_per_feed=1, recent_days=None, crawl_report.add_language(lang_code) crawl_reports[lang_code] = crawl_report - # Track simplified articles per (language, topic) per day - # This ensures topic diversity per language - caps vary by language popularity - # Initialize from database with today's counts - from zeeguu.core.content_retriever.article_downloader import get_todays_simplified_counts_by_language_topic, get_max_simplified_for_language - topic_simplification_counts = defaultdict(int) # Key: (language_id, topic_id) - - # Load today's counts from DB for all languages we're crawling + # Demand-aware ingestion funnel (Task 5). The simplification budget is no + # longer a static per-language cap; it's a per-(language, topic) quota + # derived from what active readers actually want, plus a pilot-light floor + # for dormant supported languages. Today's already-simplified counts seed the + # budget so a re-run within the day doesn't blow past the quota. + from zeeguu.core.content_retriever.article_downloader import get_todays_simplified_counts_by_language_topic + from zeeguu.core.content_retriever.funnel import compute_demand_surface, FunnelBudget + + demand_surface = compute_demand_surface(db_session) + log(f"Demand surface: {demand_surface.summary()}") + + todays_counts = {} # {(language_id, topic_id): count} for lang_code in languages_to_crawl: language = Language.find(lang_code) if language: - todays_counts = get_todays_simplified_counts_by_language_topic(db_session, language.id) - for topic_id, count in todays_counts.items(): - topic_simplification_counts[(language.id, topic_id)] = count + for topic_id, count in get_todays_simplified_counts_by_language_topic( + db_session, language.id + ).items(): + todays_counts[(language.id, topic_id)] = count + + funnel_budget = FunnelBudget(demand_surface, todays_counts=todays_counts) - if topic_simplification_counts: + if todays_counts: log(f"Today's simplified counts by language/topic:") from zeeguu.core.model import Topic - # Group by language for cleaner output by_language = defaultdict(list) - for (lang_id, topic_id), count in topic_simplification_counts.items(): + for (lang_id, topic_id), count in todays_counts.items(): by_language[lang_id].append((topic_id, count)) for lang_code in languages_to_crawl: language = Language.find(lang_code) if language and language.id in by_language: - max_cap = get_max_simplified_for_language(lang_code) topics_str = ", ".join([ - f"{Topic.find_by_id(tid).title}:{cnt}" + f"{Topic.find_by_id(tid).title}:{cnt}(q{funnel_budget.quota_for(language.id, tid)})" for tid, cnt in sorted(by_language[language.id], key=lambda x: -x[1]) ]) - log(f" {lang_code.upper()} (cap:{max_cap}): {topics_str}") + log(f" {lang_code.upper()}: {topics_str}") # Get all feeds grouped by language feeds_by_language = {} @@ -264,7 +270,7 @@ def crawl_round_robin(languages_to_crawl, articles_per_feed=1, recent_days=None, crawl_report, limit=max_articles_per_feed, simplification_provider=simplification_provider, - topic_simplification_counts=topic_simplification_counts, + funnel_budget=funnel_budget, ) feed_time = time() - feed_start_time @@ -292,23 +298,23 @@ def crawl_round_robin(languages_to_crawl, articles_per_feed=1, recent_days=None, log(f"\nFinished processing {feeds_completed} feeds across {len(languages_to_crawl)} languages") - # Log topic simplification summary grouped by language - if topic_simplification_counts: + # Log the funnel's per-(language, topic) simplification summary vs. quota. + if funnel_budget.counts: from zeeguu.core.model import Topic - log(f"\nTopic Simplification Summary:") + log(f"\nSimplification Funnel Summary (count / quota):") by_language = defaultdict(list) - for (lang_id, topic_id), count in topic_simplification_counts.items(): + for (lang_id, topic_id), count in funnel_budget.counts.items(): by_language[lang_id].append((topic_id, count)) for lang_code in languages_to_crawl: language = Language.find(lang_code) if language and language.id in by_language: - max_cap = get_max_simplified_for_language(lang_code) - log(f" {lang_code.upper()} (cap:{max_cap}):") + log(f" {lang_code.upper()}:") for topic_id, count in sorted(by_language[language.id], key=lambda x: -x[1]): topic = Topic.find_by_id(topic_id) topic_name = topic.title if topic else f"Unknown({topic_id})" - cap_indicator = " [CAP]" if count >= max_cap else "" - log(f" {topic_name}: {count}{cap_indicator}") + quota = funnel_budget.quota_for(language.id, topic_id) + cap_indicator = " [QUOTA]" if quota and count >= quota else "" + log(f" {topic_name}: {count}/{quota or 'floor'}{cap_indicator}") # Calculate and save total times per language for lang_code, crawl_report in crawl_reports.items(): From f29cc0baeb6364b6a786a6ee09cd0532f8366253 Mon Sep 17 00:00:00 2001 From: Mircea Lungu Date: Thu, 16 Jul 2026 12:50:41 +0300 Subject: [PATCH 2/2] Task 5 review fixes: floor re-firing, triage crash-safety, backfill freshness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three correctness bugs from code review of the demand-aware funnel: 1. Pilot-light floor re-fired on every intra-day crawl. FunnelBudget seeded today's demand counts but not floor_topics_used, so each hourly crawl process re-floored a fresh set of ~5 topics for a dormant language — ~5*N/day instead of ~5. FunnelBudget now derives floor_topics_used from today's counts for languages with no demand (correct by construction for any caller). 2. _apply_title_triage claimed "never raises" but had no try/except, so a titleless RSS item (or any error) aborted the entire feed instead of falling back to the full list. Wrapped the body; also use feed_item.get("title", ""). 3. backfill freshness/recency keyed off published_time, which simplified children inherit from the (backdatable) parent — so the "already stocked" guard under-counted and the candidate query missed recently-crawled but old-dated raw articles. Both now filter/order on crawled_at (ingestion time). +5 regression tests (36 total). Co-Authored-By: Claude Opus 4.8 --- .../content_retriever/article_downloader.py | 57 ++++++----- zeeguu/core/content_retriever/backfill.py | 10 +- zeeguu/core/content_retriever/funnel.py | 12 ++- zeeguu/core/test/test_funnel.py | 97 ++++++++++++++++++- 4 files changed, 146 insertions(+), 30 deletions(-) diff --git a/zeeguu/core/content_retriever/article_downloader.py b/zeeguu/core/content_retriever/article_downloader.py index a02d8ee5b..1704e292f 100644 --- a/zeeguu/core/content_retriever/article_downloader.py +++ b/zeeguu/core/content_retriever/article_downloader.py @@ -299,35 +299,42 @@ def _apply_title_triage(feed, items, limit, funnel_budget): from zeeguu.core.model import Topic items_list = list(items) - headroom = funnel_budget.language_simplification_headroom(feed.language_id) - keep = triage_keep_count(headroom, limit) + 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 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 + 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) + 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["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] + 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( diff --git a/zeeguu/core/content_retriever/backfill.py b/zeeguu/core/content_retriever/backfill.py index 328207441..06a7ee2d9 100644 --- a/zeeguu/core/content_retriever/backfill.py +++ b/zeeguu/core/content_retriever/backfill.py @@ -58,7 +58,9 @@ def fresh_simplified_count(session, language_id, topic_id, days=BACKFILL_FRESH_D session.query(Article.id) .join(ParentArticle, Article.parent_article_id == ParentArticle.id) .filter(ParentArticle.language_id == language_id) - .filter(Article.published_time >= cutoff) + # 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: @@ -84,7 +86,9 @@ def recent_unsimplified_articles( .filter(Article.parent_article_id.is_(None)) # originals only .filter(Child.id.is_(None)) # not yet simplified .filter(Article.language_id == language_id) - .filter(Article.published_time >= cutoff) + # 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: @@ -92,7 +96,7 @@ def recent_unsimplified_articles( ArticleTopicMap, ArticleTopicMap.article_id == Article.id ).filter(ArticleTopicMap.topic_id == topic_id) - return q.order_by(Article.published_time.desc()).limit(limit).all() + return q.order_by(Article.crawled_at.desc()).limit(limit).all() def maybe_backfill_bucket(user_id, topic_id=None, provider="deepseek"): diff --git a/zeeguu/core/content_retriever/funnel.py b/zeeguu/core/content_retriever/funnel.py index 957bfaa72..10ee4f5b5 100644 --- a/zeeguu/core/content_retriever/funnel.py +++ b/zeeguu/core/content_retriever/funnel.py @@ -172,7 +172,17 @@ def __init__(self, demand_surface, todays_counts=None, floor_topics_used=None): # {(language_id, topic_id): simplifications recorded so far today} self.counts = defaultdict(int, todays_counts or {}) # {language_id: set(topic_id)} distinct topics already given a floor slot - self.floor_topics_used = defaultdict(set, floor_topics_used or {}) + if floor_topics_used is None: + # A dormant language's only simplifications today came from the floor + # path, so today's counts ARE its floored topics. Seed from them, or + # an intra-day re-run (the crawler runs hourly) would floor a fresh + # set of topics every time and multiply the pilot-light cost by the + # number of crawls per day. + floor_topics_used = defaultdict(set) + for (lang_id, topic_id), count in self.counts.items(): + if count > 0 and not demand_surface.language_has_demand(lang_id): + floor_topics_used[lang_id].add(topic_id) + self.floor_topics_used = defaultdict(set, floor_topics_used) def quota_for(self, language_id, topic_id): """Demand-path quota for a bucket (0 when it has no demand; floor is separate).""" diff --git a/zeeguu/core/test/test_funnel.py b/zeeguu/core/test/test_funnel.py index 9dca766cc..1a081038c 100644 --- a/zeeguu/core/test/test_funnel.py +++ b/zeeguu/core/test/test_funnel.py @@ -148,6 +148,63 @@ def test_language_with_demand_gets_no_floor(self): self.assertEqual(reason, "capped") +class FloorSeedingTest(TestCase): + """Seeding floor_topics_used from today's counts stops the pilot-light floor + from re-firing on every intra-day crawl re-run. No DB.""" + + def test_full_floor_from_seeded_counts_does_not_refire(self): + # Dormant language 1 already floored 5 distinct topics today. + seeded = {(1, t): 1 for t in range(100, 100 + LANGUAGE_FLOOR_TOPICS)} + budget = FunnelBudget(DemandSurface({}, {}), todays_counts=seeded) + # A fresh topic gets nothing (floor already full)... + self.assertFalse(budget.should_simplify(1, [200])[0]) + # ...and an already-floored topic doesn't re-fire either. + self.assertFalse(budget.should_simplify(1, [100])[0]) + + def test_partial_floor_seed_allows_only_remaining_slots(self): + seeded = {(1, 100): 1, (1, 101): 1} # 2 of 5 slots used today + budget = FunnelBudget(DemandSurface({}, {}), todays_counts=seeded) + granted = [] + for topic_id in range(200, 210): + ok, reason = budget.should_simplify(1, [topic_id]) + if ok: + budget.record(1, [topic_id], reason) + granted.append(topic_id) + self.assertEqual(len(granted), LANGUAGE_FLOOR_TOPICS - 2) + + def test_demand_language_counts_not_seeded_as_floor(self): + # Counts for a language WITH demand came from the demand path, not the + # floor, so they must not seed floor_topics_used. + surface = DemandSurface({(1, 10): 1}, {}) + budget = FunnelBudget(surface, todays_counts={(1, 10): 5}) + self.assertEqual(budget.floor_topics_used.get(1, set()), set()) + + +class TitleTriageSafetyTest(TestCase): + """_apply_title_triage must fall back to the full item list on any error, + never let an exception drop the whole feed. No DB / no LLM.""" + + def test_falls_back_to_full_list_on_error(self): + from zeeguu.core.content_retriever.article_downloader import _apply_title_triage + + class BoomBudget: + def language_simplification_headroom(self, language_id): + raise RuntimeError("boom") + + class FakeLang: + id = 1 + code = "de" + + class FakeFeed: + language_id = 1 + language = FakeLang() + title = "Feed" + + items = [{"title": "a"}, {"title": "b"}] + result = _apply_title_triage(FakeFeed(), items, 1, BoomBudget()) + self.assertEqual(result, items) + + class HeadroomTest(TestCase): """language_simplification_headroom drives the pre-download triage; no DB.""" @@ -298,6 +355,9 @@ def make_original(days_old, broken=0): art.language = self.lang art.parent_article_id = None art.broken = broken + # Recency keys off crawled_at (ingestion time), not the backdatable + # published_time, so drive the test window with crawled_at. + art.crawled_at = datetime.datetime.now() - datetime.timedelta(days=days_old) art.published_time = datetime.datetime.now() - datetime.timedelta(days=days_old) db_session.add( ArticleTopicMap(art, self.topic, TopicOriginType.HARDSET) @@ -316,7 +376,10 @@ def make_original(days_old, broken=0): child.language = self.lang child.parent_article_id = self.parent.id child.broken = 0 - child.published_time = datetime.datetime.now() + # Child inherits the parent's (old) published_time in production; its + # crawled_at is when it was simplified, which is what freshness counts. + child.crawled_at = datetime.datetime.now() + child.published_time = self.parent.published_time db_session.add(child) db_session.commit() @@ -345,3 +408,35 @@ def test_recent_unsimplified_respects_topic_filter(self): other_topic = TopicRule.get_or_create_topic(7) # Politics — nothing tagged found = recent_unsimplified_articles(db_session, self.lang.id, other_topic.id) self.assertEqual(found, []) + + def test_freshness_uses_crawled_at_not_published_time(self): + # A child simplified just now but inheriting an old (backdated) parent + # published_time must still count as fresh — the whole point of the fix. + from zeeguu.core.test.rules.article_rule import ArticleRule + from zeeguu.core.model import ArticleTopicMap + from zeeguu.core.model.article_topic_map import TopicOriginType + from zeeguu.core.content_retriever.backfill import fresh_simplified_count + + old_parent = ArticleRule().article + old_parent.language = self.lang + old_parent.parent_article_id = None + old_parent.broken = 0 + old_parent.crawled_at = datetime.datetime.now() - datetime.timedelta(days=20) + old_parent.published_time = datetime.datetime.now() - datetime.timedelta(days=20) + db_session.add(ArticleTopicMap(old_parent, self.topic, TopicOriginType.HARDSET)) + db_session.commit() + + child = ArticleRule().article + child.language = self.lang + child.parent_article_id = old_parent.id + child.broken = 0 + child.published_time = old_parent.published_time # backdated, 20d ago + child.crawled_at = datetime.datetime.now() # but simplified now + db_session.add(child) + db_session.commit() + + # setUp already contributed 1 fresh child; this backdated-but-recent one + # makes 2. Under the old published_time filter it would have been 1. + self.assertEqual( + fresh_simplified_count(db_session, self.lang.id, self.topic.id), 2 + )