t-youtube/backend/main.py

import os
import json
import logging
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Query, Response
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from typing import Optional, List
import aiosqlite

from db import init_db, get_db
from fetcher import fetch_subscriptions
from summarizer import summarize_video, summarize_all_pending
from agent_worker import start_worker

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger(__name__)

app = FastAPI(title="T YouTube", description="Text-based YouTube Summarizer")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],

)

# API token validation for write operations
API_TOKEN=os.environ.get('API_TOKEN', '')

async def validate_api_token(request: Request, call_next):
    path = request.url.path
    method = request.method
    # Only check auth for write operations on /api/*
    if path.startswith("/api/") and method in ("POST", "PATCH", "DELETE") and API_TOKEN:
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer ") or auth_header[7:] != API_TOKEN:
            return JSONResponse(status_code=401, content={"detail": "Unauthorized"})
    return await call_next(request)

app.middleware("http")(validate_api_token)


class StatusUpdate(BaseModel):
    status: str

class NoteUpdate(BaseModel):
    note_text: str
    queue_to_agent: bool = False

class NoteMigrate(BaseModel):
    notes: dict

class AgentQueueItem(BaseModel):
    video_id: str
    user_prompt: str

class AgentQueueUpdate(BaseModel):
    status: str

@app.on_event("startup")
async def startup():
    await init_db()
    if os.environ.get("AGENT_ENABLED", "").lower() == "true":
        await start_worker()
        log.info("Agent queue worker enabled")
    else:
        log.info("Agent queue worker disabled (set AGENT_ENABLED=true to enable)")

@app.get("/api/health")
async def health():
    return {"status": "ok"}

@app.get("/api/config")
async def api_config():
    """Return API config without exposing the token."""
    return {"auth_enabled": bool(API_TOKEN)}

@app.get("/auto_categories.json")
async def auto_categories():
    path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend", "src", "auto_categories.json")
    if os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    return {}

@app.get("/api/channels")
async def get_channels(db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT DISTINCT channel_name, channel_id FROM videos ORDER BY channel_name")
    rows = await cursor.fetchall()
    return [dict(row) for row in rows]

@app.get("/")
async def root():
    from fastapi.responses import FileResponse
    return FileResponse(os.path.join(frontend_build_path, "index.html"))

@app.get("/api/videos")
async def get_videos(
    status: Optional[str] = "pending",
    limit: int = 50,
    channel: Optional[str] = None,
    search: Optional[str] = None,
    db: aiosqlite.Connection = Depends(get_db)
):
    """Get videos with optional search filter."""
    if search and len(search) >= 2:
        # Use search endpoint logic
        q = search.replace('"', '\\"')
        base_query = "SELECT v.* FROM videos_fts f JOIN videos v ON f.rowid = v.rowid WHERE videos_fts MATCH ?"
        params = [q]
        if status:
            base_query += " AND v.status = ?"
            params.append(status)
        if channel:
            base_query += " AND v.channel_name = ?"
            params.append(channel)
        base_query += " ORDER BY rank LIMIT ?"
        params.append(limit)
    else:
        if channel:
            query = "SELECT * FROM videos WHERE status = ? AND channel_name = ? ORDER BY publish_date DESC, created_at DESC LIMIT ?"
            cursor = await db.execute(query, (status, channel, limit))
        else:
            query = "SELECT * FROM videos WHERE status = ? ORDER BY publish_date DESC, created_at DESC LIMIT ?"
            cursor = await db.execute(query, (status, limit))
        rows = await cursor.fetchall()
        return [dict(row) for row in rows]
    
    try:
        cursor = await db.execute(base_query, params)
    except Exception:
        # Fallback to LIKE search
        like_query = "SELECT * FROM videos WHERE title LIKE ?"
        params = [f"%{search}%"]
        if status:
            like_query += " AND status = ?"
            params.append(status)
        if channel:
            like_query += " AND channel_name = ?"
            params.append(channel)
        like_query += " ORDER BY publish_date DESC LIMIT ?"
        params.append(limit)
        cursor = await db.execute(like_query, params)
    
    rows = await cursor.fetchall()
    return [dict(row) for row in rows]

async def bg_fetch_and_summarize():
    log.info("Starting background subscription fetch and summarization loop...")
    try:
        new_count = await fetch_subscriptions()
        if new_count > 0:
            await summarize_all_pending()
    except Exception as e:
        log.error(f"Error in background fetch and summarize task: {e}")

@app.post("/api/videos/fetch")
async def trigger_fetch(background_tasks: BackgroundTasks):
    background_tasks.add_task(bg_fetch_and_summarize)
    return {"message": "Subscription sync and summarization running in background."}

@app.post("/api/videos/{video_id}/summarize")
async def trigger_summarize(video_id: str, background_tasks: BackgroundTasks, db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT 1 FROM videos WHERE video_id = ?", (video_id,))
    exists = await cursor.fetchone()
    if not exists:
        raise HTTPException(status_code=404, detail="Video not found")
    background_tasks.add_task(summarize_video, video_id)
    return {"message": f"Summarization for video {video_id} queued."}

@app.post("/api/videos/{video_id}/status")
async def update_status(
    video_id: str,
    update: StatusUpdate,
    db: aiosqlite.Connection = Depends(get_db)
):
    if update.status not in ["pending", "read", "skipped"]:
        raise HTTPException(status_code=400, detail="Invalid status value")
    cursor = await db.execute("SELECT 1 FROM videos WHERE video_id = ?", (video_id,))
    exists = await cursor.fetchone()
    if not exists:
        raise HTTPException(status_code=404, detail="Video not found")
    await db.execute("UPDATE videos SET status = ? WHERE video_id = ?", (update.status, video_id))
    await db.commit()
    return {"message": f"Video {video_id} status updated to {update.status}."}

# ---- Notes (server-side, replaces localStorage) ----



@app.get("/api/notes")
async def list_notes(db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT video_id, note_text, updated_at FROM video_notes ORDER BY updated_at DESC")
    rows = await cursor.fetchall()
    return [dict(row) for row in rows]

@app.get("/api/notes/{video_id}")
async def get_note(video_id: str, db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT note_text, updated_at FROM video_notes WHERE video_id = ?", (video_id,))
    row = await cursor.fetchone()
    if not row:
        return {"note_text": "", "updated_at": None}
    return dict(row)

@app.post("/api/notes/{video_id}")
async def save_note(video_id: str, update: NoteUpdate, db: aiosqlite.Connection = Depends(get_db)):
    await db.execute("""
        INSERT INTO video_notes (video_id, note_text, updated_at)
        VALUES (?, ?, CURRENT_TIMESTAMP)
        ON CONFLICT(video_id) DO UPDATE SET
            note_text = excluded.note_text,
            updated_at = CURRENT_TIMESTAMP
    """, (video_id, update.note_text))
    await db.commit()

    if update.queue_to_agent and update.note_text.strip():
        await db.execute("""
            INSERT INTO agent_queue (video_id, user_prompt, status)
            VALUES (?, ?, 'pending')
        """, (video_id, update.note_text.strip()))
        await db.commit()
        log.info(f"Queued video {video_id} for agent: {update.note_text[:80]}")

    return {"message": "Note saved"}

@app.post("/api/notes/migrate")
async def migrate_notes(migration: NoteMigrate, db: aiosqlite.Connection = Depends(get_db)):
    count = 0
    for video_id, note_text in migration.notes.items():
        if not note_text.strip():
            continue
        await db.execute("""
            INSERT INTO video_notes (video_id, note_text, updated_at)
            VALUES (?, ?, CURRENT_TIMESTAMP)
            ON CONFLICT(video_id) DO UPDATE SET
                note_text = excluded.note_text,
                updated_at = CURRENT_TIMESTAMP
        """, (video_id, note_text))
        count += 1
    await db.commit()
    return {"message": f"{count} notes migrated"}

# ---- Agent Queue ----

@app.get("/api/agent-queue")
async def list_agent_queue(
    status: Optional[str] = None,
    limit: int = 50,
    db: aiosqlite.Connection = Depends(get_db)
):
    if status:
        query = """
            SELECT q.id, q.video_id, q.user_prompt, q.status, q.created_at, q.processed_at,
                   v.title, v.channel_name, v.url, v.summary
            FROM agent_queue q
            LEFT JOIN videos v ON q.video_id = v.video_id
            WHERE q.status = ?
            ORDER BY q.created_at DESC LIMIT ?
        """
        cursor = await db.execute(query, (status, limit))
    else:
        query = """
            SELECT q.id, q.video_id, q.user_prompt, q.status, q.created_at, q.processed_at,
                   v.title, v.channel_name, v.url, v.summary
            FROM agent_queue q
            LEFT JOIN videos v ON q.video_id = v.video_id
            ORDER BY q.created_at DESC LIMIT ?
        """
        cursor = await db.execute(query, (limit,))
    rows = await cursor.fetchall()
    return [dict(row) for row in rows]

@app.post("/api/agent-queue")
async def add_to_agent_queue(item: AgentQueueItem, db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT 1 FROM videos WHERE video_id = ?", (item.video_id,))
    exists = await cursor.fetchone()
    if not exists:
        raise HTTPException(status_code=404, detail="Video not found")
    await db.execute("""
        INSERT INTO agent_queue (video_id, user_prompt, status)
        VALUES (?, ?, 'pending')
    """, (item.video_id, item.user_prompt))
    await db.commit()
    return {"message": "Item queued for agent"}

@app.patch("/api/agent-queue/{item_id}")
async def update_agent_queue(item_id: int, update: AgentQueueUpdate, db: aiosqlite.Connection = Depends(get_db)):
    valid_statuses = ["pending", "processing", "done", "failed"]
    if update.status not in valid_statuses:
        raise HTTPException(status_code=400, detail=f"Invalid status. Must be one of: {valid_statuses}")
    cursor = await db.execute("SELECT 1 FROM agent_queue WHERE id = ?", (item_id,))
    exists = await cursor.fetchone()
    if not exists:
        raise HTTPException(status_code=404, detail="Queue item not found")
    if update.status in ("done", "failed"):
        await db.execute("""
            UPDATE agent_queue SET status = ?, processed_at = CURRENT_TIMESTAMP WHERE id = ?
        """, (update.status, item_id))
    else:
        await db.execute("UPDATE agent_queue SET status = ? WHERE id = ?", (update.status, item_id))
    await db.commit()
    return {"message": f"Queue item {item_id} updated to {update.status}"}

@app.delete("/api/agent-queue/{item_id}")
async def remove_from_agent_queue(item_id: int, db: aiosqlite.Connection = Depends(get_db)):
    cursor = await db.execute("SELECT 1 FROM agent_queue WHERE id = ?", (item_id,))
    exists = await cursor.fetchone()
    if not exists:
        raise HTTPException(status_code=404, detail="Queue item not found")
    await db.execute("DELETE FROM agent_queue WHERE id = ?", (item_id,))
    await db.commit()
    return {"message": f"Queue item {item_id} removed"}

# ---- Search (FTS5) ----

@app.get("/api/videos/search")
async def search_videos(
    q: str = "",
    status: Optional[str] = None,
    limit: int = 50,
    channel: Optional[str] = None,
    db: aiosqlite.Connection = Depends(get_db)
):
    """Full-text search across video titles and summaries using FTS5."""
    if not q or len(q) < 2:
        return []
    
    # Try FTS5 first
    fts_query = q.replace('"', '\"')
    base_query = "SELECT v.* FROM videos_fts f JOIN videos v ON f.rowid = v.rowid WHERE videos_fts MATCH ?"
    params = [fts_query]
    
    if status:
        base_query += " AND v.status = ?"
        params.append(status)
    if channel:
        base_query += " AND v.channel_name = ?"
        params.append(channel)
    base_query += " ORDER BY rank LIMIT ?"
    params.append(limit)
    
    try:
        cursor = await db.execute(base_query, params)
    except Exception:
        # Fallback to LIKE if FTS5 not available
        like_query = "SELECT * FROM videos WHERE title LIKE ?"
        params = [f"%{q}%"]
        if status:
            like_query += " AND status = ?"
            params.append(status)
        if channel:
            like_query += " AND channel_name = ?"
            params.append(channel)
        like_query += " LIMIT ?"
        params.append(limit)
        cursor = await db.execute(like_query, params)
    
    rows = await cursor.fetchall()
    return [dict(row) for row in rows]

# ---- Batch Operations ----

class BatchUpdate(BaseModel):
    video_ids: list[str]
    status: str

@app.post("/api/videos/batch")
async def batch_update(
    batch: BatchUpdate,
    db: aiosqlite.Connection = Depends(get_db)
):
    """Batch update video statuses."""
    if not batch.video_ids:
        return {"message": "No video IDs provided"}
    if batch.status not in ["pending", "read", "skipped"]:
        raise HTTPException(status_code=400, detail="Invalid status")
    
    placeholders = ",".join(["?" for _ in batch.video_ids])
    await db.execute(f"UPDATE videos SET status = ? WHERE video_id IN ({placeholders})", 
                     [batch.status] + batch.video_ids)
    await db.commit()
    
    return {"message": f"Updated {len(batch.video_ids)} videos"}

# ---- Stats Dashboard ----

@app.get("/api/stats")
async def get_stats(db: aiosqlite.Connection = Depends(get_db)):
    """Return dashboard statistics."""
    stats = {}
    
    # Total videos
    cursor = await db.execute("SELECT COUNT(*) FROM videos")
    stats["total"] = (await cursor.fetchone())[0]
    
    # Status breakdown
    cursor = await db.execute("SELECT status, COUNT(*) FROM videos GROUP BY status")
    stats["status_breakdown"] = {row[0]: row[1] for row in await cursor.fetchall()}
    
    # Channels
    cursor = await db.execute("SELECT COUNT(DISTINCT channel_name) FROM videos")
    stats["channels"] = (await cursor.fetchone())[0]
    
    # Videos with summaries vs without
    cursor = await db.execute("SELECT COUNT(*) FROM videos WHERE summary IS NOT NULL AND summary != ''")
    stats["with_summary"] = (await cursor.fetchone())[0]
    
    # Recent activity (last 7 days)
    cursor = await db.execute("SELECT COUNT(*) FROM videos WHERE created_at >= datetime('now', '-7 days')")
    stats["last_7_days"] = (await cursor.fetchone())[0]
    
    # Top channels by video count
    cursor = await db.execute("SELECT channel_name, COUNT(*) as cnt FROM videos GROUP BY channel_name ORDER BY cnt DESC LIMIT 10")
    stats["top_channels"] = [{"name": row[0], "count": row[1]} for row in await cursor.fetchall()]
    
    # Summary rate
    if stats["total"] > 0:
        stats["summary_rate"] = round(stats["with_summary"] / stats["total"] * 100, 1)
    else:
        stats["summary_rate"] = 0
    
    return stats

# ---- Duration + View Count Update ----

class VideoMetaUpdate(BaseModel):
    duration: Optional[int] = None
    view_count: Optional[int] = None

@app.patch("/api/videos/{video_id}/meta")
async def update_video_meta(
    video_id: str,
    meta: VideoMetaUpdate,
    db: aiosqlite.Connection = Depends(get_db)
):
    """Update video metadata (duration, view count)."""
    updates = []
    params = []
    if meta.duration is not None:
        updates.append("duration = ?")
        params.append(meta.duration)
    if meta.view_count is not None:
        updates.append("view_count = ?")
        params.append(meta.view_count)
    
    if not updates:
        return {"message": "No updates provided"}
    
    params.append(video_id)
    await db.execute(f"UPDATE videos SET {', '.join(updates)} WHERE video_id = ?", params)
    await db.commit()
    return {"message": "Metadata updated"}

# ---- Include view_count in video list ----

# ---- Static frontend ----

frontend_build_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend/dist")
if os.path.exists(frontend_build_path):
    app.mount("/", StaticFiles(directory=frontend_build_path, html=True), name="frontend")
    log.info("Mounted static frontend build successfully.")
else:
    log.warning("Frontend build directory 'frontend/dist' not found. API only mode is active.")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)