trumpsignal-backend/app/scrapers/truth_social.py

"""
Trump Truth Social scraper — CNN public archive
Source: https://ix.cnn.io/data/truth-social/truth_archive.json
Updated every ~5 minutes by CNN.
"""

import hashlib
import html
import logging
import re
from datetime import datetime, timezone
from typing import Optional

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Post, iso_utc
from app.services.analysis import analyze_post
from app.services.price_store import price_store
from app.ws.manager import manager

logger = logging.getLogger(__name__)

ARCHIVE_URL = "https://ix.cnn.io/data/truth-social/truth_archive.json"

# Liveness tracker — updated every successful poll (even when no new posts).
# Read by /api/health to detect a dead scraper. None = never ran yet.
last_successful_poll_at: Optional[datetime] = None
last_poll_error: Optional[str] = None


def _strip_html(text: str) -> str:
    text = re.sub(r"<[^>]+>", " ", text)
    text = html.unescape(text)
    return re.sub(r"\s+", " ", text).strip()


def _parse_dt(iso: str) -> datetime:
    """Parse Truth Social's ISO timestamp into a naive-UTC datetime.

    IMPORTANT: must convert to UTC *before* stripping tzinfo. Otherwise an
    input like '2026-04-24T15:07:48-04:00' would be stored as naive
    15:07:48 and later mis-read as UTC — a silent 4-hour shift.
    """
    try:
        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
        if dt.tzinfo is not None:
            dt = dt.astimezone(timezone.utc)
        return dt.replace(tzinfo=None)
    except Exception:
        return datetime.now(timezone.utc).replace(tzinfo=None)


async def _fetch_archive() -> Optional[list]:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)",
        "Accept": "application/json",
    }
    try:
        async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
            resp = await client.get(ARCHIVE_URL, headers=headers)
            resp.raise_for_status()
            return resp.json()
    except Exception as exc:
        logger.error("Failed to fetch CNN archive: %s", exc)
        return None


async def _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]:
    external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()

    result = await db.execute(select(Post).where(Post.external_id == external_id))
    if result.scalar_one_or_none():
        return None

    text = _strip_html(entry.get("content") or "").strip()
    if not text:
        return None

    published_at = _parse_dt(entry.get("created_at", ""))

    analysis = await analyze_post(text)

    asset = analysis["asset"]
    # Only capture the price AT post time. The m5/m15/m1h peaks are filled in
    # asynchronously by price_impact_monitor as the windows elapse — avoids
    # recording 0.00% because future candles don't exist yet at entry time.
    price_at_post = None
    if asset and analysis["relevant"]:
        price_at_post = price_store.get_price_at(asset, published_at)

    post = Post(
        external_id=external_id,
        text=text,
        source="truth",
        published_at=published_at,
        sentiment=analysis["sentiment"],
        signal=analysis.get("signal"),
        ai_confidence=analysis["confidence"],
        ai_reasoning=analysis.get("reasoning"),
        prefilter_reason=analysis.get("prefilter_reason"),
        analysis_version=analysis.get("analysis_version"),
        relevant=analysis["relevant"],
        # `asset` (BTC/ETH only) feeds the existing price_impact tracker.
        price_impact_asset=asset if analysis["relevant"] else None,
        price_impact_m5=None,    # filled by price_impact_monitor after 5 m
        price_impact_m15=None,   # filled by price_impact_monitor after 15 m
        price_impact_m1h=None,   # filled by price_impact_monitor after 1 h
        price_at_post=price_at_post,
        # v5 routing: AI decides the actual perp to trade. May be SOL/TRUMP/etc.
        target_asset=analysis.get("target_asset"),
        category=analysis.get("category"),
        expected_move_pct=analysis.get("expected_move_pct"),
    )
    db.add(post)
    await db.flush()

    # Register with the live peak tracker so it starts watching immediately.
    if asset and analysis["relevant"] and price_at_post:
        from app.services.price_impact_monitor import register_post
        register_post(
            post_id=post.id,
            asset=asset,
            signal=analysis.get("signal"),
            entry_price=price_at_post,
            published_at=published_at,
        )

    return post


def _post_to_ws_payload(post: Post) -> dict:
    price_impact = None
    if post.price_impact_asset and post.price_at_post is not None:
        # At broadcast time all windows are open — values are null until
        # price_impact_monitor fills them in. Frontend treats null as "pending".
        price_impact = {
            "asset": post.price_impact_asset,
            "m5": post.price_impact_m5,    # None = not yet measured
            "m15": post.price_impact_m15,
            "m1h": post.price_impact_m1h,
            "price_at_post": post.price_at_post,
        }
    return {
        "type": "new_post",
        "post": {
            "id": post.id,
            "text": post.text,
            "source": post.source,
            "published_at": iso_utc(post.published_at),
            "sentiment": post.sentiment,
            "signal": post.signal,
            "ai_confidence": post.ai_confidence,
            "ai_reasoning": post.ai_reasoning,
            "relevant": post.relevant,
            "price_impact": price_impact,
        },
    }


async def poll_truth_social(db_session_factory) -> None:
    global last_successful_poll_at, last_poll_error
    logger.info("Polling CNN Truth Social archive...")
    entries = await _fetch_archive()
    if not entries:
        last_poll_error = "fetch_archive returned empty"
        return

    # Only process the latest 50 entries each poll (archive has 30k+ posts)
    recent = entries[:50]
    logger.info("Checking %d recent entries...", len(recent))

    async with db_session_factory() as db:
        try:
            new_posts = []
            for entry in recent:
                try:
                    post = await _process_entry(entry, db)
                    if post:
                        new_posts.append(post)
                except Exception as exc:
                    logger.error("Error processing 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("Saved new post id=%d: %s", post.id, post.text[:60])
                    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)
            else:
                logger.info("No new posts found.")
            # Mark a successful poll cycle (separate from "found new posts").
            last_successful_poll_at = datetime.now(timezone.utc)
            last_poll_error = None
        except Exception as exc:
            logger.error("Transaction error: %s", exc)
            last_poll_error = f"transaction_error: {exc}"
            await db.rollback()


async def backfill_history(db_session_factory, limit: int = 500) -> None:
    """One-time backfill of historical posts (no Claude analysis, no price impact)."""
    logger.info("Starting historical backfill (limit=%d)...", limit)
    entries = await _fetch_archive()
    if not entries:
        logger.error("Backfill failed: could not fetch archive")
        return

    to_process = entries[:limit]
    saved = 0

    async with db_session_factory() as db:
        try:
            for entry in to_process:
                external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
                result = await db.execute(select(Post).where(Post.external_id == external_id))
                if result.scalar_one_or_none():
                    continue

                text = _strip_html(entry.get("content") or "").strip()
                if not text:
                    continue

                post = Post(
                    external_id=external_id,
                    text=text,
                    source="truth",
                    published_at=_parse_dt(entry.get("created_at", "")),
                    sentiment="neutral",
                    ai_confidence=0,
                    relevant=False,
                )
                db.add(post)
                saved += 1

            await db.commit()
            logger.info("Backfill complete: saved %d posts", saved)
        except Exception as exc:
            logger.error("Backfill error: %s", exc)
            await db.rollback()