Gan, Jimmy
296 lines
11 KiB
Python
296 lines
11 KiB
Python
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)
|