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), repository_id, analysis_run_id)}
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)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-abilities/{candidate_ability_id}/reject"
)
def reject_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_ability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.reject_candidate_ability(
repository_id,
analysis_run_id,
candidate_ability_id,
notes="Rejected from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-abilities/{candidate_ability_id}/edit"
)
def edit_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_ability_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.edit_candidate_ability(
repository_id,
analysis_run_id,
candidate_ability_id,
name=name,
description=description,
confidence=confidence,
notes="Edited from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{candidate_capability_id}/edit"
)
def edit_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.edit_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
name=name,
description=description,
confidence=confidence,
notes="Edited from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
def render_candidate_graph(graph: dict, repository_id: int, analysis_run_id: int) -> str:
abilities = graph.get("abilities", [])
if not abilities:
return 'No candidates generated.
'
items = []
for ability in abilities:
capabilities = "".join(
render_candidate_capability(capability, repository_id, analysis_run_id)
for capability in ability["capabilities"]
)
items.append(
f"""
{escape(ability['name'])}
{escape(ability['status'])}
{ability['confidence']:.2f}
{render_candidate_ability_actions(ability, repository_id, analysis_run_id)}
{escape(ability['description'])}
{render_candidate_edit_form('candidate-abilities', ability, repository_id, analysis_run_id)}
{render_sources(ability['source_refs'])}
"""
)
return f''
def render_candidate_ability_actions(
ability: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
if ability["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-abilities/{ability['id']}/reject"
)
return f"""
"""
def render_candidate_edit_form(
collection: str,
candidate: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
if candidate["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{candidate['id']}/edit"
)
confidence = f"{candidate['confidence']:.2f}"
return f"""
"""
def render_candidate_capability(
capability: dict,
repository_id: int,
analysis_run_id: int,
) -> 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_candidate_edit_form('candidate-capabilities', capability, repository_id, analysis_run_id)}
{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}
"