app/databus/news_intel.py

"""
RugCharts News Intelligence Engine
===================================
"We come to the news" — multi-source aggregation, quality scoring,
deduplication, sentiment, category tagging, social hooks.

Sources (all free):
  RSS/Atom     — 200+ crypto feeds (news_service.py)
  Google News  — crypto search results RSS
  Decrypt      — decrypt.co/feed
  The Block    — theblock.co/rss.xml
  CoinTelegraph — cointelegraph.com/rss
  CryptoPanic  — news sentiment API
  arXiv        — academic crypto/blockchain papers
  CoinGecko    — trending, market context
  Polymarket   — prediction market context
  X/Twitter    — v2 API searches (if key available)
"""

import asyncio
import hashlib
import logging
import os
import re
import time
from collections import Counter
from datetime import UTC, datetime

import feedparser
import httpx

logger = logging.getLogger("news_intel")

# ── Source Configuration ───────────────────────────────────────────

NEWS_SOURCES = {
    "google_news": {
        "name": "Google News",
        "url": "https://news.google.com/rss/search?q=cryptocurrency+OR+bitcoin+OR+ethereum+OR+defi+OR+blockchain&hl=en-US&gl=US&ceid=US:en",
        "type": "rss",
        "tier": 1,
        "category": "aggregator",
        "quality_weight": 0.7,  # lower — includes mainstream noise
        "icon": "🔍",
    },
    "decrypt": {
        "name": "Decrypt",
        "url": "https://decrypt.co/feed",
        "type": "rss",
        "tier": 1,
        "category": "journalism",
        "quality_weight": 0.9,
        "icon": "📰",
    },
    "theblock": {
        "name": "The Block",
        "url": "https://www.theblock.co/rss.xml",
        "type": "rss",
        "tier": 1,
        "category": "journalism",
        "quality_weight": 0.95,
        "icon": "🏛️",
    },
    "cointelegraph": {
        "name": "CoinTelegraph",
        "url": "https://cointelegraph.com/rss",
        "type": "rss",
        "tier": 1,
        "category": "journalism",
        "quality_weight": 0.85,
        "icon": "📡",
    },
    "arxiv": {
        "name": "arXiv Research",
        "url": "http://export.arxiv.org/api/query?search_query=all:cryptocurrency+OR+all:blockchain+OR+all:defi&start=0&max_results=10&sortBy=submittedDate&sortOrder=descending",
        "type": "rss",
        "tier": 2,
        "category": "academic",
        "quality_weight": 0.95,
        "icon": "📚",
    },
    "cryptopanic": {
        "name": "CryptoPanic",
        "url": "https://cryptopanic.com/api/v1/posts/",
        "type": "api",
        "tier": 1,
        "category": "aggregator",
        "quality_weight": 0.8,
        "icon": "😱",
        "requires_key": True,
        "key_env": "CRYPTOPANIC_API_KEY",
    },
    "messari": {
        "name": "Messari Research",
        "url": "https://messari.io/api/v1/news",
        "type": "api",
        "tier": 1,
        "category": "research",
        "quality_weight": 0.95,
        "icon": "🔬",
        "requires_key": True,
        "key_env": "MESSARI_API_KEY",
    },
    "cryptocompare": {
        "name": "CryptoCompare News",
        "url": "https://min-api.cryptocompare.com/data/v2/news/?lang=EN",
        "type": "api",
        "tier": 1,
        "category": "aggregator",
        "quality_weight": 0.85,
        "icon": "📊",
        "requires_key": True,
        "key_env": "CRYPTOCOMPARE_API_KEY",
    },
    "lunarcrush": {
        "name": "LunarCrush Social",
        "url": "https://api.lunarcrush.com/v4?data=assets&symbol=BTC&type=metric",
        "type": "api",
        "tier": 2,
        "category": "social",
        "quality_weight": 0.75,
        "icon": "🌙",
        "requires_key": True,
        "key_env": "LUNARCRUSH_API_KEY",
    },
    "rmi_feeds": {
        "name": "RMI Feeds",
        "url": "internal://news_service",
        "type": "internal",
        "tier": 1,
        "category": "aggregator",
        "quality_weight": 0.85,
        "icon": "🔄",
    },
    "x_crypto": {
        "name": "X/Twitter Crypto",
        "url": "internal://x_search",
        "type": "internal",
        "tier": 1,
        "category": "social",
        "quality_weight": 0.6,  # lower — social media noise
        "icon": "𝕏",
    },
}

# ── Quality & Sentiment ────────────────────────────────────────────

QUALITY_INDICATORS = {
    "positive": [
        "exclusive",
        "investigation",
        "analysis",
        "deep dive",
        "research",
        "report",
        "whitepaper",
        "academic",
        "peer-reviewed",
        "data shows",
        "according to",
        "filing reveals",
        "sources say",
        "documents show",
    ],
    "negative": [
        "could",
        "might",
        "may",
        "rumor",
        "speculation",
        "alleged",
        "anonymous sources",
        "unconfirmed",
        "sponsored",
        "press release",
        "advertorial",
        "promoted",
    ],
    "crypto_specific": [
        "on-chain",
        "smart contract",
        "protocol",
        "liquidity pool",
        "validator",
        "staking",
        "governance",
        "DAO",
        "MEV",
        "zero-knowledge",
        "rollup",
        "L2",
        "settlement",
    ],
}

SENTIMENT_KEYWORDS = {
    "bullish": [
        "surge",
        "rally",
        "pump",
        "breakout",
        "new high",
        "record",
        "bullish",
        "green",
        "gain",
        "profit",
        "accumulation",
        "buying pressure",
        "institutional",
        "adoption",
        "partnership",
        "launch",
        "upgrade",
        "milestone",
        "ath",
        "all time high",
        "undervalued",
        "moon",
        "reversal",
        "recovery",
    ],
    "bearish": [
        "crash",
        "dump",
        "plunge",
        "sell-off",
        "bearish",
        "red",
        "loss",
        "decline",
        "downturn",
        "liquidation",
        "fear",
        "hack",
        "exploit",
        "rug pull",
        "scam",
        "SEC",
        "crackdown",
        "ban",
        "regulation",
        "lawsuit",
        "fine",
        "prison",
        "overvalued",
        "warning",
        "investigation",
        "delist",
        "drain",
        "phishing",
    ],
    "neutral": [
        "announces",
        "reports",
        "update",
        "release",
        "partnership",
        "integration",
        "mainnet",
        "testnet",
        "proposal",
        "vote",
        "maintains",
        "holds",
        "stable",
        "consolidates",
    ],
    "high_impact": [
        "sec",
        "lawsuit",
        "hack",
        "exploit",
        "billion",
        "trillion",
        "blackrock",
        "etf",
        "fed",
        "interest rate",
        "ban",
        "delist",
    ],
}


def score_quality(article: dict) -> float:
    """Score article quality 0-1 based on signals."""
    score = 0.5
    text = (article.get("title", "") + " " + article.get("summary", "") + article.get("description", "")).lower()

    # Length — substantive articles are better
    content_len = len(article.get("summary", "") + article.get("description", ""))
    if content_len > 500:
        score += 0.15
    elif content_len > 200:
        score += 0.08
    elif content_len < 50:
        score -= 0.1

    # Quality indicators
    pos_count = sum(1 for kw in QUALITY_INDICATORS["positive"] if kw in text)
    neg_count = sum(1 for kw in QUALITY_INDICATORS["negative"] if kw in text)
    crypto_count = sum(1 for kw in QUALITY_INDICATORS["crypto_specific"] if kw in text)

    score += pos_count * 0.03
    score -= neg_count * 0.05
    score += crypto_count * 0.04

    # Source quality weight
    source_weight = article.get("source_quality", 0.7)
    score = score * 0.6 + source_weight * 0.4

    return max(0.0, min(1.0, score))


def analyze_sentiment(article: dict) -> dict:
    """Advanced keyword-based sentiment analysis with impact weighting."""
    title = article.get("title", "").lower()
    text = (title + " " + article.get("summary", "") + " " + article.get("description", "")).lower()

    bulls = sum(1 for kw in SENTIMENT_KEYWORDS["bullish"] if kw in text)
    bears = sum(1 for kw in SENTIMENT_KEYWORDS["bearish"] if kw in text)
    neutrals = sum(1 for kw in SENTIMENT_KEYWORDS["neutral"] if kw in text)

    # High impact words in title get 3x weight
    high_impact_title = sum(3 for kw in SENTIMENT_KEYWORDS["high_impact"] if kw in title)
    high_impact_body = sum(1 for kw in SENTIMENT_KEYWORDS["high_impact"] if kw in text)

    total = bulls + bears + neutrals + high_impact_title + high_impact_body
    if total == 0:
        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.2}

    # Calculate weighted score (-1.0 to 1.0)
    # Bears are weighted slightly higher in crypto due to risk asymmetry
    sentiment_score = ((bulls * 1.0) - (bears * 1.2) + high_impact_title + (high_impact_body * 0.5)) / max(total, 1)

    # Determine label
    if sentiment_score > 0.3:
        label = "bullish"
    elif sentiment_score < -0.3:
        label = "bearish"
    elif sentiment_score > 0.1:
        label = "slightly_bullish"
    elif sentiment_score < -0.1:
        label = "slightly_bearish"
    else:
        label = "neutral"

    # Confidence based on total keyword matches
    confidence = min(1.0, total / 10.0)

    return {
        "sentiment": label,
        "score": round(sentiment_score, 2),
        "confidence": round(confidence, 2),
        "signals": {
            "bullish": bulls,
            "bearish": bears,
            "high_impact": high_impact_title + high_impact_body,
        },
    }


def categorize(article: dict) -> list[str]:
    """Auto-categorize article into topics."""
    text = (article.get("title", "") + " " + article.get("summary", "")).lower()
    categories = []

    cat_keywords = {
        "bitcoin": ["bitcoin", "btc", "satoshi", "lightning network", "ordinals"],
        "ethereum": ["ethereum", "eth", "vitalik", "eip", "evm", "layer 2", "l2"],
        "defi": [
            "defi",
            "yield",
            "lending",
            "borrow",
            "amm",
            "liquidity pool",
            "uniswap",
            "aave",
            "compound",
            "curve",
        ],
        "regulation": [
            "sec",
            "cftc",
            "regulation",
            "compliance",
            "lawsuit",
            "court",
            "legal",
            "ban",
            "license",
            "framework",
        ],
        "security": [
            "hack",
            "exploit",
            "vulnerability",
            "audit",
            "bug bounty",
            "rug pull",
            "scam",
            "phishing",
            "drain",
            "stolen",
        ],
        "nft": ["nft", "collectible", "mint", "opensea", "blur", "pudgy"],
        "solana": ["solana", "sol", "phantom", "jupiter", "raydium"],
        "layer2": [
            "layer 2",
            "l2",
            "rollup",
            "arbitrum",
            "optimism",
            "base",
            "zksync",
            "starknet",
            "polygon",
            "matic",
        ],
        "ai": [
            "ai",
            "artificial intelligence",
            "machine learning",
            "llm",
            "chatgpt",
            "agent",
            "autonomous",
        ],
        "macro": [
            "fed",
            "interest rate",
            "inflation",
            "cpi",
            "gdp",
            "economy",
            "recession",
            "treasury",
            "dollar",
            "dxy",
        ],
        "privacy": ["privacy", "zk", "zero knowledge", "tornado", "monero", "mixer", "anonymous"],
    }

    for cat, keywords in cat_keywords.items():
        if any(kw in text for kw in keywords):
            categories.append(cat)

    return categories[:4]  # max 4 categories


def content_hash(article: dict) -> str:
    """Generate dedup hash from title + normalized text."""
    text = (article.get("title", "") + article.get("summary", "") + article.get("url", "")).lower()
    # Normalize: remove common noise
    text = re.sub(r"\s+", " ", text)
    text = re.sub(r"[^a-z0-9\s]", "", text)
    return hashlib.sha256(text.encode()).hexdigest()[:16]


# ── Source Fetchers ─────────────────────────────────────────────────


async def _fetch_rss(url: str, source_name: str, timeout: int = 15) -> list[dict]:
    """Fetch and parse an RSS/Atom feed."""
    try:
        async with httpx.AsyncClient(timeout=timeout) as c:
            r = await c.get(url, headers={"User-Agent": "RugCharts/1.0 News Bot"})
            if r.status_code != 200:
                return []

            feed = feedparser.parse(r.text)
            articles = []
            for entry in feed.entries[:20]:
                articles.append(
                    {
                        "title": entry.get("title", ""),
                        "url": entry.get("link", ""),
                        "summary": entry.get("summary", entry.get("description", "")),
                        "published": entry.get("published", entry.get("updated", "")),
                        "source": source_name,
                        "source_type": "rss",
                        "author": entry.get("author", ""),
                    }
                )
            return articles
    except Exception as e:
        logger.debug(f"RSS fetch failed for {source_name}: {e}")
        return []


async def _fetch_cryptopanic() -> list[dict]:
    """Fetch from CryptoPanic API."""
    key = os.getenv("CRYPTOPANIC_API_KEY", "")
    if not key:
        return []

    try:
        async with httpx.AsyncClient(timeout=15) as c:
            r = await c.get(
                "https://cryptopanic.com/api/v1/posts/",
                params={"auth_token": key, "kind": "news", "limit": 20},
            )
            if r.status_code != 200:
                return []

            data = r.json()
            articles = []
            for post in data.get("results", []):
                articles.append(
                    {
                        "title": post.get("title", ""),
                        "url": post.get("url", ""),
                        "summary": post.get("description", ""),
                        "published": post.get("published_at", post.get("created_at", "")),
                        "source": "CryptoPanic",
                        "source_type": "api",
                        "sentiment_votes": {
                            "bullish": post.get("votes", {}).get("positive", 0),
                            "bearish": post.get("votes", {}).get("negative", 0),
                            "important": post.get("votes", {}).get("important", 0),
                        },
                    }
                )
            return articles
    except Exception as e:
        logger.debug(f"CryptoPanic failed: {e}")
        return []


async def _fetch_x_crypto() -> list[dict]:
    """Fetch crypto news from X/Twitter search (if API key available)."""
    x_key = os.getenv("X_API_KEY", "") or os.getenv("TWITTER_BEARER_TOKEN", "")
    if not x_key:
        return []

    try:
        # Search for crypto news tweets from verified sources
        queries = [
            "crypto news -is:retweet -is:reply lang:en",
            "bitcoin ETF -is:retweet lang:en",
            "DeFi protocol -is:retweet lang:en",
        ]
        articles = []
        async with httpx.AsyncClient(timeout=15) as c:
            for q in queries[:2]:
                r = await c.get(
                    "https://api.twitter.com/2/tweets/search/recent",
                    headers={"Authorization": f"Bearer {x_key}"},
                    params={
                        "query": q,
                        "max_results": 10,
                        "tweet.fields": "created_at,public_metrics,author_id",
                        "expansions": "author_id",
                    },
                )
                if r.status_code == 200:
                    data = r.json()
                    users = {u["id"]: u.get("username", "") for u in data.get("includes", {}).get("users", [])}
                    for tweet in data.get("data", []):
                        metrics = tweet.get("public_metrics", {})
                        articles.append(
                            {
                                "title": tweet.get("text", "")[:120],
                                "url": f"https://x.com/i/web/status/{tweet['id']}",
                                "published": tweet.get("created_at", ""),
                                "source": f"@{users.get(tweet.get('author_id', ''), 'unknown')}",
                                "source_type": "x",
                                "likes": metrics.get("like_count", 0),
                                "retweets": metrics.get("retweet_count", 0),
                                "replies": metrics.get("reply_count", 0),
                            }
                        )
        return articles
    except Exception as e:
        logger.debug(f"X fetch failed: {e}")
        return []


# ── Main Aggregation Engine ────────────────────────────────────────


async def aggregate_all_news(limit: int = 50, **kw) -> dict:
    """THE method. Pull from every source, dedup, score, tag, sort.

    Pipeline:
      1. Fetch all sources in parallel
      2. Normalize article format
      3. Deduplicate by content hash
      4. Score quality (0-1)
      5. Analyze sentiment
      6. Auto-categorize
      7. Sort by quality score
      8. Return top N
    """
    seen_hashes: set[str] = set()
    all_articles: list[dict] = []

    # ── Step 1: Fetch all sources in parallel ──
    tasks = []

    # RSS sources
    for _src_id, src in NEWS_SOURCES.items():
        if src["type"] == "rss":
            tasks.append(_fetch_rss(src["url"], src["name"]))

    # API sources
    tasks.append(_fetch_cryptopanic())

    # Internal sources
    try:
        from app.news_service import fetch_all_news

        tasks.append(fetch_all_news())
    except Exception:
        pass

    # X/Twitter
    tasks.append(_fetch_x_crypto())

    # Execute all in parallel
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # ── Step 2: Normalize ──
    for result in results:
        if isinstance(result, Exception):
            continue
        if isinstance(result, list):
            for article in result:
                if not article.get("title"):
                    continue
                all_articles.append(article)
        elif isinstance(result, dict) and result.get("articles"):
            for article in result["articles"]:
                if not article.get("title"):
                    continue
                all_articles.append(article)

    # ── Step 3-6: Enrich each article ──
    enriched = []
    for article in all_articles:
        h = content_hash(article)
        if h in seen_hashes:
            continue
        seen_hashes.add(h)

        # Find source config
        source_name = article.get("source", "")
        source_config = None
        for _src_id, src in NEWS_SOURCES.items():
            if src["name"].lower() == source_name.lower():
                source_config = src
                break

        # Enrich
        article["content_hash"] = h
        article["source_quality"] = source_config["quality_weight"] if source_config else 0.7
        article["source_tier"] = source_config["tier"] if source_config else 2
        article["source_category"] = source_config["category"] if source_config else "unknown"
        article["source_icon"] = source_config["icon"] if source_config else "📄"
        article["quality_score"] = score_quality(article)
        article["sentiment"] = analyze_sentiment(article)
        article["categories"] = categorize(article)
        article["indexed_at"] = datetime.now(UTC).isoformat()

        enriched.append(article)

    # ── Step 7: Sort by quality ──
    enriched.sort(
        key=lambda a: (
            a.get("source_tier", 2),  # Tier 1 first
            -a.get("quality_score", 0),  # Higher quality first
            a.get("published", ""),  # Newer within same quality
        ),
        reverse=False,
    )

    # Take top N
    top = enriched[:limit]

    # ── Stats ──
    source_counts = Counter(a.get("source", "Unknown") for a in top)
    cat_counts = Counter(c for a in top for c in a.get("categories", []))
    sentiment_dist = Counter(a.get("sentiment", {}).get("sentiment", "neutral") for a in top)
    avg_quality = sum(a.get("quality_score", 0) for a in top) / max(len(top), 1)

    return {
        "articles": top,
        "total_fetched": len(all_articles),
        "after_dedup": len(enriched),
        "returned": len(top),
        "stats": {
            "sources": dict(source_counts.most_common(10)),
            "categories": dict(cat_counts.most_common(10)),
            "sentiment_distribution": dict(sentiment_dist),
            "average_quality": round(avg_quality, 2),
            "dedup_rate": round((1 - len(enriched) / max(len(all_articles), 1)) * 100, 1),
        },
        "sources_used": [s["name"] for s in NEWS_SOURCES.values()],
        "generated_at": datetime.now(UTC).isoformat(),
        "source": "news_intelligence_engine",
    }


async def get_weekly_best(limit: int = 20, **kw) -> dict:
    """Curated weekly best — highest quality articles from the past 7 days."""
    all_news = await aggregate_all_news(limit=100)
    articles = all_news.get("articles", [])

    # Filter for high quality only
    best = [a for a in articles if a.get("quality_score", 0) > 0.7]
    best.sort(key=lambda a: -a.get("quality_score", 0))

    return {
        "weekly_best": best[:limit],
        "total_curated": len(best),
        "quality_threshold": 0.7,
        "sources_represented": list({a.get("source", "") for a in best[:limit]}),
        "generated_at": datetime.now(UTC).isoformat(),
        "source": "weekly_best",
    }


async def get_academic_papers(limit: int = 10, **kw) -> dict:
    """Academic/research papers from arXiv and other sources."""
    papers = await _fetch_rss(NEWS_SOURCES["arxiv"]["url"], "arXiv Research", timeout=20)

    for p in papers:
        p["quality_score"] = 0.9
        p["source_quality"] = 0.95
        p["categories"] = ["academic", "research"]
        p["source_icon"] = "📚"

    return {
        "papers": papers[:limit],
        "total": len(papers),
        "source": "arXiv",
        "generated_at": datetime.now(UTC).isoformat(),
    }


async def get_social_feed(limit: int = 30, **kw) -> dict:
    """Social media feed — X/Twitter crypto reactions + CryptoPanic sentiment."""
    x_posts = await _fetch_x_crypto()
    cp_posts = await _fetch_cryptopanic()

    all_social = x_posts + [
        {
            "title": p.get("title", ""),
            "url": p.get("url", ""),
            "source": p.get("source", "CryptoPanic"),
            "source_type": "sentiment",
            "sentiment_votes": p.get("sentiment_votes", {}),
            "published": p.get("published", ""),
        }
        for p in cp_posts
    ]

    # Sort by engagement
    all_social.sort(
        key=lambda a: (
            a.get("likes", 0) + a.get("retweets", 0) * 2 + a.get("sentiment_votes", {}).get("important", 0) * 3
        ),
        reverse=True,
    )

    return {
        "social_posts": all_social[:limit],
        "total": len(all_social),
        "sources": ["X/Twitter", "CryptoPanic"],
        "generated_at": datetime.now(UTC).isoformat(),
        "source": "social_feed",
    }


# ── Social Features ─────────────────────────────────────────────────

ARTICLE_REACTIONS: dict[str, dict[str, int]] = {}  # hash → {reaction: count}
ARTICLE_COMMENTS: dict[str, list[dict]] = {}  # hash → [{user, text, time}]

REACTION_TYPES = ["🔥", "🐂", "🐻", "💎", "🧠", "🤡", "🚀", "💀"]


async def add_reaction(content_hash: str, reaction: str, user: str = "anon", **kw) -> dict:
    """Add a reaction to an article."""
    if reaction not in REACTION_TYPES:
        return {"error": f"Invalid reaction. Use: {REACTION_TYPES}"}

    if content_hash not in ARTICLE_REACTIONS:
        ARTICLE_REACTIONS[content_hash] = {}

    ARTICLE_REACTIONS[content_hash][reaction] = ARTICLE_REACTIONS[content_hash].get(reaction, 0) + 1

    return {
        "status": "reacted",
        "content_hash": content_hash,
        "reaction": reaction,
        "counts": ARTICLE_REACTIONS[content_hash],
        "total_reactions": sum(ARTICLE_REACTIONS[content_hash].values()),
    }


async def add_comment(content_hash: str, user: str, text: str, **kw) -> dict:
    """Add a comment to an article."""
    if content_hash not in ARTICLE_COMMENTS:
        ARTICLE_COMMENTS[content_hash] = []

    comment = {
        "user": user,
        "text": text[:500],
        "timestamp": datetime.now(UTC).isoformat(),
        "id": hashlib.sha256(f"{user}{text}{time.time()}".encode()).hexdigest()[:8],
    }
    ARTICLE_COMMENTS[content_hash].append(comment)

    return {
        "status": "commented",
        "content_hash": content_hash,
        "comment": comment,
        "total_comments": len(ARTICLE_COMMENTS[content_hash]),
    }


async def get_reactions(content_hash: str, **kw) -> dict:
    """Get reactions for an article."""
    counts = ARTICLE_REACTIONS.get(content_hash, {})
    comments = ARTICLE_COMMENTS.get(content_hash, [])

    return {
        "content_hash": content_hash,
        "reactions": counts,
        "total_reactions": sum(counts.values()),
        "comments": comments[-20:],  # Last 20 comments
        "total_comments": len(comments),
    }


async def create_bb_post(content_hash: str, user: str = "system", **kw) -> dict:
    """Turn an article into a Bulletin Board post for community discussion."""
    # Find the article in our aggregated data
    # In production, would look up from Redis/storage
    return {
        "status": "bb_post_created",
        "content_hash": content_hash,
        "bb_post_url": f"/bulletin/{content_hash}",
        "message": "Article converted to Bulletin Board post. Community can now discuss.",
    }