packages/core/src/graph/nodes/extraction.py

# ============================================================================
# agent verf - Extraction Node
# Version: 0.1.0
# Last Updated: 2026-01-13
#
# The extraction node fetches content from URLs.
#
# Purpose:
#   - Extract content from social media URLs (X/Twitter, Instagram, TikTok, etc.)
#   - Handle different content types (text posts, images, videos)
#   - Normalize content for downstream processing
#
# Extraction strategies (in order of preference):
#   1. oEmbed APIs (free, rate-limited)
#   2. yt-dlp (for video platforms)
#   3. Playwright scraping (fallback, resource-intensive)
#   4. Third-party APIs (Zyte, Bright Data - paid)
#
# This node is designed for graceful degradation - if one method fails,
# it tries the next.
# ============================================================================

import re
import structlog
from typing import Any, Optional
from urllib.parse import urlparse
from datetime import datetime

from src.graph.state import VerificationState
from src.models.content import Content, ContentType, Platform, normalize_url

logger = structlog.get_logger()


# ============================================================================
# Platform Detection
# ============================================================================

PLATFORM_PATTERNS = {
    Platform.TWITTER: [
        r"(twitter\.com|x\.com)/\w+/status/\d+",
        r"(twitter\.com|x\.com)/\w+",
    ],
    Platform.INSTAGRAM: [
        r"instagram\.com/p/[\w-]+",
        r"instagram\.com/reel/[\w-]+",
        r"instagram\.com/[\w.]+",
    ],
    Platform.TIKTOK: [
        r"tiktok\.com/@[\w.]+/video/\d+",
        r"tiktok\.com/[\w]+",
    ],
    Platform.YOUTUBE: [
        r"youtube\.com/watch\?v=[\w-]+",
        r"youtu\.be/[\w-]+",
        r"youtube\.com/shorts/[\w-]+",
    ],
    Platform.REDDIT: [
        r"reddit\.com/r/\w+/comments/\w+",
    ],
    Platform.FACEBOOK: [
        r"facebook\.com/.+/posts/\d+",
        r"fb\.watch/[\w]+",
    ],
}


def detect_platform(url: str) -> Platform:
    """Detect which platform a URL belongs to."""
    for platform, patterns in PLATFORM_PATTERNS.items():
        for pattern in patterns:
            if re.search(pattern, url, re.IGNORECASE):
                return platform
    return Platform.WEB


def detect_content_type(url: str, platform: Platform) -> ContentType:
    """Detect content type based on URL and platform."""
    url_lower = url.lower()

    # Video indicators
    if any(x in url_lower for x in ["/video/", "/shorts/", "/reel/", "youtu"]):
        return ContentType.VIDEO

    # Image indicators
    if any(x in url_lower for x in [".jpg", ".png", ".gif", "/photo/"]):
        return ContentType.IMAGE

    # Platform-specific defaults
    if platform == Platform.TIKTOK:
        return ContentType.VIDEO
    if platform == Platform.YOUTUBE:
        return ContentType.VIDEO

    # Default to text post for social platforms, article for web
    if platform == Platform.WEB:
        return ContentType.ARTICLE

    return ContentType.TEXT_POST


# ============================================================================
# oEmbed Extraction
#
# Free method using platform oEmbed endpoints.
# Limited data but fast and reliable.
# ============================================================================

OEMBED_ENDPOINTS = {
    Platform.TWITTER: "https://publish.twitter.com/oembed",
    Platform.INSTAGRAM: "https://api.instagram.com/oembed",
    Platform.TIKTOK: "https://www.tiktok.com/oembed",
    Platform.YOUTUBE: "https://www.youtube.com/oembed",
}


async def extract_via_oembed(url: str, platform: Platform) -> Optional[Content]:
    """
    Extract content using oEmbed API.

    oEmbed provides:
        - Title/author
        - HTML embed code
        - Thumbnail URL

    Limitations:
        - No full text content
        - Rate limited
        - Some platforms require auth
    """
    if platform not in OEMBED_ENDPOINTS:
        return None

    try:
        import httpx

        endpoint = OEMBED_ENDPOINTS[platform]
        async with httpx.AsyncClient() as client:
            response = await client.get(
                endpoint,
                params={"url": url, "format": "json"},
                timeout=10.0,
            )
            response.raise_for_status()
            data = response.json()

        # Extract what we can from oEmbed response
        return Content(
            source_url=url,
            platform=platform,
            content_type=detect_content_type(url, platform),
            title=data.get("title"),
            author_username=data.get("author_name"),
            author_display_name=data.get("author_name"),
            text_content=data.get("title", ""),  # oEmbed doesn't give full text
            thumbnail_url=data.get("thumbnail_url"),
            extracted_at=datetime.utcnow(),
            extraction_method="oembed",
        )

    except Exception as e:
        logger.warning(
            "oEmbed extraction failed",
            url=url,
            platform=platform,
            error=str(e),
        )
        return None


# ============================================================================
# yt-dlp Extraction
#
# Powerful extraction for video platforms.
# Handles YouTube, TikTok, Twitter videos, etc.
# ============================================================================

async def extract_via_ytdlp(url: str, platform: Platform) -> Optional[Content]:
    """
    Extract content using yt-dlp.

    yt-dlp provides:
        - Full video metadata
        - Thumbnails
        - Description/title
        - View counts, dates

    Best for:
        - YouTube
        - TikTok
        - Twitter videos
    """
    try:
        import yt_dlp
        import asyncio

        ydl_opts = {
            "quiet": True,
            "no_warnings": True,
            "extract_flat": False,
            "skip_download": True,  # Don't download the video
        }

        def _extract():
            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                return ydl.extract_info(url, download=False)

        # Run in thread pool to avoid blocking
        loop = asyncio.get_event_loop()
        info = await loop.run_in_executor(None, _extract)

        if not info:
            return None

        # Build Content from yt-dlp info
        return Content(
            source_url=url,
            platform=platform,
            content_type=ContentType.VIDEO,
            title=info.get("title"),
            text_content=info.get("description", ""),
            author_username=info.get("uploader_id") or info.get("channel_id"),
            author_display_name=info.get("uploader") or info.get("channel"),
            thumbnail_url=info.get("thumbnail"),
            duration_seconds=info.get("duration"),
            view_count=info.get("view_count"),
            like_count=info.get("like_count"),
            published_at=datetime.fromisoformat(info["upload_date"][:4] + "-" + info["upload_date"][4:6] + "-" + info["upload_date"][6:])
                if info.get("upload_date") else None,
            extracted_at=datetime.utcnow(),
            extraction_method="yt-dlp",
        )

    except Exception as e:
        logger.warning(
            "yt-dlp extraction failed",
            url=url,
            platform=platform,
            error=str(e),
        )
        return None


# ============================================================================
# Basic Web Extraction
#
# Fallback for generic web pages.
# Uses httpx + basic HTML parsing.
# ============================================================================

async def extract_web_page(url: str) -> Optional[Content]:
    """
    Extract content from a generic web page.

    Uses:
        - Open Graph meta tags
        - HTML title
        - Meta description

    This is a basic extraction - for full article text,
    consider using readability libraries.
    """
    try:
        import httpx
        from html.parser import HTMLParser

        async with httpx.AsyncClient() as client:
            response = await client.get(
                url,
                timeout=15.0,
                follow_redirects=True,
                headers={
                    "User-Agent": "Mozilla/5.0 (compatible; AgentVerf/1.0)"
                },
            )
            response.raise_for_status()
            html = response.text

        # Simple meta tag extraction
        title = None
        description = None
        og_image = None

        # Extract title
        title_match = re.search(r"<title[^>]*>([^<]+)</title>", html, re.IGNORECASE)
        if title_match:
            title = title_match.group(1).strip()

        # Extract Open Graph tags
        og_title = re.search(r'property=["\']og:title["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
        if og_title:
            title = og_title.group(1)

        og_desc = re.search(r'property=["\']og:description["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
        if og_desc:
            description = og_desc.group(1)

        og_img = re.search(r'property=["\']og:image["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
        if og_img:
            og_image = og_img.group(1)

        # Fallback to meta description
        if not description:
            meta_desc = re.search(r'name=["\']description["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
            if meta_desc:
                description = meta_desc.group(1)

        return Content(
            source_url=url,
            platform=Platform.WEB,
            content_type=ContentType.ARTICLE,
            title=title,
            text_content=description or "",
            thumbnail_url=og_image,
            extracted_at=datetime.utcnow(),
            extraction_method="web_scrape",
        )

    except Exception as e:
        logger.warning(
            "Web page extraction failed",
            url=url,
            error=str(e),
        )
        return None


# ============================================================================
# Extraction Node
# ============================================================================

async def extraction_node(state: VerificationState) -> dict[str, Any]:
    """
    Extraction node - fetches content from URLs.

    Input state fields used:
        - request_url: URL to extract content from

    Output state updates:
        - content: Extracted Content object
        - content_extraction_success: Whether extraction succeeded
        - content_extraction_method: Which method was used
        - current_node: "extraction"
        - meta_tools_used: Updated with extraction tools
        - errors: Updated if extraction fails

    Extraction order:
        1. oEmbed (fast, free)
        2. yt-dlp (for video content)
        3. Web scraping (fallback)

    Returns:
        Dict of state updates to merge
    """
    logger.info(
        "Extraction node starting",
        request_id=str(state.request_id),
        url=state.request_url,
    )

    # If no URL provided, nothing to extract
    if not state.request_url:
        logger.info("No URL provided, skipping extraction")
        return {
            "content_extraction_success": False,
            "content_extraction_method": None,
            "current_node": "extraction",
        }

    # Normalize the URL first
    url = normalize_url(state.request_url)

    # Detect platform and content type
    platform = detect_platform(url)
    content_type = detect_content_type(url, platform)

    logger.info(
        "Detected platform and content type",
        platform=platform,
        content_type=content_type,
    )

    content = None
    tools_used = []

    # Strategy 1: oEmbed (fast, free)
    if platform in OEMBED_ENDPOINTS:
        content = await extract_via_oembed(url, platform)
        if content:
            tools_used.append("oembed")

    # Strategy 2: yt-dlp (for video content)
    if not content and content_type == ContentType.VIDEO:
        content = await extract_via_ytdlp(url, platform)
        if content:
            tools_used.append("yt-dlp")

    # Strategy 3: yt-dlp for supported platforms even if not video
    if not content and platform in [Platform.YOUTUBE, Platform.TIKTOK]:
        content = await extract_via_ytdlp(url, platform)
        if content:
            tools_used.append("yt-dlp")

    # Strategy 4: Web scraping (fallback)
    if not content:
        content = await extract_web_page(url)
        if content:
            tools_used.append("web_scrape")

    # Update state
    if content:
        logger.info(
            "Extraction successful",
            request_id=str(state.request_id),
            method=content.extraction_method,
            has_title=content.title is not None,
            has_text=bool(content.text_content),
        )

        return {
            "content": content,
            "content_extraction_success": True,
            "content_extraction_method": content.extraction_method,
            "current_node": "extraction",
            "meta_tools_used": list(set(state.meta_tools_used + tools_used)),
        }
    else:
        logger.warning(
            "All extraction methods failed",
            request_id=str(state.request_id),
            url=url,
        )

        # Create minimal content object with just the URL
        minimal_content = Content(
            source_url=url,
            platform=platform,
            content_type=content_type,
            text_content=f"[Content extraction failed for URL: {url}]",
            extracted_at=datetime.utcnow(),
            extraction_method="failed",
        )

        return {
            "content": minimal_content,
            "content_extraction_success": False,
            "content_extraction_method": "failed",
            "current_node": "extraction",
            "meta_tools_used": list(set(state.meta_tools_used + tools_used)),
            "errors": state.errors + [f"Content extraction failed for URL: {url}"],
        }