from __future__ import annotations
from dataclasses import asdict
from html import escape
from fastapi import APIRouter, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from repo_registry.core.service import RegistryService
from repo_registry.web_api.app import get_service
router = APIRouter(include_in_schema=False)
def page(title: str, body: str) -> HTMLResponse:
return HTMLResponse(
f"""
{escape(title)} · Repository Ability Registry
{body}
"""
)
@router.get("/ui")
def repository_index(service: RegistryService = Depends(get_service)) -> HTMLResponse:
repositories = service.list_repositories()
rows = "\n".join(
f"""
| {escape(repo.name)} |
{escape(repo.status)} |
{escape(repo.branch)} |
{escape(repo.url)} |
"""
for repo in repositories
)
body = f"""
Repositories
Registry
| Name | Status | Branch | Source |
|---|
{rows or '| No repositories yet. |
'}
"""
return page("Repositories", body)
@router.post("/ui/repos")
def create_repository_from_form(
url: str = Form(...),
branch: str = Form("main"),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
repository = service.register_repository(
url=url,
branch=branch or "main",
)
return RedirectResponse(f"/ui/repos/{repository.id}", status_code=303)
@router.get("/ui/repos/{repository_id}")
def repository_detail(
repository_id: int,
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
repository = service.get_repository(repository_id)
runs = service.list_analysis_runs(repository_id)
ability_map = service.ability_map(repository_id)
run_rows = "\n".join(
f"""
| #{run.id} |
{escape(run.status)} |
{escape(run.started_at)} |
{escape(run.error_message or '')} |
"""
for run in runs
)
body = f"""
{escape(repository.name)}
Back
{escape(repository.description or '')}
{escape(repository.status)} {escape(repository.url)}
Run Analysis
Analysis Runs
| Run | Status | Started | Error |
|---|
{run_rows or '| No runs yet. |
'}
Approved Ability Map
{render_ability_map(asdict(ability_map))}
"""
return page(repository.name, body)
@router.post("/ui/repos/{repository_id}/analysis-runs")
def create_analysis_run_from_form(
repository_id: int,
source_path: str = Form(""),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
summary = service.analyze_repository(
repository_id,
source_path=source_path or None,
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{summary.analysis_run.id}",
status_code=303,
)
@router.get("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}")
def analysis_run_detail(
repository_id: int,
analysis_run_id: int,
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
repository = service.get_repository(repository_id)
candidate_graph = service.candidate_graph(repository_id, analysis_run_id)
facts = service.list_observed_facts(repository_id, analysis_run_id)
fact_rows = "\n".join(
f"""
| {escape(fact.kind)} |
{escape(fact.name)} |
{escape(fact.path)} |
{escape(fact.value)} |
"""
for fact in facts
)
body = f"""
{escape(repository.name)} · Run #{analysis_run_id}
Repository
Candidate Graph
{render_candidate_graph(asdict(candidate_graph))}
Observed Facts
| Kind | Name | Path | Value |
|---|
{fact_rows or '| No observed facts. |
'}
"""
return page(f"{repository.name} Run {analysis_run_id}", body)
@router.post("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-graph/approve")
def approve_candidate_graph_from_form(
repository_id: int,
analysis_run_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.approve_candidate_graph(
repository_id,
analysis_run_id,
notes="Approved from web UI",
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
def render_candidate_graph(graph: dict) -> str:
abilities = graph.get("abilities", [])
if not abilities:
return 'No candidates generated.
'
items = []
for ability in abilities:
capabilities = "".join(render_candidate_capability(capability) for capability in ability["capabilities"])
items.append(
f"""
{escape(ability['name'])}
{escape(ability['status'])}
{ability['confidence']:.2f}
{escape(ability['description'])}
{render_sources(ability['source_refs'])}
"""
)
return f''
def render_candidate_capability(capability: dict) -> str:
features = "".join(
f'{escape(feature["name"])} {escape(feature["type"])} {escape(feature["location"])}'
for feature in capability["features"]
)
evidence = "".join(
f'{escape(item["type"])} {escape(item["strength"])} {escape(item["reference"])}'
for item in capability["evidence"]
)
return f"""
{escape(capability['name'])}
{escape(capability['status'])}
{capability['confidence']:.2f}
{escape(capability['description'])}
{render_sources(capability['source_refs'])}
Features
{features or '- No feature candidates.
'}
Evidence
{evidence or '- No evidence candidates.
'}
"""
def render_ability_map(ability_map: dict) -> str:
abilities = ability_map.get("abilities", [])
if not abilities:
return 'No approved entries yet.
'
items = []
for ability in abilities:
capabilities = []
for capability in ability["capabilities"]:
features = "".join(
f'{escape(feature["name"])} {escape(feature["type"])} {escape(feature["location"])}'
for feature in capability["features"]
)
evidence = "".join(
f'{escape(item["type"])} {escape(item["strength"])} {escape(item["reference"])}'
for item in capability["evidence"]
)
capabilities.append(
f"""
{escape(capability['name'])}
{escape(capability['description'])}
"""
)
items.append(
f"""
{escape(ability['name'])}
{escape(ability['description'])}
"""
)
return f''
def render_sources(source_refs: list[dict]) -> str:
if not source_refs:
return ""
sources = ", ".join(
f'{escape(ref["kind"])}:{escape(ref["path"] or ref["name"])}'
for ref in source_refs[:5]
)
if len(source_refs) > 5:
sources += f' +{len(source_refs) - 5} more'
return f"{sources}
"