youtube-analytics/main.py

"""CLI entry point for PBS YouTube Analytics."""

import click
from pathlib import Path

from src.database import init_db, get_connection, DEFAULT_DB_PATH
from src.seed import seed_database


@click.group()
def cli():
    """PBS YouTube Analytics CLI."""
    pass


@cli.command()
def init_db_cmd():
    """Initialize the database schema."""
    try:
        init_db(DEFAULT_DB_PATH)
        click.echo(click.style("✓ Database initialized successfully", fg="green"))
    except Exception as e:
        click.echo(click.style(f"✗ Error initializing database: {e}", fg="red"), err=True)
        raise SystemExit(1)


@cli.command()
def seed():
    """Populate the database with sample data."""
    try:
        seed_database(DEFAULT_DB_PATH)
        click.echo(click.style("✓ Database seeded successfully", fg="green"))
    except Exception as e:
        click.echo(click.style(f"✗ Error seeding database: {e}", fg="red"), err=True)
        raise SystemExit(1)


@cli.command("clear-seed")
def clear_seed():
    """Remove sample seed data (videos with IDs matching short_* or long_*)."""
    try:
        conn = get_connection(DEFAULT_DB_PATH)

        # Seed video IDs are short_NNN / long_NNN — real YouTube IDs are 11-char alphanumeric,
        # so these patterns can never collide with collected data.
        seed_pattern = "video_id LIKE 'short_%' OR video_id LIKE 'long_%'"

        totals = {}
        for table in ("video_traffic_sources", "video_retention", "video_daily_metrics"):
            cur = conn.execute(f"DELETE FROM {table} WHERE {seed_pattern}")
            totals[table] = cur.rowcount

        cur = conn.execute(f"DELETE FROM videos WHERE {seed_pattern}")
        totals["videos"] = cur.rowcount

        conn.commit()
        conn.close()

        for table, count in totals.items():
            click.echo(f"  {table}: {count} rows deleted")
        click.echo(click.style("✓ Seed data cleared", fg="green"))
        click.echo("  Note: channel_daily_metrics rows have no seed identifier and are left untouched.")
    except Exception as e:
        click.echo(click.style(f"✗ Error clearing seed data: {e}", fg="red"), err=True)
        raise SystemExit(1)


@cli.command()
@click.option("--start-date", default=None, help="Start date (YYYY-MM-DD), default: 30 days ago")
@click.option("--end-date", default=None, help="End date (YYYY-MM-DD), default: 2 days ago")
def collect(start_date, end_date):
    """Collect real data from YouTube API."""
    import sqlite3

    try:
        from src.auth import get_authenticated_service
        from src.collector import (
            collect_channel_metrics,
            collect_daily_metrics,
            collect_retention,
            collect_traffic_sources,
            collect_video_list,
        )

        click.echo("Attempting to authenticate with YouTube API...")
        youtube, youtube_analytics = get_authenticated_service()
        click.echo(click.style("✓ Successfully authenticated with YouTube API", fg="green"))

        # Ensure database exists
        init_db(DEFAULT_DB_PATH)
        conn = sqlite3.connect(DEFAULT_DB_PATH)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA foreign_keys=ON")

        # Step 1: Collect video list
        click.echo("\n📹 Collecting video list...")
        video_ids = collect_video_list(youtube, conn)

        # Step 2: Collect daily metrics per video
        click.echo("📊 Collecting daily metrics...")
        collect_daily_metrics(youtube_analytics, conn, video_ids, start_date, end_date)

        # Step 3: Collect retention curves
        click.echo("📈 Collecting retention data...")
        collect_retention(youtube_analytics, conn, video_ids)

        # Step 4: Collect traffic sources
        click.echo("🔗 Collecting traffic sources...")
        collect_traffic_sources(youtube_analytics, conn, video_ids, start_date, end_date)

        # Step 5: Collect channel-level metrics
        click.echo("📺 Collecting channel metrics...")
        collect_channel_metrics(youtube_analytics, conn, start_date, end_date)

        conn.close()
        click.echo(click.style("\n✓ All data collected successfully!", fg="green"))

    except FileNotFoundError as e:
        click.echo(click.style(f"✗ {e}", fg="red"), err=True)
        click.echo("\nTo use the collect command:")
        click.echo("1. Create credentials at: https://console.cloud.google.com/apis/credentials")
        click.echo("2. Download OAuth credentials as client_secret.json")
        click.echo("3. Save to: config/client_secret.json")
        raise SystemExit(1)
    except Exception as e:
        click.echo(click.style(f"✗ Error during collection: {e}", fg="red"), err=True)
        import traceback
        traceback.print_exc()
        raise SystemExit(1)


@cli.command()
def dashboard():
    """Launch the Streamlit dashboard."""
    try:
        import streamlit  # noqa: F401 — just checking it's installed
    except ImportError:
        click.echo(
            click.style("✗ Streamlit not installed. Install with: uv add streamlit", fg="red"),
            err=True,
        )
        raise SystemExit(1)

    import subprocess
    import sys

    dashboard_path = str(Path(__file__).parent / "src" / "dashboard.py")
    try:
        subprocess.run(
            [sys.executable, "-m", "streamlit", "run", dashboard_path],
            check=True,
        )
    except Exception as e:
        click.echo(click.style(f"✗ Error launching dashboard: {e}", fg="red"), err=True)
        raise SystemExit(1)


if __name__ == "__main__":
    cli()