feat: visual top stories, redis caching, and automated 4-hour refresh

src/api/routes/top_stories.py

@@ -14,6 +14,10 @@ from fastapi import APIRouter, Query
 from pydantic import BaseModel
 from datetime import datetime
 
+from src.api.dependencies import get_cache_port, get_live_search_port
+from src.core.ports.cache_port import CachePort
+from src.infrastructure.adapters.duckduckgo_adapter import DuckDuckGoAdapter
+
 try:
     import msgpack
     HAS_MSGPACK = True
@@ -33,6 +37,7 @@ class TopStory(BaseModel):
     published_at: str
     category: str = "NEWS"
     excerpt: Optional[str] = None
+    image_url: Optional[str] = None
     origin: str = "kafka"  # "kafka" or "live"
 
 
@@ -45,9 +50,8 @@ class TopStoriesResponse(BaseModel):
     live_count: int = 0
 
 
-# Simple in-memory cache (2 minutes — shorter TTL for freshness)
-_cache: dict = {}
-_cache_ttl = 120  # 2 minutes
+# Default TTL for top stories (15 minutes — balanced for performance)
+_cache_ttl = 900 
 
 
 # ── Kafka: read latest N messages from news.processed ────────────────────────
@@ -187,6 +191,7 @@ def _fetch_kafka_stories_sync(n: int = 3) -> List[TopStory]:
                     published_at=pub_at,
                     category="NEWS",
                     excerpt=excerpt,
+                    image_url=event.get("image_url") or event.get("thumbnail"),
                     origin="kafka",
                 ))
 
@@ -221,46 +226,32 @@ async def fetch_kafka_stories(n: int = 3) -> List[TopStory]:
 
 # ── DuckDuckGo: fetch N live stories ─────────────────────────────────────────
 
-async def fetch_ddg_stories(n: int = 3) -> List[TopStory]:
-    """Fetch N live stories from DuckDuckGo"""
+async def fetch_live_stories(n: int = 4, adapter: DuckDuckGoAdapter = None) -> List[TopStory]:
+    """Fetch N live stories from DuckDuckGo using the dedicated adapter"""
+    if not adapter:
+        return []
+    
     try:
-        from ddgs import DDGS
-
-        loop = asyncio.get_event_loop()
-
-        def _search():
-            ddgs = DDGS()
-            return list(ddgs.news("Ethiopia", region="et-en", max_results=n))
-
-        raw = await asyncio.wait_for(
-            loop.run_in_executor(None, _search),
-            timeout=5.0
-        )
-
+        # Search for fresh Ethiopia news
+        query = "Ethiopia news today"
+        results = await adapter.search(query)
+        
         stories = []
-        for r in raw:
-            title = r.get("title", "").strip()
-            url   = r.get("url", "").strip()
-            if not title or not url:
-                continue
+        for r in results[:n]:
             stories.append(TopStory(
-                title=title,
-                url=url,
-                source=r.get("source", "Unknown").strip(),
-                published_at=r.get("date", datetime.utcnow().isoformat()),
-                category="NEWS",
-                excerpt=r.get("body", "")[:150] if r.get("body") else None,
+                title=r.get("title", "Untitled"),
+                url=r.get("url", "#"),
+                source=r.get("source", "Live News"),
+                published_at=r.get("published_at", datetime.utcnow().isoformat()),
+                category="BREAKING",
+                excerpt=r.get("content", "")[:150],
+                image_url=r.get("image_url") or r.get("thumbnail"),
                 origin="live",
             ))
-
-        logger.info(f"DuckDuckGo top stories: fetched {len(stories)}")
-        return stories[:n]
-
-    except asyncio.TimeoutError:
-        logger.warning("DuckDuckGo top stories timeout")
-        return []
+        
+        return stories
     except Exception as e:
-        logger.error(f"DuckDuckGo top stories error: {e}")
+        logger.error(f"Live top stories error: {e}")
         return []
 
 
@@ -269,76 +260,67 @@ async def fetch_ddg_stories(n: int = 3) -> List[TopStory]:
 @router.get("/top-stories", response_model=TopStoriesResponse)
 async def get_top_stories(
     force_refresh: bool = Query(default=False, description="Force cache refresh"),
+    cache: CachePort = Depends(get_cache_port),
+    adapter: DuckDuckGoAdapter = Depends(get_live_search_port)
 ):
     """
     Get top 6 news stories for the landing page.
-
-    **Sources:**
-    - 3 from Kafka `news.processed` topic (pipeline-fresh, multilingual)
-    - 3 from DuckDuckGo live search (real-time, English)
-
-    **Cache:** 2-minute TTL for freshness.
+    Combines pipeline-fresh Kafka news with live-search results.
+    Uses Redis for global caching.
     """
-    cache_key = "top_stories_hybrid"
-
-    if not force_refresh and cache_key in _cache:
-        cached_data, cached_time = _cache[cache_key]
-        age = (datetime.utcnow() - cached_time).total_seconds()
-        if age < _cache_ttl:
-            logger.info(f"Top stories cache HIT (age={age:.0f}s)")
-            return TopStoriesResponse(
-                stories=cached_data["stories"],
-                fetched_at=cached_time.isoformat(),
-                cache_hit=True,
-                kafka_count=cached_data["kafka_count"],
-                live_count=cached_data["live_count"],
-            )
+    cache_key = "arki_top_stories_v2"
+
+    if not force_refresh:
+        cached = cache.get(cache_key)
+        if cached:
+            try:
+                data = json.loads(cached)
+                logger.info("Top stories Redis cache HIT")
+                return TopStoriesResponse(
+                    stories=[TopStory(**s) for s in data["stories"]],
+                    fetched_at=data["fetched_at"],
+                    cache_hit=True,
+                    kafka_count=data["kafka_count"],
+                    live_count=data["live_count"],
+                )
+            except Exception as e:
+                logger.warning(f"Failed to parse top stories cache: {e}")
 
     # Fetch both sources in parallel
-    kafka_stories, ddg_stories = await asyncio.gather(
+    kafka_stories, live_stories = await asyncio.gather(
         fetch_kafka_stories(3),
-        fetch_ddg_stories(3),
+        fetch_live_stories(4, adapter),
     )
 
-    # Merge: Kafka first (pipeline-fresh), then DuckDuckGo (live)
-    # Deduplicate by title similarity
+    # Merge and deduplicate
     all_stories: List[TopStory] = []
     seen_titles: set = set()
 
-    for story in kafka_stories + ddg_stories:
+    for story in live_stories + kafka_stories:  # Prioritize live for today's top stories
         title_key = story.title.lower()[:60]
         if title_key not in seen_titles:
             seen_titles.add(title_key)
             all_stories.append(story)
 
-    # Fallback: if Kafka returned nothing, fill with more DuckDuckGo
-    if len(kafka_stories) == 0:
-        extra_ddg = await fetch_ddg_stories(6)
-        for story in extra_ddg:
-            title_key = story.title.lower()[:60]
-            if title_key not in seen_titles and len(all_stories) < 6:
-                seen_titles.add(title_key)
-                all_stories.append(story)
-
-    now = datetime.utcnow()
+    now_iso = datetime.utcnow().isoformat()
+    final_stories = all_stories[:6]
+    
     payload = {
-        "stories": all_stories[:6],
+        "stories": [s.dict() for s in final_stories],
+        "fetched_at": now_iso,
         "kafka_count": len(kafka_stories),
-        "live_count": len(ddg_stories),
+        "live_count": len(live_stories),
     }
-    _cache[cache_key] = (payload, now)
-
-    logger.info(
-        f"Top stories: {len(kafka_stories)} Kafka + {len(ddg_stories)} DuckDuckGo "
-        f"= {len(all_stories[:6])} total"
-    )
+    
+    # Store in Redis
+    cache.set(cache_key, json.dumps(payload), expiration=_cache_ttl)
 
     return TopStoriesResponse(
-        stories=all_stories[:6],
-        fetched_at=now.isoformat(),
+        stories=final_stories,
+        fetched_at=now_iso,
         cache_hit=False,
         kafka_count=len(kafka_stories),
-        live_count=len(ddg_stories),
+        live_count=len(live_stories),
     )