t-youtube/.hermes/plans/2026-06-21_youtube-api-v3-oauth.md

t-youtube: YouTube Data API v3 + OAuth Implementation Plan

For Hermes: Use subagent-driven-development skill to implement this plan task-by-task.

Goal: Replace the broken yt-dlp/cookies subscription fetch with YouTube Data API v3 OAuth, restoring live sync.

Architecture: Add Google OAuth 2.0 credential management. Replace fetcher.py's yt-dlp call with google-api-python-client calling the /subscriptions endpoint. Keep DB schema, summarizer, and frontend unchanged.

Tech Stack: Python 3.13, FastAPI, aiosqlite, google-api-python-client, google-auth-oauthlib

---

Prerequisites (manual, before implementation)

1. Go to Google Cloud Console → create a project → enable YouTube Data API v3
2. Create OAuth 2.0 Client ID (Desktop app type)
3. Download client_secret.json → place in config/client_secret.json
4. Run OAuth flow once to generate config/token.json

---

Task 1: Add Google API dependencies

Objective: Install required Python packages

Files:

• Modify: backend/requirements.txt

Step 1: Add to requirements.txt

google-api-python-client>=2.140.0
google-auth-oauthlib>=1.2.0
google-auth-httplib2>=0.2.0

Step 2: Install

cd backend && .venv/bin/pip install -r requirements.txt

Step 3: Verify

.venv/bin/python3 -c "from googleapiclient.discovery import build; print('OK')"

Expected: OK

---

Task 2: Create OAuth credential manager

Objective: Module to load/store OAuth tokens, handle refresh

Files:

• Create: backend/auth.py

Complete code:

import os
import pickle
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow

SCOPES = ["https://www.googleapis.com/auth/youtube.readonly"]
CONFIG_DIR = os.environ.get("CONFIG_DIR", "/app/config")
CLIENT_SECRET = os.path.join(CONFIG_DIR, "client_secret.json")
TOKEN_FILE = os.path.join(CONFIG_DIR, "token.pickle")

def get_credentials():
    """Load or refresh OAuth credentials, triggering browser flow if needed."""
    creds = None
    if os.path.exists(TOKEN_FILE):
        with open(TOKEN_FILE, "rb") as token:
            creds = pickle.load(token)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            if not os.path.exists(CLIENT_SECRET):
                raise FileNotFoundError(
                    f"client_secret.json not found at {CLIENT_SECRET}. "
                    "Download from Google Cloud Console."
                )
            flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET, SCOPES)
            creds = flow.run_local_server(port=0)

        os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
        with open(TOKEN_FILE, "wb") as token:
            pickle.dump(creds, token)

    return creds

Step 1: Verify module loads

cd backend && .venv/bin/python3 -c "import auth; print('auth module OK')"

Expected: auth module OK

Step 2: Commit

git add backend/auth.py backend/requirements.txt
git commit -m "feat: add OAuth credential manager for YouTube API"

---

Task 3: Test OAuth flow (dry run)

Objective: Verify token.pickle is generated from client_secret.json

Files:

• Create: backend/tests/test_auth.py

Step 1: Write test

import os
import tempfile
from unittest.mock import patch
import auth

def test_credentials_flow_with_valid_secret():
    """When client_secret.json exists, get_credentials() should not raise."""
    # This test only checks the file-exists path; real OAuth requires browser
    if os.path.exists(auth.CLIENT_SECRET):
        # Just verify module doesn't crash on import
        assert auth.SCOPES == ["https://www.googleapis.com/auth/youtube.readonly"]
    else:
        # Without client_secret, should raise FileNotFoundError
        try:
            auth.get_credentials()
            assert False, "Should have raised"
        except FileNotFoundError:
            pass

Step 2: Run test

cd backend && .venv/bin/python3 -m pytest tests/test_auth.py -v

Expected: PASS (skips OAuth if no secret, or passes structure check)

Step 3: Commit

git add backend/tests/test_auth.py
git commit -m "test: add OAuth auth module test"

---

Task 4: Rewrite fetcher.py to use YouTube Data API v3

Objective: Replace yt-dlp with googleapiclient subscription fetch

Files:

• Modify: backend/fetcher.py

Complete replacement code:

import asyncio
import logging
import aiosqlite
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from auth import get_credentials
from db import DB_PATH

log = logging.getLogger(__name__)

async def fetch_subscriptions():
    """Fetch latest videos from subscribed channels using YouTube Data API v3."""
    try:
        creds = get_credentials()
    except FileNotFoundError as e:
        log.error(f"OAuth setup incomplete: {e}")
        return 0
    except Exception as e:
        log.error(f"OAuth credential error: {e}")
        return 0

    youtube = build("youtube", "v3", credentials=creds)

    # Step 1: Get subscribed channel IDs
    channel_ids = []
    try:
        request = youtube.subscriptions().list(
            part="snippet",
            mine=True,
            maxResults=50,
            order="alphabetical"
        )
        while request:
            response = request.execute()
            for item in response.get("items", []):
                cid = item["snippet"]["resourceId"]["channelId"]
                if cid not in channel_ids:
                    channel_ids.append(cid)
            request = youtube.subscriptions().list_next(request, response)
    except HttpError as e:
        log.error(f"YouTube API error fetching subscriptions: {e}")
        return 0

    log.info(f"Found {len(channel_ids)} subscribed channels")

    if not channel_ids:
        return 0

    # Step 2: For each channel, get latest videos
    new_count = 0
    async with aiosqlite.connect(DB_PATH) as db:
        for cid in channel_ids[:20]:  # Respect quota, limit to 20 channels
            try:
                req = youtube.search().list(
                    part="snippet",
                    channelId=cid,
                    maxResults=10,
                    order="date",
                    type="video"
                )
                resp = req.execute()
                for item in resp.get("items", []):
                    vid = item["id"]["videoId"]
                    snippet = item["snippet"]

                    cursor = await db.execute(
                        "SELECT 1 FROM videos WHERE video_id = ?", (vid,)
                    )
                    if await cursor.fetchone():
                        continue

                    await db.execute("""
                        INSERT INTO videos
                        (video_id, title, channel_name, channel_id, url, thumbnail_url, publish_date)
                        VALUES (?, ?, ?, ?, ?, ?, ?)
                    """, (
                        vid,
                        snippet.get("title", "Untitled"),
                        snippet.get("channelTitle", "Unknown"),
                        snippet.get("channelId", cid),
                        f"https://www.youtube.com/watch?v={vid}",
                        snippet.get("thumbnails", {}).get("high", {}).get("url", ""),
                        snippet.get("publishTime", "")[:10]
                    ))
                    new_count += 1
                    log.debug(f"New video: {snippet.get('title', 'Untitled')[:50]}")
            except HttpError as e:
                log.warning(f"Channel {cid} fetch error: {e}")
                continue

        await db.commit()

    log.info(f"Ingestion completed. Found {new_count} new videos.")
    return new_count


if __name__ == "__main__":
    import sys
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

    async def main():
        from db import init_db
        await init_db()
        await fetch_subscriptions()

    asyncio.run(main())

Step 1: Run existing tests to ensure nothing else broke

cd backend && .venv/bin/python3 -m pytest tests/test_fetcher.py -v

Expected: Mock tests should still pass (fetcher import may need auth module available)

Step 2: Update test_fetcher.py to mock googleapiclient

Modify backend/tests/test_fetcher.py:

# Replace yt-dlp mocks with googleapiclient mocks

Step 3: Commit

git add backend/fetcher.py backend/tests/test_fetcher.py
git commit -m "feat: replace yt-dlp with YouTube Data API v3 OAuth"

---

Task 5: Update Docker config for new dependencies

Objective: Ensure Docker build includes google-auth and config mounts

Files:

• Modify: Dockerfile
• Modify: docker-compose.yml

Dockerfile changes:

# In the Python stage, ensure google packages are installed
# requirements.txt already includes them from Task 1

docker-compose.yml changes:

# Mount config directory for OAuth secrets
volumes:
  - ./config:/app/config

Step 1: Verify Docker build

docker compose build

Expected: Build succeeds

Step 2: Commit

git add Dockerfile docker-compose.yml
git commit -m "chore: update Docker config for OAuth and API v3"

---

Task 6: Integration test with real API

Objective: Run a live fetch against real YouTube subscription data

Files:

• Create: backend/tests/test_live_fetch.py

Step 1: Write integration test

"""Live integration test — requires valid OAuth token in config/"""
import pytest
import asyncio
from fetcher import fetch_subscriptions
from db import init_db

@pytest.mark.asyncio
@pytest.mark.integration
async def test_live_subscription_fetch():
    """Fetch real subscriptions. Skip if no OAuth token."""
    await init_db()
    count = await fetch_subscriptions()
    # With valid OAuth and subscriptions, should find videos
    # Even 0 is acceptable (no new videos since last fetch)
    assert count >= 0, "Fetch should not crash"
    print(f"Fetched {count} new videos")

Step 2: Run (requires OAuth setup)

cd backend && .venv/bin/python3 -m pytest tests/test_live_fetch.py -v -s

Expected: PASS or SKIP if no OAuth

Step 3: Commit

git add backend/tests/test_live_fetch.py
git commit -m "test: add live YouTube API integration test"

---

Task 7: Final verification — full pipeline

Objective: Run full test suite and verify everything works end-to-end

Step 1: Run all tests

cd ~/repos/t-youtube
./run_tests.sh

Step 2: Verify frontend still builds

cd frontend && npm run build 2>/dev/null || echo "Frontend build skipped"

Step 3: Commit final state

git add -A && git commit -m "feat: complete YouTube Data API v3 migration"

---

Verification Checklist

• config/client_secret.json exists
• config/token.pickle generated after first OAuth flow
• fetch_subscriptions() returns video count > 0
• Videos appear in DB with correct metadata
• Summarizer still works with newly fetched videos
• All mock tests pass
• Docker build succeeds
• Frontend displays new videos

Risks

1. API quota: YouTube Data API v3 has 10,000 units/day quota. Each subscription list call costs 1 unit, each search costs 100 units. With 20 channels × 10 results, that's ~2,000 units per sync.
2. OAuth browser flow: First run requires browser for Google login. On headless NAS, run locally first and copy token.pickle.
3. Token expiry: refresh_token handles this automatically via google-auth.