trumpsignal-backend/app/scrapers/trumpstruth.py

"""
Trump Truth Social scraper — trumpstruth.org RSS fallback.

Why a second scraper?
  CNN's archive (the primary source) sometimes lags 5–10 minutes behind real
  posts. trumpstruth.org publishes an RSS feed of the same Truth Social account
  that often updates faster. Running both in parallel and deduping by the
  Truth Social `originalId` gives us "min(latency_a, latency_b)" — i.e. whoever
  sees the post first wins.

Dedup strategy:
  Both CNN and trumpstruth expose the underlying Truth Social post id. We
  hash it the same way (md5(str(id))) so the second source is a no-op when
  the first already inserted the row.

Source: https://www.trumpstruth.org/feed (RSS 2.0 with custom truth:originalId tag)
"""

import hashlib
import logging
import re
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from typing import Optional

import httpx

from app.scrapers.truth_social import _process_entry, _post_to_ws_payload
from app.ws.manager import manager

logger = logging.getLogger(__name__)

FEED_URL = "https://www.trumpstruth.org/feed"
NS = {
    "atom": "http://www.w3.org/2005/Atom",
    "truth": "https://truthsocial.com/ns",
}

# Liveness — read by /api/health/deep
last_successful_poll_at: Optional[datetime] = None
last_poll_error: Optional[str] = None


async def _fetch_feed() -> Optional[str]:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)",
        "Accept": "application/rss+xml, application/xml",
    }
    try:
        async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
            resp = await client.get(FEED_URL, headers=headers)
            resp.raise_for_status()
            return resp.text
    except Exception as exc:
        # Include type name — httpx often raises bare ConnectError/RemoteProtocolError
        # with empty .args, which formats as just "Failed to fetch ..." with no body.
        logger.warning("Failed to fetch trumpstruth.org feed: %s (%s)",
                       type(exc).__name__, exc)
        return None


_HTML_TAG = re.compile(r"<[^>]+>")


def _to_cnn_shape(item: ET.Element) -> Optional[dict]:
    """Convert one <item> from the RSS feed into the dict shape the existing
    `_process_entry` expects (CNN archive format).

    Required output keys: id, created_at (ISO), content (HTML)."""
    orig_id_el = item.find("truth:originalId", NS)
    if orig_id_el is None or not (orig_id_el.text or "").strip():
        return None
    orig_id = orig_id_el.text.strip()

    pub_el = item.find("pubDate")
    if pub_el is None or not pub_el.text:
        return None
    try:
        dt = parsedate_to_datetime(pub_el.text)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        else:
            dt = dt.astimezone(timezone.utc)
        created_iso = dt.isoformat().replace("+00:00", "Z")
    except Exception:
        return None

    desc_el = item.find("description")
    content_html = (desc_el.text or "").strip() if desc_el is not None else ""

    return {
        "id": orig_id,
        "created_at": created_iso,
        "content": content_html,
    }


async def poll_trumpstruth(db_session_factory) -> None:
    """One poll cycle. Called by APScheduler.

    Idempotent: posts already inserted by the CNN scraper are skipped via the
    `external_id` uniqueness check inside `_process_entry`.
    """
    global last_successful_poll_at, last_poll_error

    raw = await _fetch_feed()
    if raw is None:
        last_poll_error = "fetch_feed returned None"
        return

    try:
        root = ET.fromstring(raw)
    except ET.ParseError as exc:
        logger.warning("trumpstruth RSS parse error: %s", exc)
        last_poll_error = f"parse: {exc}"
        return

    items = root.findall(".//item")
    if not items:
        last_poll_error = "feed had no <item>"
        return

    # Same as CNN: process the latest 50 only — keeps poll fast and avoids
    # re-scanning the whole feed every 15s.
    recent = items[:50]
    entries = [e for e in (_to_cnn_shape(it) for it in recent) if e]

    async with db_session_factory() as db:
        try:
            new_posts = []
            for entry in entries:
                try:
                    post = await _process_entry(entry, db)
                    if post:
                        new_posts.append(post)
                except Exception as exc:
                    logger.error("trumpstruth: error on entry %s: %s",
                                 entry.get("id"), exc)

            if new_posts:
                await db.commit()
                for post in new_posts:
                    await manager.broadcast(_post_to_ws_payload(post))
                    logger.info("[trumpstruth] beat CNN — new post id=%d: %s",
                                post.id, post.text[:60])
                    # Telegram fan-out — matches truth_social.py. Without this,
                    # whichever poller wins the race determines whether users
                    # get pushed — flaky 50% delivery.
                    try:
                        from app.services.telegram import notify_signal
                        notify_signal(post)
                    except Exception as exc:
                        logger.warning("Telegram notify failed for post %d: %s", post.id, exc)
                    try:
                        from app.services.bot_engine import process_post
                        await process_post(post, db)
                    except Exception as exc:
                        logger.error("process_post failed for post %d: %s",
                                     post.id, exc)
            last_successful_poll_at = datetime.now(timezone.utc)
            last_poll_error = None
        except Exception as exc:
            logger.error("trumpstruth transaction error: %s", exc)
            last_poll_error = f"transaction: {exc}"
            await db.rollback()