yt-channel-scraper/tests/test_store_platform.py

from __future__ import annotations

import json
from pathlib import Path

import pytest

from yt_scraper.store import Store, VideoRef
from yt_scraper.parse import Segment


@pytest.fixture()
def store(tmp_path):
    return Store(tmp_path / "test.db")


@pytest.fixture()
def seeded_store(store):
    store.upsert_channel("UC1", "@alpha", "Alpha", 2)
    store.upsert_channel("UC2", "@beta", "Beta", 1)
    refs = [
        VideoRef("v1", "UC1", "Hello world vaporwave", "https://y/watch?v=v1", "20240101", 120),
        VideoRef("v2", "UC1", "Dragon ball nostalgia", "https://y/watch?v=v2", "20240201", 200),
        VideoRef("v3", "UC2", "Vaporwave aesthetics", "https://y/watch?v=v3", "20240301", 300),
    ]
    store.upsert_videos(refs)
    return store


def test_migration_idempotent(tmp_path):
    db = tmp_path / "a.db"
    s1 = Store(db)
    s2 = Store(db)  # second instantiation should not error
    # new columns exist
    cols = {r["name"] for r in s1._connect().execute("PRAGMA table_info(videos)")}
    for c in ("view_count", "tags", "segments_json", "chapters_json", "thumbnail"):
        assert c in cols
    # new tables exist
    names = {r["name"] for r in s1._connect().execute("SELECT name FROM sqlite_master WHERE type='table'")}
    for t in ("transcript_segments", "cookies_meta", "scrape_jobs"):
        assert t in names
    # fts table
    fts = {r["name"] for r in s1._connect().execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE '%fts%'")}
    assert "transcript_fts" in fts


def test_store_and_search_segments(seeded_store):
    store = seeded_store
    segs = [
        Segment(0.0, 2.0, "vaporwave is an aesthetic"),
        Segment(3.0, 5.0, "dragon ball is a classic anime"),
    ]
    store.store_segments("v1", segs)
    store.store_segments("v3", [Segment(0.0, 2.0, "retro vaporwave vibes")])

    hits = store.search_segments("vaporwave")
    assert len(hits) == 2
    vids = {h.video_id for h in hits}
    assert vids == {"v1", "v3"}

    # channel-scoped search
    hits_c = store.search_segments("vaporwave", channel_id="UC2")
    assert len(hits_c) == 1
    assert hits_c[0].video_id == "v3"

    # get_segments ordered
    seg_rows = store.get_segments("v1")
    assert len(seg_rows) == 2
    assert seg_rows[0].idx == 0


def test_store_segments_overwrites(seeded_store):
    store = seeded_store
    store.store_segments("v1", [Segment(0.0, 1.0, "old")])
    store.store_segments("v1", [Segment(0.0, 1.0, "new"), Segment(2.0, 3.0, "second")])
    rows = store.get_segments("v1")
    assert len(rows) == 2
    assert rows[0].text == "new"


def test_update_video_metadata(seeded_store):
    store = seeded_store
    store.update_video_metadata("v1", view_count=1000, like_count=50, tags=["music", "retro"], thumbnail="http://t.png")
    v = store.get_video("v1")
    assert v.view_count == 1000
    assert v.like_count == 50
    assert json.loads(v.tags) == ["music", "retro"]
    assert v.thumbnail == "http://t.png"


def test_query_videos_filters(seeded_store):
    store = seeded_store
    rows, total = store.query_videos(channel_id="UC1")
    assert total == 2
    rows2, total2 = store.query_videos(min_duration=250)
    assert total2 == 1
    assert rows2[0].video_id == "v3"
    rows3, total3 = store.query_videos(q="dragon")
    assert total3 == 1
    assert rows3[0].video_id == "v2"


def test_cookie_vault(store):
    from yt_scraper import cookies
    sample = (
        "# Netscape HTTP Cookie File\n"
        ".youtube.com\tTRUE\t/\tTRUE\t1800287621\tSID\tabc\n"
        ".youtube.com\tTRUE\t/\tTRUE\t1800287621\tSAPISID\tdef\n"
        ".youtube.com\tTRUE\t/\tFALSE\t1800287621\tPREF\tf1=2\n"
    )
    cid = cookies.import_text(store, sample, label="test", cookie_dir=str(Path(store.db_path).parent / "ck"))
    vault = cookies.list_vault(store)
    assert len(vault) == 1
    assert vault[0].has_session is True
    assert vault[0].cookie_count == 3
    # set active + resolve
    cookies.set_active(store, cid)
    path = cookies.resolve_active_path(store, cookie_dir=str(Path(store.db_path).parent / "ck"))
    assert path and Path(path).exists()
    # delete
    cookies.delete(store, cid, cookie_dir=str(Path(store.db_path).parent / "ck"))
    assert cookies.list_vault(store) == []


def test_cookie_invalid_rejected(store):
    from yt_scraper import cookies
    with pytest.raises(ValueError):
        cookies.import_text(store, "not a cookie file", label="bad", cookie_dir=str(Path(store.db_path).parent / "ck"))


def test_cookie_only_one_active(store):
    from yt_scraper import cookies
    cdir = str(Path(store.db_path).parent / "ck2")
    sample = "# Netscape\n.youtube.com\tTRUE\t/\tTRUE\t1800287621\tSID\tx\n"
    c1 = cookies.import_text(store, sample, label="a", cookie_dir=cdir)
    c2 = cookies.import_text(store, sample, label="b", cookie_dir=cdir)
    cookies.set_active(store, c1)
    cookies.set_active(store, c2)
    actives = [c for c in cookies.list_vault(store) if c.is_active]
    assert len(actives) == 1
    assert actives[0].id == c2


def test_jobs_crud(store):
    store.create_job("job1", "UC1", {"limit": 5})
    store.update_job("job1", status="running", total=5)
    store.update_job("job1", completed=3)
    j = store.get_job("job1")
    assert j.status == "running"
    assert j.total == 5
    assert j.completed == 3
    store.update_job("job1", status="done", completed=5, finished=True)
    j2 = store.get_job("job1")
    assert j2.status == "done"
    assert j2.finished_at is not None


def test_dashboard_aggregates(seeded_store):
    store = seeded_store
    store.store_segments("v1", [Segment(0.0, 2.0, "vaporwave hello")])
    store.update_video_metadata("v1", view_count=500, tags=["music", "vaporwave"])
    store.update_video_metadata("v2", tags=["music", "anime"])
    d = store.dashboard()
    assert len(d["channels"]) == 2
    ch1 = [c for c in d["channels"] if c["channel_id"] == "UC1"][0]
    assert ch1["video_count_db"] == 2
    assert ch1["total_views"] == 500
    tags = {t["tag"]: t["count"] for t in d["top_tags"]}
    assert tags.get("music") == 2