AegisSight-Monitor/src/feeds/rss_parser.py

"""RSS-Feed Parser: Durchsucht vorkonfigurierte Feeds nach relevanten Meldungen."""
import asyncio
import logging
import feedparser
import httpx
from datetime import datetime, timezone
from config import TIMEZONE, MAX_ARTICLES_PER_DOMAIN_RSS
from source_rules import _extract_domain
from feeds.transcript_extractors._common import html_to_text
from services.post_refresh_qc import normalize_german_umlauts
from agents.researcher import keywords_for_language, flatten_keywords

logger = logging.getLogger("osint.rss")


def _is_specific_word(w: str) -> bool:
    """Spezifisches Keyword = 1-Treffer reicht für Match.

    - Lateinisch: ab 7 Zeichen (alte Heuristik).
    - Nicht-ASCII (CJK, Arabisch, Hebräisch, Kyrillisch etc.): ab 3 Zeichen.
      Beispiel: '自衛隊' (3 Kanji) oder 'путин' (5 Kyrillisch) sind spezifisch genug.
    """
    if not w:
        return False
    if any(ord(c) > 127 for c in w):
        return len(w) >= 3
    return len(w) >= 7


class RSSParser:
    """Durchsucht RSS-Feeds nach relevanten Artikeln."""

    # Stoppwörter die bei der RSS-Suche ignoriert werden
    STOP_WORDS = {
        "und", "oder", "der", "die", "das", "ein", "eine", "in", "im", "am", "an",
        "auf", "für", "mit", "von", "zu", "zum", "zur", "bei", "nach", "vor",
        "über", "unter", "ist", "sind", "hat", "the", "and", "for", "with", "from",
    }

    @staticmethod
    def _clean_search_words(words: list[str]) -> list[str]:
        """Entfernt rein-numerische Wörter (Jahreszahlen etc.) aus Suchbegriffen."""
        cleaned = [w for w in words if not w.isdigit()]
        return cleaned if cleaned else words

    def _fallback_search_words(self, search_term: str) -> list[str]:
        words = [
            w for w in search_term.lower().split()
            if w not in self.STOP_WORDS and len(w) >= 3
        ]
        if not words:
            words = search_term.lower().split()[:2]
        return self._clean_search_words(words)

    async def search_feeds(self, search_term: str, international: bool = True, tenant_id: int = None, keywords: dict | list | None = None, user_id: int = None) -> list[dict]:
        """Durchsucht RSS-Feeds nach einem Suchbegriff.

        Args:
            search_term: Suchbegriff
            international: Wenn False, nur Feeds in der Org-Sprache + Behoerden (keine internationalen)
            tenant_id: Optionale Org-ID fuer tenant-spezifische Quellen
            keywords: Sprach-Dict {iso_lang: [keyword, ...]} oder flache Liste (Backward).
        """
        all_articles = []
        if keywords:
            logger.info(f"RSS-Suche mit Claude-Keywords (Sprachen): "
                        f"{ {k: len(v) for k, v in keywords.items()} if isinstance(keywords, dict) else len(keywords) }")
            fallback_words = None
        else:
            fallback_words = self._fallback_search_words(search_term)

        rss_feeds = await self._get_rss_feeds(tenant_id=tenant_id)

        # User-spezifische Ausschluesse anwenden
        if user_id:
            try:
                from source_rules import get_user_excluded_domains
                user_excluded = await get_user_excluded_domains(user_id)
                if user_excluded:
                    for cat in rss_feeds:
                        rss_feeds[cat] = [f for f in rss_feeds[cat]
                                          if not any(excl in (f.get("url", "") + f.get("name", "")).lower()
                                                     for excl in user_excluded)]
            except Exception as e:
                logger.warning(f"User-Ausschluesse konnten nicht geladen werden: {e}")

        # Feed-Kategorien filtern
        if international:
            categories = rss_feeds.keys()
        else:
            categories = [c for c in rss_feeds.keys() if c != "international"]

        tasks = []
        for category in categories:
            for feed_config in rss_feeds.get(category, []):
                feed_lang = feed_config.get("primary_language")
                if keywords:
                    words = keywords_for_language(keywords, feed_lang)
                    words = [w.lower() for w in words]
                else:
                    words = fallback_words
                tasks.append(self._fetch_feed(feed_config, words))

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for result in results:
            if isinstance(result, Exception):
                logger.warning(f"Feed-Fehler: {result}")
                continue
            all_articles.extend(result)

        cat_info = "alle" if international else "nur primary + behörden"
        logger.info(f"RSS-Suche nach '{search_term}' ({cat_info}): {len(all_articles)} Treffer")
        all_articles = self._apply_domain_cap(all_articles)
        return all_articles

    async def search_feeds_selective(self, search_term: str, selected_feeds: list[dict], keywords: dict | list | None = None) -> list[dict]:
        """Durchsucht nur die übergebenen Feeds (vorselektiert durch Claude).

        Args:
            search_term: Suchbegriff
            selected_feeds: Liste von Feed-Dicts mit mindestens {"name", "url"} und idealerweise "primary_language"
            keywords: Sprach-Dict {iso_lang: [keyword, ...]} oder flache Liste (Backward).
        """
        all_articles = []
        if keywords:
            if isinstance(keywords, dict):
                logger.info(f"RSS-Selektiv mit Claude-Keywords (Sprachen): "
                            f"{ {k: len(v) for k, v in keywords.items()} }")
            else:
                logger.info(f"RSS-Selektiv mit Claude-Keywords (flach): {keywords}")
            fallback_words = None
        else:
            fallback_words = self._fallback_search_words(search_term)

        tasks = []
        for feed_config in selected_feeds:
            feed_lang = feed_config.get("primary_language")
            if keywords:
                words = keywords_for_language(keywords, feed_lang)
                words = [w.lower() for w in words]
            else:
                words = fallback_words
            tasks.append(self._fetch_feed(feed_config, words))

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for result in results:
            if isinstance(result, Exception):
                logger.warning(f"Feed-Fehler: {result}")
                continue
            all_articles.extend(result)

        logger.info(f"RSS-Selektiv nach '{search_term}': {len(all_articles)} Treffer aus {len(selected_feeds)} Feeds")
        all_articles = self._apply_domain_cap(all_articles)
        return all_articles

    async def _get_rss_feeds(self, tenant_id: int = None) -> dict:
        """Laedt RSS-Feeds aus der Datenbank (global + org-spezifisch)."""
        try:
            from source_rules import get_source_rules
            rules = await get_source_rules(tenant_id=tenant_id)
            return rules.get("rss_feeds", {})
        except Exception as e:
            logger.warning(f"Fallback auf config.py fuer RSS-Feeds: {e}")
            from config import RSS_FEEDS
            return dict(RSS_FEEDS)

    async def _fetch_feed(self, feed_config: dict, search_words: list[str]) -> list[dict]:
        """Einzelnen RSS-Feed abrufen und durchsuchen."""
        name = feed_config["name"]
        url = feed_config["url"]
        articles = []

        try:
            async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
                response = await client.get(url, headers={
                    "User-Agent": "OSINT-Monitor/1.0 (News Aggregator)"
                })
                response.raise_for_status()

            feed = await asyncio.to_thread(feedparser.parse, response.text)

            for entry in feed.entries[:50]:
                title = entry.get("title", "")
                # RSS-summary ist bei vielen Quellen HTML (Guardian, AP, SZ, ...).
                # Vor weiterer Verwendung strippen, sonst landet HTML in DB
                # und KI-Agenten und Sprach-Heuristik werden gestoert.
                summary_raw = entry.get("summary", "")
                summary = html_to_text(summary_raw) if summary_raw else ""
                # ASCII-Umlaut-Normalisierung (z.B. dpa-AFX schreibt "Gespraeche").
                # Dictionary-basiert, sicher gegen englische Woerter wie "Boeing".
                title, _ = normalize_german_umlauts(title)
                summary, _ = normalize_german_umlauts(summary)
                text = f"{title} {summary}".lower()

                # Adaptive Match-Schwelle:
                # - Bei mindestens einem spezifischen Keyword (Latin ≥7 Zeichen oder
                #   CJK/Arabisch/Hebräisch/Kyrillisch ≥3 Zeichen) im Text reicht 1 Treffer.
                #   Damit matched z.B. "自衛隊" (3 Kanji) wie "buckelwal" (9 Zeichen).
                # - Sonst: alte Heuristik (mindestens halb der Wörter, max. 2).
                specific_in_text = any(w in text for w in search_words if _is_specific_word(w))
                if specific_in_text:
                    min_matches = 1
                else:
                    min_matches = min(2, max(1, (len(search_words) + 1) // 2))
                match_count = sum(1 for word in search_words if word in text)

                if match_count >= min_matches:
                    published = None
                    if hasattr(entry, "published_parsed") and entry.published_parsed:
                        try:
                            published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc).astimezone(TIMEZONE).isoformat()
                        except (TypeError, ValueError):
                            pass

                    # Relevanz-Score: Anteil der gematchten Suchworte (0.0-1.0)
                    relevance_score = match_count / len(search_words) if search_words else 0.0

                    articles.append({
                        "headline": title,
                        "headline_de": title if self._is_german(title) else None,
                        "source": name,
                        "source_url": entry.get("link", ""),
                        # Die Quell-Domain aus der DB (z.B. "mod.go.jp"), nicht aus
                        # der URL — relevant für Google-News-RSS-Quellen, deren URLs
                        # alle "news.google.com" sind, obwohl sie für 14 verschiedene
                        # Behörden/Zeitungen stehen. Wird vom Domain-Cap genutzt.
                        "source_domain": feed_config.get("domain") or "",
                        "content_original": summary[:1000] if summary else None,
                        "content_de": summary[:1000] if summary and self._is_german(summary) else None,
                        "language": "de" if self._is_german(title) else "en",
                        "published_at": published,
                        "relevance_score": relevance_score,
                    })

        except Exception as e:
            logger.debug(f"Feed {name} ({url}): {e}")

        return articles

    def _apply_domain_cap(self, articles: list[dict]) -> list[dict]:
        """Begrenzt die Anzahl der Artikel pro Domain auf MAX_ARTICLES_PER_DOMAIN_RSS.

        Gruppiert nach Domain, sortiert pro Domain nach relevance_score (beste zuerst),
        behält nur die Top-N pro Domain.
        """
        if not articles:
            return articles

        # Nach Domain gruppieren. Bevorzugt source_domain (aus dem Feed-Eintrag,
        # z.B. "mod.go.jp" bei einer Google-News-Site-Search-RSS-Quelle), fällt
        # erst dann auf die URL-Domain zurück. Sonst landen alle Google-News-
        # Feeds (14 ja-Quellen) im selben "news.google.com"-Topf und werden
        # vom Cap auf 10 begrenzt.
        by_domain: dict[str, list[dict]] = {}
        for article in articles:
            domain = (article.get("source_domain") or "").strip().lower()
            if not domain:
                domain = _extract_domain(article.get("source_url", ""))
            if not domain:
                domain = "__unknown__"
            by_domain.setdefault(domain, []).append(article)

        capped = []
        for domain, domain_articles in by_domain.items():
            # Nach Relevanz sortieren (beste zuerst)
            domain_articles.sort(key=lambda a: a.get("relevance_score", 0), reverse=True)
            kept = domain_articles[:MAX_ARTICLES_PER_DOMAIN_RSS]
            if len(domain_articles) > MAX_ARTICLES_PER_DOMAIN_RSS:
                logger.info(
                    f"Domain-Cap: {domain} von {len(domain_articles)} auf {MAX_ARTICLES_PER_DOMAIN_RSS} Artikel begrenzt"
                )
            capped.extend(kept)

        if len(capped) < len(articles):
            logger.info(f"Domain-Cap gesamt: {len(articles)} → {len(capped)} Artikel")

        return capped

    def _is_german(self, text: str) -> bool:
        """Einfache Heuristik ob ein Text deutsch ist."""
        german_words = {"der", "die", "das", "und", "ist", "von", "mit", "für", "auf", "ein",
                        "eine", "den", "dem", "des", "sich", "wird", "nach", "bei", "auch",
                        "über", "wie", "aus", "hat", "zum", "zur", "als", "noch", "mehr",
                        "nicht", "aber", "oder", "sind", "vor", "einem", "einer", "wurde"}
        words = set(text.lower().split())
        matches = words & german_words
        return len(matches) >= 2