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 — disabled. Authelia (Caddy forward_auth) handles auth upstream. 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/* # Skip token check if Authelia/Caddy already authenticated the user if path.startswith("/api/") and method in ("POST", "PATCH", "DELETE") and API_TOKEN: # Check if request came through Authelia (Remote-User header set by Caddy forward_auth) if request.headers.get("Remote-User"): return await call_next(request) 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)