| import requests |
| from errors import get_logger, fmt_exc |
|
|
| log = get_logger("worldmonitor") |
|
|
| |
| WM_INTERNAL_URL = "http://127.0.0.1:8080" |
|
|
|
|
| def get_news_digest(lang: str = "it", max_items: int = 8) -> list[dict]: |
| """Interroga l'istanza locale di World Monitor per un digest di notizie recenti. |
| |
| Ritorna [] se World Monitor non è raggiungibile (es. ancora in avvio) o non |
| ha ancora dati — la ricerca web di GenerAI resta il percorso principale, |
| questo è un arricchimento opzionale. |
| """ |
| try: |
| r = requests.get( |
| f"{WM_INTERNAL_URL}/api/news/v1/list-feed-digest", |
| params={"variant": "full", "lang": lang}, |
| timeout=5, |
| ) |
| if r.status_code != 200: |
| log.debug("World Monitor digest non disponibile (HTTP %s)", r.status_code) |
| return [] |
| data = r.json() |
| except Exception as e: |
| log.debug("World Monitor non raggiungibile: %s", fmt_exc(e)) |
| return [] |
|
|
| items = [] |
| for category in (data.get("categories") or {}).values(): |
| for item in category.get("items", []): |
| title = item.get("title", "").strip() |
| if not title: |
| continue |
| items.append({ |
| "title": title, |
| "url": item.get("link", ""), |
| "source": item.get("source", ""), |
| "score": item.get("importanceScore", 0), |
| }) |
|
|
| items.sort(key=lambda x: -x["score"]) |
| return items[:max_items] |
|
|
|
|
| def find_relevant(query: str, max_items: int = 3) -> list[dict]: |
| """Filtra il digest di World Monitor per gli item con parole in comune con la query.""" |
| words = {w for w in query.lower().split() if len(w) > 3} |
| if not words: |
| return [] |
|
|
| digest = get_news_digest(max_items=40) |
| if not digest: |
| return [] |
|
|
| scored = [] |
| for item in digest: |
| overlap = sum(1 for w in words if w in item["title"].lower()) |
| if overlap > 0: |
| scored.append((overlap, item)) |
|
|
| scored.sort(key=lambda x: -x[0]) |
| return [item for _, item in scored[:max_items]] |
|
|
