packages/core/src/models/content.py

# ============================================================================
# agent verf - Content Models
# Version: 0.1.0
# Last Updated: 2026-01-13
#
# Models for content submitted for verification.
# Content can be URLs, images, videos, or raw text.
#
# The system extracts and normalizes content from various sources
# into this common format for verification.
# ============================================================================

from enum import Enum
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, HttpUrl


class ContentType(str, Enum):
    """
    Types of content that can be verified.

    Each type has different extraction and analysis strategies.
    """

    TEXT = "text"        # Plain text, quotes, claims
    IMAGE = "image"      # Static images (jpg, png, etc.)
    VIDEO = "video"      # Videos (mp4, webm, etc.)
    AUDIO = "audio"      # Audio files, podcasts
    MIXED = "mixed"      # Posts with multiple content types


class Platform(str, Enum):
    """
    Source platforms for content.

    Used for:
      - Choosing extraction strategy
      - Platform-specific metadata
      - Analytics
    """

    # Social media
    TWITTER = "twitter"      # X/Twitter
    INSTAGRAM = "instagram"
    TIKTOK = "tiktok"
    FACEBOOK = "facebook"
    YOUTUBE = "youtube"
    REDDIT = "reddit"

    # News/articles
    NEWS = "news"            # News websites
    BLOG = "blog"            # Blogs, Medium, Substack

    # Other
    OTHER = "other"          # Unknown or other sources
    DIRECT = "direct"        # Direct upload (no URL)


class Content(BaseModel):
    """
    Content submitted for verification.

    This represents the input to the verification system.
    Can come from a URL, direct upload, or extracted from DMs.
    """

    # -------------------------------------------------------------------------
    # Source Information
    # -------------------------------------------------------------------------
    source_url: Optional[HttpUrl] = Field(
        default=None,
        description="Original URL where content was found"
    )
    source_url_normalized: Optional[str] = Field(
        default=None,
        description="Normalized URL for caching (removes tracking params, etc.)"
    )
    platform: Platform = Field(
        default=Platform.OTHER,
        description="Platform where content originated"
    )

    # -------------------------------------------------------------------------
    # Content Type & Data
    # -------------------------------------------------------------------------
    content_type: ContentType = Field(
        ...,
        description="Type of content (text, image, video, etc.)"
    )

    # Text content (always present, even if extracted from image/video)
    text: Optional[str] = Field(
        default=None,
        description="Text content or extracted transcript"
    )

    # Media URL (for images/videos)
    media_url: Optional[HttpUrl] = Field(
        default=None,
        description="Direct URL to media file"
    )

    # Local path (if downloaded)
    local_path: Optional[str] = Field(
        default=None,
        description="Local filesystem path to downloaded content"
    )

    # -------------------------------------------------------------------------
    # Fingerprinting (for deduplication)
    # -------------------------------------------------------------------------
    content_hash: Optional[str] = Field(
        default=None,
        description="Perceptual hash (pHash) for images/videos, or text hash"
    )

    # -------------------------------------------------------------------------
    # Metadata
    # -------------------------------------------------------------------------
    author_username: Optional[str] = Field(
        default=None,
        description="Username/handle of content author"
    )
    author_display_name: Optional[str] = Field(
        default=None,
        description="Display name of content author"
    )
    posted_at: Optional[datetime] = Field(
        default=None,
        description="When the content was originally posted"
    )
    extracted_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="When we extracted this content"
    )

    # Engagement metrics (if available)
    likes: Optional[int] = Field(default=None)
    shares: Optional[int] = Field(default=None)
    comments: Optional[int] = Field(default=None)
    views: Optional[int] = Field(default=None)

    # -------------------------------------------------------------------------
    # Extraction Status
    # -------------------------------------------------------------------------
    extraction_method: Optional[str] = Field(
        default=None,
        description="How we extracted this (oembed, ytdlp, playwright, etc.)"
    )
    extraction_success: bool = Field(
        default=True,
        description="Whether extraction was fully successful"
    )
    extraction_notes: Optional[str] = Field(
        default=None,
        description="Notes about extraction (partial success, fallbacks used, etc.)"
    )

    class Config:
        use_enum_values = True


# ============================================================================
# URL Normalization
#
# Normalize URLs for caching - same content should have same normalized URL.
# This lets us avoid re-verifying the same tweet/post.
# ============================================================================

import re
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode


def normalize_url(url: str) -> str:
    """
    Normalize a URL for caching purposes.

    Removes:
      - Tracking parameters (utm_*, fbclid, etc.)
      - Mobile prefixes (m.twitter.com -> twitter.com)
      - Unnecessary query parameters

    Standardizes:
      - Protocol to https
      - Domain casing (lowercase)
      - Path formatting

    Args:
        url: The URL to normalize

    Returns:
        Normalized URL string

    Example:
        >>> normalize_url("http://m.twitter.com/user/status/123?utm_source=foo")
        "https://twitter.com/user/status/123"
    """
    # Parse the URL
    parsed = urlparse(url.strip())

    # Normalize scheme to https
    scheme = "https"

    # Normalize domain
    domain = parsed.netloc.lower()

    # Remove mobile prefixes
    mobile_prefixes = ["m.", "mobile.", "www."]
    for prefix in mobile_prefixes:
        if domain.startswith(prefix):
            domain = domain[len(prefix):]

    # Normalize X/Twitter
    if domain == "x.com":
        domain = "twitter.com"

    # Parse query parameters
    query_params = parse_qs(parsed.query)

    # Remove tracking parameters
    tracking_params = {
        "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
        "fbclid", "gclid", "ref", "source", "ref_src", "ref_url",
        "s", "t", "si",  # Twitter/TikTok tracking
    }
    filtered_params = {
        k: v for k, v in query_params.items()
        if k.lower() not in tracking_params
    }

    # Reconstruct query string
    query = urlencode(filtered_params, doseq=True) if filtered_params else ""

    # Reconstruct URL
    normalized = urlunparse((
        scheme,
        domain,
        parsed.path.rstrip("/"),  # Remove trailing slash
        parsed.params,
        query,
        "",  # Remove fragment
    ))

    return normalized


def detect_platform(url: str) -> Platform:
    """
    Detect the platform from a URL.

    Args:
        url: The URL to analyze

    Returns:
        Platform enum value
    """
    domain = urlparse(url).netloc.lower()

    # Remove www prefix
    if domain.startswith("www."):
        domain = domain[4:]

    # Platform detection
    platform_domains = {
        "twitter.com": Platform.TWITTER,
        "x.com": Platform.TWITTER,
        "instagram.com": Platform.INSTAGRAM,
        "tiktok.com": Platform.TIKTOK,
        "facebook.com": Platform.FACEBOOK,
        "fb.com": Platform.FACEBOOK,
        "youtube.com": Platform.YOUTUBE,
        "youtu.be": Platform.YOUTUBE,
        "reddit.com": Platform.REDDIT,
    }

    for pattern, platform in platform_domains.items():
        if domain == pattern or domain.endswith("." + pattern):
            return platform

    return Platform.OTHER