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, db: aiosqlite.Connection = Depends(get_db) ): 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] 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"} # ---- 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)