diff --git a/src/fabledassistant/services/rss_classifier.py b/src/fabledassistant/services/rss_classifier.py new file mode 100644 index 0000000..6b368e3 --- /dev/null +++ b/src/fabledassistant/services/rss_classifier.py @@ -0,0 +1,147 @@ +""" +RSS item topic classifier. + +Classifies RSS items into topic tags using a fast non-streaming LLM call. +Called from rss.py after new items are stored — fire-and-forget. +""" + +import json +import logging +from datetime import datetime, timezone + +import httpx + +from fabledassistant.config import Config + +logger = logging.getLogger(__name__) + +STANDARD_TOPICS = [ + "technology", "science", "politics", "business", + "health", "environment", "local", "entertainment", "sports", "other", +] + +_CLASSIFY_PROMPT = """\ +Classify each news item into 1-3 topics. Use only topics from this list: {vocab}. +Return ONLY a JSON object mapping item_id (as string) to a list of topics. +Example: {{"1": ["technology", "ai"], "2": ["politics"]}} + +Items: +{items_block}""" + + +async def _llm_classify(prompt: str, model: str) -> str: + """Make a fast non-streaming LLM call and return the raw text response.""" + payload = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": {"num_ctx": 2048, "temperature": 0.0}, + } + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload) + resp.raise_for_status() + return resp.json().get("message", {}).get("content", "") + + +async def classify_items_batch( + items: list[dict], + user_include_topics: list[str], + model: str | None = None, +) -> dict[int, list[str]]: + """ + Classify a batch of RSS items into topic tags. + + Args: + items: list of dicts with 'id', 'title', 'content' + user_include_topics: extra topics from user preferences to add to vocabulary + model: Ollama model name; defaults to Config.OLLAMA_MODEL + + Returns: + dict mapping item_id (int) -> list of topic strings. + Items not returned had classification fail; callers should leave classified_at=NULL. + """ + if not items: + return {} + + if model is None: + model = Config.OLLAMA_MODEL + + vocab = STANDARD_TOPICS + [t for t in user_include_topics if t not in STANDARD_TOPICS] + items_block = "\n".join( + f"[{item['id']}] {item['title']} — {item.get('content', '')[:300]}" + for item in items + ) + prompt = _CLASSIFY_PROMPT.format(vocab=", ".join(vocab), items_block=items_block) + + try: + raw = await _llm_classify(prompt, model) + # Extract JSON from response (LLM may wrap it in markdown) + raw = raw.strip() + if raw.startswith("```"): + raw = raw.split("```")[1] + if raw.startswith("json"): + raw = raw[4:] + parsed = json.loads(raw) + return {int(k): v for k, v in parsed.items() if isinstance(v, list)} + except Exception: + logger.warning("RSS classification failed", exc_info=True) + return {} + + +async def classify_and_store( + item_ids: list[int], + user_id: int, +) -> None: + """ + Classify unclassified RSS items and write results to DB. + Called as a fire-and-forget task from rss.py. + """ + from sqlalchemy import select + + from fabledassistant.models import async_session + from fabledassistant.models.rss_feed import RssItem + from fabledassistant.services.settings import get_setting + + if not item_ids: + return + + # Load the items + async with async_session() as session: + result = await session.execute( + select(RssItem).where(RssItem.id.in_(item_ids)) + ) + items = list(result.scalars().all()) + + if not items: + return + + # Get user's include topics to extend vocabulary + raw_include = await get_setting(user_id, "briefing_include_topics", "[]") + try: + include_topics = json.loads(raw_include) if isinstance(raw_include, str) else [] + except Exception: + include_topics = [] + + model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) + + # Classify in batches of 10 + batch_size = 10 + all_results: dict[int, list[str]] = {} + for i in range(0, len(items), batch_size): + batch = items[i: i + batch_size] + batch_dicts = [{"id": it.id, "title": it.title, "content": it.content} for it in batch] + results = await classify_items_batch(batch_dicts, include_topics, model=model) + all_results.update(results) + + # Write back to DB + now = datetime.now(timezone.utc) + async with async_session() as session: + for item in items: + item_db = await session.get(RssItem, item.id) + if item_db is None: + continue + topics = all_results.get(item.id) + if topics is not None: + item_db.topics = topics + item_db.classified_at = now + await session.commit() diff --git a/tests/test_rss_service.py b/tests/test_rss_service.py index 9ba0dad..a18a4ed 100644 --- a/tests/test_rss_service.py +++ b/tests/test_rss_service.py @@ -40,3 +40,91 @@ def test_extract_item_prefers_content_over_summary(): entry.published_parsed = None item = extract_item(entry) assert item["content"] == "Full content here" + + +@pytest.mark.asyncio +async def test_classify_items_batch_returns_topic_map(): + """classify_items_batch should return a dict mapping item_id to topic list.""" + from unittest.mock import AsyncMock + from fabledassistant.services.rss_classifier import classify_items_batch + + fake_response = '{"1": ["technology", "ai"], "2": ["politics"]}' + + with patch( + "fabledassistant.services.rss_classifier._llm_classify", + new_callable=AsyncMock, + return_value=fake_response, + ): + items = [ + {"id": 1, "title": "OpenAI releases GPT-5", "content": "..."}, + {"id": 2, "title": "EU passes new law", "content": "..."}, + ] + result = await classify_items_batch(items, user_include_topics=[]) + assert result[1] == ["technology", "ai"] + assert result[2] == ["politics"] + + +@pytest.mark.asyncio +async def test_classify_items_batch_handles_llm_failure(): + """classify_items_batch should return empty dict on LLM error.""" + from unittest.mock import AsyncMock + from fabledassistant.services.rss_classifier import classify_items_batch + + with patch( + "fabledassistant.services.rss_classifier._llm_classify", + new_callable=AsyncMock, + side_effect=Exception("LLM unavailable"), + ): + items = [{"id": 5, "title": "Some news", "content": ""}] + result = await classify_items_batch(items, user_include_topics=[]) + assert result == {} + + +def test_score_rss_items_excludes_blacklisted_topics(): + """Items with excluded topics should be removed.""" + from fabledassistant.services.briefing_preferences import score_and_filter_items + + items = [ + {"id": 1, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T08:00:00"}, + {"id": 2, "title": "Sports score", "topics": ["sports"], "published_at": "2026-03-25T08:00:00"}, + ] + result = score_and_filter_items( + items, + include_topics=["technology"], + exclude_topics=["sports"], + topic_scores={}, + max_items=10, + ) + ids = [r["id"] for r in result] + assert 1 in ids + assert 2 not in ids + + +def test_score_rss_items_boosts_included_topics(): + """Items matching include_topics should rank higher than neutral items.""" + from fabledassistant.services.briefing_preferences import score_and_filter_items + + items = [ + {"id": 1, "title": "Random news", "topics": ["other"], "published_at": "2026-03-25T07:00:00"}, + {"id": 2, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T06:00:00"}, + ] + result = score_and_filter_items( + items, + include_topics=["technology"], + exclude_topics=[], + topic_scores={}, + max_items=10, + ) + assert result[0]["id"] == 2 + + +def test_score_rss_items_no_preferences_returns_all(): + """With no preferences, all items should be returned sorted by recency.""" + from fabledassistant.services.briefing_preferences import score_and_filter_items + + items = [ + {"id": 1, "title": "A", "topics": [], "published_at": "2026-03-24T10:00:00"}, + {"id": 2, "title": "B", "topics": [], "published_at": "2026-03-25T10:00:00"}, + ] + result = score_and_filter_items(items, [], [], {}, max_items=10) + assert result[0]["id"] == 2 # Newer first