"""WebUI FastAPI application."""
import html
import os
from pathlib import Path
from typing import List, Optional
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.responses import HTMLResponse, 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
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
@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/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)):
client = get_client()
try:
result = await client.post("/sources/sync", json={"override": override})
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)