""" 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()