siftlode/backend/app/sync/videos.py

"""Video ingestion: free RSS detection, recent-first (and deep) backfill from the
uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts
and livestream classification)."""
import re
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone

from sqlalchemy import and_, func, or_, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from app import progress, sysconfig
from app.models import LIVE_OR_UPCOMING, Channel, Video
from app.titles import normalize_title
from app.youtube.client import YouTubeClient, best_thumbnail
from app.youtube.rss import fetch_channel_feed
from app.youtube.shorts import make_client, probe_is_short

_DURATION_RE = re.compile(
    r"P(?:(\d+)D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE
)


def _now() -> datetime:
    return datetime.now(timezone.utc)


def parse_iso8601_duration(value: str | None) -> int | None:
    if not value:
        return None
    match = _DURATION_RE.fullmatch(value)
    if not match:
        return None
    days, hours, minutes, seconds = (int(g) if g else 0 for g in match.groups())
    return days * 86400 + hours * 3600 + minutes * 60 + seconds


def parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def _insert_stubs(db: Session, rows: list[dict]) -> int:
    """Idempotently insert video stubs; returns the number of newly inserted rows.

    Uses RETURNING so the count is exact (ON CONFLICT DO NOTHING leaves rowcount
    unreliable across drivers). Duplicates within the batch are dropped first so a
    single INSERT can't raise CardinalityViolation on the conflict target."""
    if not rows:
        return 0
    seen: dict[str, dict] = {}
    for row in rows:
        # Normalize the title at first insert too (RSS/backfill stubs), so a brand-new video
        # never flashes its raw title before enrichment refreshes it.
        raw = row.get("title")
        if raw:
            row = {**row, "original_title": raw, "title": normalize_title(raw)}
        seen[row["id"]] = row
    stmt = (
        pg_insert(Video)
        .values(list(seen.values()))
        .on_conflict_do_nothing(index_elements=[Video.id])
        .returning(Video.id)
    )
    inserted = len(db.execute(stmt).fetchall())
    db.commit()
    return inserted


# --- RSS (free) ---
def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int:
    """Insert new video stubs from an already-fetched feed and stamp the channel's last RSS
    time. DB-only — must run on the session's own thread; the network fetch
    (fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it."""
    inserted = _insert_stubs(db, entries)
    channel.last_rss_at = _now()
    db.add(channel)
    db.commit()
    return inserted


def poll_rss_channel(db: Session, channel: Channel) -> int:
    return apply_rss_feed(db, channel, fetch_channel_feed(channel.id))


# --- Backfill (uploads playlist; costs quota) ---
def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
    content = item.get("contentDetails", {})
    snippet = item.get("snippet", {})
    video_id = content.get("videoId")
    if not video_id:
        return None
    published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt"))
    return {
        "id": video_id,
        "channel_id": channel_id,
        "title": snippet.get("title"),
        "published_at": published,
        "thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
    }


def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> int:
    """First pass: newest videos up to the configured count or age cutoff. Records a
    resume cursor for the later deep backfill."""
    if not channel.uploads_playlist_id:
        channel.recent_synced_at = _now()
        db.add(channel)
        db.commit()
        return 0

    cutoff = _now() - timedelta(days=sysconfig.get_int(db, "backfill_recent_max_days"))
    collected: list[dict] = []
    page_token: str | None = None
    next_cursor: str | None = None
    done = False

    while True:
        page_start = page_token  # token that fetched THIS page (None = first page)
        data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
        stopped_mid_page = False
        for item in data.get("items", []):
            stub = _stub_from_playlist_item(item, channel.id)
            if stub is None:
                continue
            if stub["published_at"] and stub["published_at"] < cutoff:
                stopped_mid_page = True
                break
            collected.append(stub)
            if len(collected) >= sysconfig.get_int(db, "backfill_recent_max_videos"):
                stopped_mid_page = True
                break
        page_token = data.get("nextPageToken")
        if stopped_mid_page:
            # We stopped partway through this page (age cutoff or count cap), so the older
            # items still on it weren't collected. Resume the deep backfill FROM THIS page
            # (re-fetch it; inserts are idempotent) so that tail isn't silently skipped.
            next_cursor = page_start
            done = False
            break
        if not page_token:
            next_cursor = None
            done = True
            break
        next_cursor = page_token  # whole page consumed, more pages remain

    channel.recent_synced_at = _now()
    channel.backfill_cursor = next_cursor
    channel.backfill_done = done
    inserted = _insert_stubs(db, collected)
    db.add(channel)
    db.commit()
    return inserted


def backfill_channel_deep(
    db: Session, yt: YouTubeClient, channel: Channel, max_pages: int = 5
) -> int:
    """Continue paging older uploads from the stored cursor (bounded per run)."""
    if channel.backfill_done or not channel.uploads_playlist_id:
        return 0
    page_token = channel.backfill_cursor
    total = 0
    for _ in range(max_pages):
        data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
        rows = [
            s
            for s in (
                _stub_from_playlist_item(it, channel.id) for it in data.get("items", [])
            )
            if s is not None
        ]
        total += _insert_stubs(db, rows)
        page_token = data.get("nextPageToken")
        channel.backfill_cursor = page_token
        if not page_token:
            channel.backfill_done = True
            break
    db.add(channel)
    db.commit()
    return total


def reconcile_full_history(db: Session) -> int:
    """Mark channels fully backfilled once their stored uploads meet or exceed YouTube's
    advertised video_count. The uploads playlist length is normally <= the advertised
    count, so 'stored >= video_count' means we already hold the whole history — even if
    the deep cursor never reached the end (e.g. a small channel that was never
    deep-requested, so the demand-driven deep job never ran). Without this such a channel
    sits in 'needs full history' forever despite having every video. Idempotent, no API
    quota; returns the number of channels flipped."""
    stored = (
        select(func.count(Video.id))
        .where(Video.channel_id == Channel.id)
        .scalar_subquery()
    )
    result = db.execute(
        update(Channel)
        .where(
            Channel.backfill_done.is_(False),
            Channel.video_count.is_not(None),
            stored >= Channel.video_count,
        )
        .values(backfill_done=True, backfill_cursor=None)
    )
    db.commit()
    return result.rowcount


# --- Enrichment (videos.list) ---
def _live_status(snippet: dict, live_details: dict | None) -> str:
    broadcast = snippet.get("liveBroadcastContent")
    if broadcast == "live":
        return "live"
    if broadcast == "upcoming":
        return "upcoming"
    if live_details and live_details.get("actualEndTime"):
        return "was_live"
    return "none"


def apply_video_details(video: Video, item: dict) -> None:
    snippet = item.get("snippet", {})
    content = item.get("contentDetails", {})
    stats = item.get("statistics", {})
    topics = item.get("topicDetails", {})
    live = item.get("liveStreamingDetails")
    status = item.get("status", {})

    raw_title = snippet.get("title")
    if raw_title:
        # Store the raw title and a normalized one for display (feed/search/downloads).
        video.original_title = raw_title
        video.title = normalize_title(raw_title)
    video.description = snippet.get("description")
    if not video.published_at:
        video.published_at = parse_dt(snippet.get("publishedAt"))
    video.thumbnail_url = best_thumbnail(snippet.get("thumbnails")) or video.thumbnail_url
    video.duration_seconds = parse_iso8601_duration(content.get("duration"))
    video.view_count = int(stats["viewCount"]) if stats.get("viewCount") else None
    video.like_count = int(stats["likeCount"]) if stats.get("likeCount") else None
    video.category_id = int(snippet["categoryId"]) if snippet.get("categoryId") else None
    # Creator keyword tags (free with the snippet we already fetch) → search document.
    tags = snippet.get("tags")
    video.keywords = " ".join(tags) if tags else None
    video.topic_categories = topics.get("topicCategories")
    video.default_language = snippet.get("defaultLanguage") or snippet.get(
        "defaultAudioLanguage"
    )
    video.live_status = _live_status(snippet, live)
    # status part: used by the maintenance/validation job to judge playability.
    if status:
        video.embeddable = status.get("embeddable")
        video.privacy_status = status.get("privacyStatus")
        video.upload_status = status.get("uploadStatus")
    # is_short is decided separately by the youtube.com/shorts probe (run_shorts_classification).


def _apply_video_batch(
    db: Session,
    yt: YouTubeClient,
    videos: list[Video],
    on_missing: Callable[[Video], None],
) -> int:
    """Fetch videos.list details for the given rows, apply them (stamping enriched_at), and
    return how many were applied. Rows YouTube omits (deleted/private) are handled by the
    caller-supplied `on_missing` so each caller decides how to retire them."""
    if not videos:
        return 0
    items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
    now = _now()
    applied = 0
    for video in videos:
        item = items.get(video.id)
        if item is None:
            on_missing(video)
            continue
        apply_video_details(video, item)
        video.enriched_at = now
        applied += 1
    db.commit()
    return applied


def _mark_enriched(video: Video) -> None:
    # Unavailable (deleted/private): stamp so we don't keep retrying.
    video.enriched_at = _now()


def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
    limit = limit or sysconfig.get_int(db, "enrich_batch_size")
    videos = (
        db.execute(select(Video).where(Video.enriched_at.is_(None)).limit(limit))
        .scalars()
        .all()
    )
    return _apply_video_batch(db, yt, list(videos), _mark_enriched)


def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
    """Re-fetch videos in a transient live state so they pick up their final shape.

    `enrich_pending` is one-shot (it only touches enriched_at IS NULL), but a live broadcast
    is not a final state: a stream ends and becomes a normal VOD — gaining a real duration
    and flipping live_status to was_live — and an upcoming premiere eventually goes live then
    ends. So we revisit anything still live/upcoming, plus a just-ended stream that hasn't got
    its duration yet (YouTube can lag a bit after the broadcast), until it settles. Cheap:
    videos.list is 1 unit per 50 ids, and the live/upcoming set is normally tiny."""
    limit = limit or sysconfig.get_int(db, "enrich_batch_size")
    videos = (
        db.execute(
            select(Video)
            .where(
                or_(
                    Video.live_status.in_(LIVE_OR_UPCOMING),
                    and_(
                        Video.live_status == "was_live",
                        Video.duration_seconds.is_(None),
                    ),
                )
            )
            .limit(limit)
        )
        .scalars()
        .all()
    )

    def _retire(video: Video) -> None:
        # The broadcast vanished (deleted/private). Stop treating it as live so it
        # doesn't get refreshed forever; it just becomes an ordinary (dead) row.
        video.live_status = "none"

    return _apply_video_batch(db, yt, list(videos), _retire)


def run_shorts_classification(db: Session, limit: int | None = None) -> dict:
    """Finalize Shorts classification: cheaply mark non-candidates, then probe the
    youtube.com/shorts URL for short-enough, non-live, enriched videos."""
    limit = limit or sysconfig.get_int(db, "shorts_probe_batch")
    probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds")

    # 1) Anything enriched that can't be a Short -> finalize without a probe.
    db.execute(
        update(Video)
        .where(
            Video.shorts_probed.is_(False),
            Video.enriched_at.is_not(None),
            or_(
                Video.duration_seconds.is_(None),
                Video.duration_seconds > probe_max,
                Video.live_status != "none",
            ),
        )
        .values(is_short=False, shorts_probed=True)
    )
    db.commit()

    # 2) Probe the remaining candidates.
    candidates = (
        db.execute(
            select(Video).where(
                Video.shorts_probed.is_(False),
                Video.enriched_at.is_not(None),
                Video.duration_seconds <= probe_max,
                Video.live_status == "none",
            ).limit(limit)
        )
        .scalars()
        .all()
    )
    if not candidates:
        return {"probed": 0, "shorts": 0}

    probed = 0
    shorts = 0
    with make_client() as client:
        def work(video: Video):
            return video, probe_is_short(client, video.id)

        with ThreadPoolExecutor(max_workers=16) as pool:
            for video, result in pool.map(work, candidates):
                if result is None:
                    continue  # leave unprobed; retry on a later run
                video.is_short = result
                video.shorts_probed = True
                probed += 1
                if result:
                    shorts += 1
                progress.report(probed, len(candidates), "probe")
    db.commit()
    return {"probed": probed, "shorts": shorts}