Build Taxonomy Workbench

backend/app/api/routes_taxonomy.py

@@ -1,9 +1,77 @@
-from backend.app.schemas.taxonomy import TaxonomyListResponse
-from backend.app.services.taxonomy_service import get_active_pack
+from __future__ import annotations
+
+from argument_risk_engine.taxonomy.indexer import TaxonomyFilters
+
+from backend.app.schemas.taxonomy import (
+    TaxonomyCategoriesResponse,
+    TaxonomyEntryResponse,
+    TaxonomyListResponse,
+    TaxonomySearchResponse,
+    TaxonomySummaryResponse,
+)
+from backend.app.services import taxonomy_service
 from fastapi import APIRouter
 
+
+def _bool_filter(value: bool | str | None) -> bool | None:
+    if isinstance(value, bool) or value is None:
+        return value
+    text = str(value).strip().lower()
+    if text in {"true", "1", "yes", "enabled", "active"}:
+        return True
+    if text in {"false", "0", "no", "disabled", "inactive"}:
+        return False
+    return None
+
+
 router = APIRouter(prefix="/taxonomy", tags=["taxonomy"])
 
+
 @router.get("", response_model=TaxonomyListResponse)
-def list_taxonomy() -> TaxonomyListResponse:
-    return TaxonomyListResponse(entries=get_active_pack().entries)
+def list_taxonomy(
+    category: str | None = None,
+    pack: str | None = None,
+    academic_status: str | None = None,
+    academic_consensus: str | None = None,
+    detection_level: str | None = None,
+    activation_status: str | None = None,
+    enabled_for_classification: bool | None = None,
+    false_positive_sensitivity: str | None = None,
+) -> TaxonomyListResponse:
+    entries = taxonomy_service.list_entries(
+        TaxonomyFilters(
+            category=category,
+            pack=pack,
+            academic_status=academic_status,
+            academic_consensus=academic_consensus,
+            detection_level=detection_level,
+            activation_status=activation_status,
+            enabled_for_classification=_bool_filter(enabled_for_classification),
+            false_positive_sensitivity=false_positive_sensitivity,
+        )
+    )
+    return TaxonomyListResponse(entries=entries, total=len(entries))
+
+
+@router.get("/search", response_model=TaxonomySearchResponse)
+def search_taxonomy(q: str = "") -> TaxonomySearchResponse:
+    entries = taxonomy_service.search_taxonomy(q)
+    return TaxonomySearchResponse(entries=entries, query=q, total=len(entries))
+
+
+@router.get("/categories", response_model=TaxonomyCategoriesResponse)
+def taxonomy_categories() -> TaxonomyCategoriesResponse:
+    return TaxonomyCategoriesResponse(categories=taxonomy_service.categories())
+
+
+@router.get("/summary", response_model=TaxonomySummaryResponse)
+def taxonomy_summary() -> TaxonomySummaryResponse:
+    return TaxonomySummaryResponse(**taxonomy_service.summary())
+
+
+@router.get("/{risk_id}", response_model=TaxonomyEntryResponse)
+def get_taxonomy_entry(risk_id: str) -> TaxonomyEntryResponse | dict[str, str]:
+    entry = taxonomy_service.get_entry(risk_id)
+    if entry is None:
+        return {"detail": "taxonomy entry not found"}
+    return TaxonomyEntryResponse(entry=entry)

backend/app/api/routes_taxonomy_workbench.py

@@ -1,31 +1,101 @@
+from __future__ import annotations
+
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
-from backend.app.services.taxonomy_workbench_service import export_workbook, import_workbook, quality
-from fastapi import APIRouter, File, UploadFile
+from backend.app.schemas.taxonomy_workbench import (
+    ActivationRequest,
+    ActivationResponse,
+    TaxonomyCoverageResponse,
+    TaxonomyImportResult,
+    TaxonomyPacksResponse,
+    TaxonomyQualityResponse,
+    TaxonomyValidationResponse,
+)
+from backend.app.services.taxonomy_workbench_service import (
+    coverage,
+    export_workbook,
+    import_workbook,
+    packs,
+    quality,
+    set_activation,
+    validate_current_taxonomy,
+)
+from fastapi import APIRouter, File, Response, UploadFile
 
 router = APIRouter(prefix="/taxonomy-workbench", tags=["taxonomy-workbench"])
 
 
-@router.get("/quality")
-def taxonomy_quality() -> dict[str, object]:
-    return quality()
+@router.get("/packs", response_model=TaxonomyPacksResponse)
+def taxonomy_packs() -> TaxonomyPacksResponse:
+    return TaxonomyPacksResponse(**packs())
 
 
-@router.post("/import")
-def import_taxonomy(file: UploadFile = File(...)) -> dict[str, object]:
+@router.get("/coverage", response_model=TaxonomyCoverageResponse)
+def taxonomy_coverage() -> TaxonomyCoverageResponse:
+    return TaxonomyCoverageResponse(**coverage())
+
+
+@router.get("/quality-report", response_model=TaxonomyQualityResponse)
+def taxonomy_quality() -> TaxonomyQualityResponse:
+    return TaxonomyQualityResponse(**quality())
+
+
+@router.post("/validate", response_model=TaxonomyValidationResponse)
+def validate_taxonomy() -> TaxonomyValidationResponse:
+    return TaxonomyValidationResponse(**validate_current_taxonomy())
+
+
+@router.post("/import-excel", response_model=TaxonomyImportResult)
+def import_taxonomy_excel(file: UploadFile = File(...)) -> TaxonomyImportResult:
     suffix = Path(getattr(file, "filename", "taxonomy.xlsx") or "taxonomy.xlsx").suffix or ".xlsx"
     with NamedTemporaryFile(delete=False, suffix=suffix) as handle:
         content = file.file.read()
         handle.write(content)
         temp_path = Path(handle.name)
     try:
-        return import_workbook(temp_path)
+        return TaxonomyImportResult(**import_workbook(temp_path))
     finally:
         temp_path.unlink(missing_ok=True)
 
 
+@router.get("/export-excel")
+def export_taxonomy_excel() -> Response:
+    path = export_workbook()
+    content = path.read_bytes()
+    return Response(
+        content=content,
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={"Content-Disposition": f'attachment; filename="{path.name}"'},
+    )
+
+
+@router.patch("/entries/{risk_id}/activation", response_model=ActivationResponse)
+def update_activation(risk_id: str, request: ActivationRequest) -> ActivationResponse | dict[str, str]:
+    try:
+        result = set_activation(
+            risk_id=risk_id,
+            activation_status=request.activation_status,
+            enabled_for_classification=request.enabled_for_classification,
+        )
+    except KeyError:
+        return {"detail": "taxonomy entry not found"}
+    except ValueError as error:
+        return {"detail": str(error)}
+    return ActivationResponse(**result)
+
+
+# Backwards-compatible aliases for the original MVP endpoints.
+@router.get("/quality")
+def taxonomy_quality_alias() -> TaxonomyQualityResponse:
+    return taxonomy_quality()
+
+
+@router.post("/import")
+def import_taxonomy_alias(file: UploadFile = File(...)) -> TaxonomyImportResult:
+    return import_taxonomy_excel(file)
+
+
 @router.get("/export")
-def export_taxonomy() -> dict[str, str]:
-    path = export_workbook(Path("data/taxonomy/exports/taxonomy.xlsx"))
-    return {"path": str(path)}
+def export_taxonomy_alias() -> Response:
+    return export_taxonomy_excel()

backend/app/schemas/taxonomy.py

@@ -1,7 +1,33 @@
+from __future__ import annotations
+
 from argument_risk_engine.taxonomy.models import TaxonomyEntry
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class TaxonomyListResponse(BaseModel):
     entries: list[TaxonomyEntry]
+    total: int = 0
+
+
+class TaxonomyEntryResponse(BaseModel):
+    entry: TaxonomyEntry
+
+
+class TaxonomyCategoriesResponse(BaseModel):
+    categories: list[str]
+
+
+class TaxonomySummaryResponse(BaseModel):
+    entry_count: int
+    active_count: int
+    enabled_for_classification_count: int
+    by_category: dict[str, int] = Field(default_factory=dict)
+    by_pack: dict[str, int] = Field(default_factory=dict)
+    by_activation_status: dict[str, int] = Field(default_factory=dict)
+
+
+class TaxonomySearchResponse(BaseModel):
+    entries: list[TaxonomyEntry]
+    query: str
+    total: int = 0

backend/app/schemas/taxonomy_workbench.py

@@ -1,6 +1,67 @@
-from pydantic import BaseModel
+from __future__ import annotations
+
+from argument_risk_engine.taxonomy.models import TaxonomyEntry
+
+from pydantic import BaseModel, Field
+
+
+class TaxonomyPackSummary(BaseModel):
+    pack: str
+    entry_count: int
+    active_count: int
+    enabled_for_classification_count: int
+
+
+class TaxonomyPacksResponse(BaseModel):
+    packs: list[TaxonomyPackSummary]
+
+
+class TaxonomyCoverageResponse(BaseModel):
+    entry_count: int
+    active_count: int
+    enabled_for_classification_count: int
+    review_required_count: int = 0
+    deprecated_count: int = 0
+    by_category: dict[str, int] = Field(default_factory=dict)
+    by_pack: dict[str, int] = Field(default_factory=dict)
+    by_detection_level: dict[str, int] = Field(default_factory=dict)
+    by_academic_status: dict[str, int] = Field(default_factory=dict)
+    by_activation_status: dict[str, int] = Field(default_factory=dict)
+    missing_examples_count: int = 0
+    missing_false_positive_warnings_count: int = 0
 
 
 class TaxonomyQualityResponse(BaseModel):
-    errors: list[str]
+    ok: bool
     entry_count: int
+    active_classification_count: int = 0
+    error_count: int = 0
+    warning_count: int = 0
+    errors: list[dict[str, object]] = Field(default_factory=list)
+    warnings: list[dict[str, object]] = Field(default_factory=list)
+    coverage: dict[str, object] = Field(default_factory=dict)
+
+
+class TaxonomyImportResult(BaseModel):
+    entry_count: int
+    errors: list[str] = Field(default_factory=list)
+    warnings: list[str] = Field(default_factory=list)
+    backup_paths: list[str] = Field(default_factory=list)
+
+
+class TaxonomyValidationResponse(BaseModel):
+    ok: bool
+    entry_count: int
+    active_classification_count: int = 0
+    errors: list[dict[str, object]] = Field(default_factory=list)
+    warnings: list[dict[str, object]] = Field(default_factory=list)
+
+
+class ActivationRequest(BaseModel):
+    activation_status: str = "active"
+    enabled_for_classification: bool | None = None
+
+
+class ActivationResponse(BaseModel):
+    entry: TaxonomyEntry
+    backup_path: str

backend/app/services/taxonomy_service.py

@@ -1,5 +1,17 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from argument_risk_engine.taxonomy.indexer import (
+    TaxonomyFilters,
+    build_index,
+    facet_counts,
+    filter_entries,
+    search_entries,
+)
 from argument_risk_engine.taxonomy.loader import load_taxonomy_pack, save_taxonomy_pack
-from argument_risk_engine.taxonomy.models import TaxonomyPack
+from argument_risk_engine.taxonomy.models import TaxonomyEntry, TaxonomyPack
+from argument_risk_engine.taxonomy.quality_audit import coverage_report
 
 from backend.app.core.paths import TAXONOMY_PACK_PATH
 
@@ -10,3 +22,38 @@ def get_active_pack() -> TaxonomyPack:
 
 def save_active_pack(pack: TaxonomyPack) -> None:
     save_taxonomy_pack(pack, TAXONOMY_PACK_PATH)
+
+
+def list_entries(filters: TaxonomyFilters | None = None) -> list[TaxonomyEntry]:
+    entries = get_active_pack().entries
+    if filters is None:
+        return entries
+    return filter_entries(entries, filters)
+
+
+def get_entry(risk_id: str) -> TaxonomyEntry | None:
+    return build_index(get_active_pack()).get(risk_id)
+
+
+def search_taxonomy(query: str) -> list[TaxonomyEntry]:
+    return search_entries(get_active_pack().entries, query)
+
+
+def categories() -> list[str]:
+    return sorted(facet_counts(get_active_pack().entries, "canonical_category"))
+
+
+def summary() -> dict[str, object]:
+    report = coverage_report(get_active_pack())
+    return {
+        "entry_count": report["entry_count"],
+        "active_count": report["active_count"],
+        "enabled_for_classification_count": report["enabled_for_classification_count"],
+        "by_category": report["by_category"],
+        "by_pack": report["by_pack"],
+        "by_activation_status": report["by_activation_status"],
+    }
+
+
+def taxonomy_path() -> Path:
+    return TAXONOMY_PACK_PATH