"""WebUI FastAPI application."""
import asyncio
import html
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from .api_client import DocsAPIClient
app = FastAPI(
title="Context7 Docs WebUI",
description="Web dashboard for managing documentation system",
version="1.0.0",
)
templates = Jinja2Templates(directory=os.path.join(os.path.dirname(__file__), "templates"))
templates.env.globals["escapeHtml"] = lambda value: html.escape(str(value or ""))
app.mount("/static", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")), name="static")
_client: Optional[DocsAPIClient] = None
_sync_jobs: Dict[str, Dict[str, Any]] = {}
_sync_tasks: set[asyncio.Task] = set()
def get_client() -> DocsAPIClient:
global _client
if _client is None:
_client = DocsAPIClient(
os.environ.get("DOCS_API_URL", "http://docs-api:8787"),
os.environ.get("WEBUI_API_KEY"),
)
return _client
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
async def run_sync_job(job_id: str, override: bool) -> None:
job = _sync_jobs[job_id]
job["status"] = "running"
job["started_at"] = utc_now()
try:
result = await get_client().post(
"/sources/sync",
json={"override": override},
timeout=None,
)
job["result"] = result
job["status"] = "succeeded" if result.get("success") else "failed"
except Exception as exc:
job["status"] = "failed"
job["error"] = str(exc)
finally:
job["finished_at"] = utc_now()
@app.on_event("shutdown")
async def shutdown() -> None:
if _client is not None:
await _client.close()
def page(title: str, body: str) -> HTMLResponse:
return HTMLResponse(
f"""
{html.escape(title)}
{body}"""
)
@app.get("/")
async def dashboard(request: Request):
client = get_client()
health = await client.health()
try:
collections_data = await client.get("/collections")
total_vectors = sum(
item.get("vectors", 0)
for item in collections_data.get("collections", {}).values()
if isinstance(item, dict)
)
except Exception:
total_vectors = 0
try:
libs_data = await client.get("/libraries")
libraries = libs_data.get("libraries", [])
except Exception:
libraries = []
return templates.TemplateResponse(
"dashboard.html",
{"request": request, "health": health, "vectors": total_vectors, "libraries": libraries},
)
@app.post("/actions/ingest-all")
async def ingest_all():
client = get_client()
try:
result = await client.post("/ingest/all")
body = f"Ingestion Complete
{html.escape(str(result))}Back"
except Exception as e:
body = f"Ingestion Failed
{html.escape(str(e))}Back"
return page("Ingestion", body)
@app.post("/actions/sync-sources")
async def sync_sources_action():
client = get_client()
try:
result = await client.post("/sources/sync", json={"override": False})
body = f"Git Sync Complete
{html.escape(str(result))}Back"
except Exception as e:
body = f"Git Sync Failed
{html.escape(str(e))}Back"
return page("Git Sync", body)
@app.get("/libraries")
async def libraries(request: Request):
client = get_client()
try:
data = await client.get("/libraries")
libraries_data = data.get("libraries", [])
except Exception:
libraries_data = []
return templates.TemplateResponse("libraries.html", {"request": request, "data": libraries_data})
@app.post("/libraries/create")
async def create_library(
library_id: str = Form(...),
name: str = Form(...),
description: Optional[str] = Form(None),
):
client = get_client()
try:
result = await client.post(
f"/api/v1/libraries/{library_id.strip()}",
data={"name": name, "description": description or ""},
)
body = f"Library Created
{html.escape(str(result))}Back"
except Exception as e:
body = f"Create Failed
{html.escape(str(e))}Back"
return page("Library Created", body)
@app.post("/libraries/{library_id}/ingest")
async def ingest_library(library_id: str):
client = get_client()
try:
result = await client.post(f"/ingest/{library_id}")
body = f"Ingestion Complete
{html.escape(str(result))}Back"
except Exception as e:
body = f"Ingestion Failed
{html.escape(str(e))}Back"
return page("Ingest Library", body)
@app.post("/libraries/{library_id}/delete")
async def delete_library(library_id: str):
client = get_client()
try:
result = await client.delete(f"/api/v1/libraries/{library_id}")
body = f"Library Deleted
{html.escape(str(result))}Back"
except Exception as e:
body = f"Delete Failed
{html.escape(str(e))}Back"
return page("Delete Library", body)
@app.get("/libraries/{library_id}/docs")
async def view_library_docs(library_id: str):
client = get_client()
try:
result = await client.get(f"/docs/{library_id}")
content = result.get("content", "")
except Exception as e:
content = str(e)
return page(
f"Docs: {library_id}",
f"{html.escape(library_id)}
{html.escape(content)}Back",
)
@app.get("/upload")
async def upload_form(request: Request):
client = get_client()
try:
libs_data = await client.get("/libraries")
libraries = libs_data.get("libraries", [])
except Exception:
libraries = []
return templates.TemplateResponse("upload.html", {"request": request, "libraries": libraries})
@app.post("/upload")
async def upload_file(
request: Request,
library_id: str = Form(""),
ingest_after_upload: Optional[str] = Form(None),
files: List[UploadFile] = File(...),
):
client = get_client()
results = []
total_size = 0
for upload in files:
filename = upload.filename or "upload.txt"
target_library = library_id.strip()
if not target_library:
target_library = Path(filename).stem.lower().replace(" ", "-") or "uploaded"
try:
contents = await upload.read()
total_size += len(contents)
result = await client.upload_file(target_library, filename, contents)
results.append({"filename": filename, "status": "success", "message": result})
except Exception as e:
results.append({"filename": filename, "status": "error", "message": str(e)})
if ingest_after_upload == "on":
for result in list(results):
if result["status"] != "success":
continue
target_library = result["message"]["library_id"]
try:
ingest_result = await client.post(f"/ingest/{target_library}")
results.append({"filename": "__INGEST__", "status": "success", "message": ingest_result})
except Exception as e:
results.append({"filename": "__INGEST__", "status": "error", "message": str(e)})
return templates.TemplateResponse(
"upload.html",
{"request": request, "libraries": [], "results": results, "total_size_bytes": total_size},
)
@app.get("/search")
async def search_form(request: Request):
return templates.TemplateResponse("search.html", {"request": request, "query": "", "results": []})
@app.get("/search/results")
async def search_results(request: Request, q: str = "", limit: int = 10):
client = get_client()
results = []
if q:
try:
data = await client.post("/search", json={"query": q, "library_id": None, "limit": limit})
results = data.get("results", [])
except Exception:
results = []
return templates.TemplateResponse(
"search.html",
{"request": request, "query": q, "results": results, "limit": limit},
)
@app.get("/sources")
async def sources_page(request: Request):
client = get_client()
try:
data = await client.get("/api/v1/sources")
sources = data.get("sources", [])
except Exception:
sources = []
return templates.TemplateResponse("sources.html", {"request": request, "sources": sources})
def parse_path_list(value: str) -> List[str]:
paths = []
for line in value.replace(",", "\n").splitlines():
item = line.strip()
if item:
paths.append(item)
return paths
@app.post("/sources/browse")
async def browse_source(repo_url: str = Form(...), branch: str = Form("main")):
client = get_client()
try:
result = await client.post(
"/api/v1/sources/browse",
json={"repo_url": repo_url, "branch": branch},
)
return JSONResponse(content=result)
except Exception as e:
return JSONResponse(status_code=400, content={"success": False, "error": str(e)})
@app.post("/sources/add")
async def add_source(
library_id: str = Form(...),
repo_url: str = Form(...),
name: str = Form(""),
description: str = Form(""),
branch: str = Form("main"),
include_paths: str = Form("docs"),
exclude_paths: str = Form("node_modules\n.git"),
):
client = get_client()
payload = {
"library_id": library_id,
"repo_url": repo_url,
"name": name or None,
"description": description or None,
"branch": branch or "main",
"include_paths": parse_path_list(include_paths),
"exclude_paths": parse_path_list(exclude_paths),
}
try:
result = await client.post("/api/v1/sources", json=payload)
body = f"Git Source Saved
{html.escape(str(result))}Back"
except Exception as e:
body = f"Save Failed
{html.escape(str(e))}Back"
return page("Git Source Saved", body)
@app.post("/sources/sync")
async def sync_sources(override: bool = Form(False)):
job_id = uuid.uuid4().hex
_sync_jobs[job_id] = {
"id": job_id,
"status": "queued",
"created_at": utc_now(),
"started_at": None,
"finished_at": None,
"result": None,
"error": None,
}
task = asyncio.create_task(run_sync_job(job_id, override))
_sync_tasks.add(task)
task.add_done_callback(_sync_tasks.discard)
return RedirectResponse(url=f"/sources/jobs/{job_id}", status_code=303)
@app.get("/sources/jobs/{job_id}")
async def sync_job_page(request: Request, job_id: str):
job = _sync_jobs.get(job_id)
if job is None:
return page("Git Sync Not Found", "Git Sync Not Found
Back")
return templates.TemplateResponse("sync_job.html", {"request": request, "job": job})
@app.get("/sources/jobs/{job_id}/status")
async def sync_job_status(job_id: str):
job = _sync_jobs.get(job_id)
if job is None:
return JSONResponse(status_code=404, content={"error": "Sync job not found"})
return job