Implement Export and Reporting features (PDF, CSV exports) - resolves #104"

Co-authored-by: bg-playground <259109604+bg-playground@users.noreply.github.com>

Author: Copilot

backend/app/api/analytics.py

@@ -1,11 +1,14 @@
 """API endpoints for Suggestion Analytics"""
 
+import csv
+import io
 from typing import Any
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.auth.dependencies import get_current_user
+from app.crud import traceability as traceability_crud
 from app.db.session import get_db
 from app.models.user import User
 from app.services.analytics import SuggestionAnalytics
@@ -83,3 +86,63 @@ async def get_algorithm_comparison(
     """
     analytics = SuggestionAnalytics(db)
     return await analytics.get_algorithm_comparison()
+
+
+@router.get("/metrics/export/csv")
+async def export_metrics_csv(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """
+    Export system metrics as CSV including coverage stats, suggestion counts by status,
+    and algorithm breakdown.
+    """
+    metrics = await traceability_crud.get_metrics(db)
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+
+    # Coverage stats section
+    writer.writerow(["Coverage Statistics"])
+    writer.writerow(["Metric", "Value"])
+    writer.writerow(["Coverage Percentage", f"{metrics.coverage_percentage:.2f}%"])
+    writer.writerow(["Total Requirements", metrics.total_requirements])
+    writer.writerow(["Total Test Cases", metrics.total_test_cases])
+    writer.writerow(["Total Links", metrics.total_links])
+    writer.writerow(["Manual Links", metrics.manual_links])
+    writer.writerow(["AI Suggested Links", metrics.ai_suggested_links])
+
+    writer.writerow([])  # Separator
+
+    # Suggestion counts by status
+    writer.writerow(["Suggestion Counts by Status"])
+    writer.writerow(["Status", "Count"])
+    writer.writerow(["Total", metrics.total_suggestions])
+    writer.writerow(["Accepted", metrics.accepted_suggestions])
+    writer.writerow(["Rejected", metrics.rejected_suggestions])
+    writer.writerow(["Pending", metrics.pending_suggestions])
+    writer.writerow(["Suggestion Acceptance Rate", f"{metrics.suggestion_acceptance_rate:.2f}%"])
+
+    writer.writerow([])  # Separator
+
+    # Algorithm breakdown
+    writer.writerow(["Algorithm Breakdown"])
+    writer.writerow(["Algorithm", "Total", "Accepted", "Rejected", "Pending", "Acceptance Rate"])
+    for algo in metrics.algorithm_breakdown:
+        writer.writerow(
+            [
+                algo.algorithm,
+                algo.total_suggestions,
+                algo.accepted_suggestions,
+                algo.rejected_suggestions,
+                algo.pending_suggestions,
+                f"{algo.acceptance_rate:.2f}%",
+            ]
+        )
+
+    csv_content = output.getvalue()
+    return Response(
+        content=csv_content,
+        media_type="text/csv",
+        headers={"Content-Disposition": "attachment; filename=metrics.csv"},
+    )

backend/app/api/suggestions.py

@@ -1,21 +1,115 @@
 """API endpoints for AI Suggestion Generation"""
 
-from typing import Any
+import csv
+import io
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 
 from app.ai_suggestions.config import SuggestionConfig
 from app.ai_suggestions.engine import SuggestionEngine
-from app.auth.dependencies import require_admin
+from app.auth.dependencies import get_current_user, require_admin
 from app.crud.audit_log import create_audit_entry
 from app.db.session import get_db
+from app.models.suggestion import LinkSuggestion, SuggestionStatus
 from app.models.user import User
 
 router = APIRouter()
 
 
-@router.post("/suggestions/generate", response_model=dict[str, Any])
+@router.get("/suggestions/export/csv")
+async def export_suggestions_csv(
+    status: str | None = Query(None, description="Filter by status: pending, accepted, rejected"),
+    algorithm: str | None = Query(None, description="Filter by algorithm"),
+    min_score: float | None = Query(None, ge=0.0, le=1.0, description="Minimum confidence score"),
+    max_score: float | None = Query(None, ge=0.0, le=1.0, description="Maximum confidence score"),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """
+    Export suggestions as CSV with optional filtering.
+
+    Supports filtering by status, algorithm, and confidence score range.
+    """
+    query = select(LinkSuggestion).options(
+        selectinload(LinkSuggestion.requirement),
+        selectinload(LinkSuggestion.test_case),
+    )
+
+    if status is not None:
+        try:
+            status_enum = SuggestionStatus(status.lower())
+            query = query.where(LinkSuggestion.status == status_enum)
+        except ValueError:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid status '{status}'. Must be one of: pending, accepted, rejected.",
+            )
+
+    if algorithm is not None:
+        from app.models.suggestion import SuggestionMethod
+
+        try:
+            method_enum = SuggestionMethod(algorithm.lower())
+            query = query.where(LinkSuggestion.suggestion_method == method_enum)
+        except ValueError:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid algorithm '{algorithm}'.",
+            )
+
+    if min_score is not None:
+        query = query.where(LinkSuggestion.similarity_score >= min_score)
+
+    if max_score is not None:
+        query = query.where(LinkSuggestion.similarity_score <= max_score)
+
+    result = await db.execute(query)
+    suggestions = result.scalars().all()
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(
+        [
+            "ID",
+            "Requirement Title",
+            "Test Case Title",
+            "Algorithm",
+            "Confidence Score",
+            "Status",
+            "Reviewed By",
+            "Feedback",
+            "Created At",
+            "Reviewed At",
+        ]
+    )
+    for s in suggestions:
+        writer.writerow(
+            [
+                str(s.id),
+                s.requirement.title if s.requirement else "",
+                s.test_case.title if s.test_case else "",
+                s.suggestion_method.value if s.suggestion_method else "",
+                f"{s.similarity_score:.4f}" if s.similarity_score is not None else "",
+                s.status.value if s.status else "",
+                s.reviewed_by or "",
+                s.feedback or "",
+                s.created_at.isoformat() if s.created_at else "",
+                s.reviewed_at.isoformat() if s.reviewed_at else "",
+            ]
+        )
+
+    csv_content = output.getvalue()
+    return Response(
+        content=csv_content,
+        media_type="text/csv",
+        headers={"Content-Disposition": "attachment; filename=suggestions.csv"},
+    )
+
+
+@router.post("/suggestions/generate", response_model=dict)
 async def generate_suggestions(
     algorithm: str | None = Query(
         None, description="Algorithm to use: 'tfidf', 'keyword', 'hybrid', or 'llm'. Uses default if not specified."

backend/app/api/traceability.py

@@ -2,9 +2,15 @@
 
 import csv
 import io
+from datetime import datetime, timezone
 from typing import Literal
 
 from fastapi import APIRouter, Depends, Query, Response
+from reportlab.lib import colors
+from reportlab.lib.pagesizes import letter
+from reportlab.lib.styles import getSampleStyleSheet
+from reportlab.lib.units import inch
+from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.auth.dependencies import get_current_user
@@ -53,15 +59,15 @@ async def get_metrics(
 
 @router.get("/traceability-matrix/export")
 async def export_traceability_matrix(
-    format: Literal["csv", "json"] = Query(..., description="Export format: csv or json"),
+    format: Literal["csv", "json", "pdf"] = Query(..., description="Export format: csv, json, or pdf"),
     db: AsyncSession = Depends(get_db),
     current_user: User = Depends(get_current_user),
 ):
     """
-    Export the traceability matrix in CSV or JSON format.
+    Export the traceability matrix in CSV, JSON, or PDF format.
 
     Args:
-        format: Export format (csv or json)
+        format: Export format (csv, json, or pdf)
 
     Returns:
         File download with appropriate content type
@@ -144,3 +150,146 @@ async def export_traceability_matrix(
             media_type="application/json",
             headers={"Content-Disposition": "attachment; filename=traceability_matrix.json"},
         )
+
+    elif format == "pdf":
+        # Build PDF using reportlab
+        buffer = io.BytesIO()
+        doc = SimpleDocTemplate(buffer, pagesize=letter, topMargin=0.75 * inch, bottomMargin=0.75 * inch)
+        styles = getSampleStyleSheet()
+        elements = []
+
+        # Title
+        elements.append(Paragraph("BGSTM Traceability Matrix Report", styles["Title"]))
+        gen_date = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+        elements.append(Paragraph(f"Generated: {gen_date}", styles["Normal"]))
+        elements.append(Spacer(1, 0.2 * inch))
+
+        # Coverage Summary
+        elements.append(Paragraph("Coverage Summary", styles["Heading2"]))
+        summary_data = [
+            ["Metric", "Value"],
+            ["Total Requirements", str(matrix_data.total_requirements)],
+            ["Covered Requirements", str(matrix_data.covered_requirements)],
+            ["Uncovered Requirements", str(matrix_data.uncovered_requirements)],
+            ["Coverage Percentage", f"{matrix_data.coverage_percentage:.1f}%"],
+            ["Total Test Cases", str(matrix_data.total_test_cases)],
+            ["Orphan Test Cases", str(matrix_data.orphan_test_cases)],
+        ]
+        summary_table = Table(summary_data, colWidths=[3 * inch, 2 * inch])
+        summary_table.setStyle(
+            TableStyle(
+                [
+                    ("BACKGROUND", (0, 0), (-1, 0), colors.grey),
+                    ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
+                    ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
+                    ("ALIGN", (0, 0), (-1, -1), "LEFT"),
+                    ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+                    ("BACKGROUND", (0, 1), (-1, -1), colors.beige),
+                    ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.lightgrey]),
+                ]
+            )
+        )
+        elements.append(summary_table)
+        elements.append(Spacer(1, 0.2 * inch))
+
+        # Gap Analysis
+        uncovered = [r for r in matrix_data.matrix if r.coverage_status == "uncovered"]
+        elements.append(Paragraph("Gap Analysis – Uncovered Requirements", styles["Heading2"]))
+        if uncovered:
+            gap_data = [["External ID", "Requirement Title"]]
+            for r in uncovered:
+                gap_data.append([r.external_id or "", r.requirement_title])
+            gap_table = Table(gap_data, colWidths=[1.5 * inch, 5 * inch])
+            gap_table.setStyle(
+                TableStyle(
+                    [
+                        ("BACKGROUND", (0, 0), (-1, 0), colors.red),
+                        ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
+                        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
+                        ("ALIGN", (0, 0), (-1, -1), "LEFT"),
+                        ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+                        ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.lightgrey]),
+                    ]
+                )
+            )
+            elements.append(gap_table)
+        else:
+            elements.append(Paragraph("No uncovered requirements.", styles["Normal"]))
+        elements.append(Spacer(1, 0.2 * inch))
+
+        # Full Matrix Table
+        elements.append(Paragraph("Requirements Coverage Matrix", styles["Heading2"]))
+        matrix_table_data = [["Req ID", "Requirement Title", "Coverage", "Test Case Title", "Link Type", "Score"]]
+        for req_coverage in matrix_data.matrix:
+            if req_coverage.linked_test_cases:
+                for i, tc in enumerate(req_coverage.linked_test_cases):
+                    score = f"{tc.confidence_score:.2f}" if tc.confidence_score is not None else "—"
+                    matrix_table_data.append(
+                        [
+                            req_coverage.external_id or "" if i == 0 else "",
+                            req_coverage.requirement_title if i == 0 else "",
+                            req_coverage.coverage_status.replace("_", " ").title() if i == 0 else "",
+                            tc.title,
+                            tc.link_type,
+                            score,
+                        ]
+                    )
+            else:
+                matrix_table_data.append(
+                    [
+                        req_coverage.external_id or "",
+                        req_coverage.requirement_title,
+                        req_coverage.coverage_status.replace("_", " ").title(),
+                        "—",
+                        "—",
+                        "—",
+                    ]
+                )
+        col_widths = [1.0 * inch, 2.2 * inch, 1.1 * inch, 2.0 * inch, 0.8 * inch, 0.6 * inch]
+        matrix_pdf_table = Table(matrix_table_data, colWidths=col_widths, repeatRows=1)
+        matrix_pdf_table.setStyle(
+            TableStyle(
+                [
+                    ("BACKGROUND", (0, 0), (-1, 0), colors.darkblue),
+                    ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
+                    ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
+                    ("FONTSIZE", (0, 0), (-1, -1), 8),
+                    ("ALIGN", (0, 0), (-1, -1), "LEFT"),
+                    ("VALIGN", (0, 0), (-1, -1), "TOP"),
+                    ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+                    ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.lightgrey]),
+                ]
+            )
+        )
+        elements.append(matrix_pdf_table)
+        elements.append(Spacer(1, 0.2 * inch))
+
+        # Orphan Test Cases
+        if matrix_data.orphans:
+            elements.append(Paragraph("Orphan Test Cases (No Linked Requirements)", styles["Heading2"]))
+            orphan_data = [["External ID", "Test Case Title"]]
+            for orphan in matrix_data.orphans:
+                orphan_data.append([orphan.external_id or "", orphan.title])
+            orphan_table = Table(orphan_data, colWidths=[1.5 * inch, 5 * inch])
+            orphan_table.setStyle(
+                TableStyle(
+                    [
+                        ("BACKGROUND", (0, 0), (-1, 0), colors.orange),
+                        ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
+                        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
+                        ("FONTSIZE", (0, 0), (-1, -1), 8),
+                        ("ALIGN", (0, 0), (-1, -1), "LEFT"),
+                        ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+                        ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.lightgrey]),
+                    ]
+                )
+            )
+            elements.append(orphan_table)
+
+        doc.build(elements)
+        pdf_bytes = buffer.getvalue()
+        return Response(
+            content=pdf_bytes,
+            media_type="application/pdf",
+            headers={"Content-Disposition": "attachment; filename=traceability_matrix.pdf"},
+        )

backend/requirements.txt

@@ -15,6 +15,7 @@ scikit-learn==1.8.0
 PyJWT==2.10.1
 bcrypt>=4.0.0
 email-validator==2.2.0
+reportlab==4.4.1
 
 # Optional dependencies for LLM embeddings
 # Uncomment the lines below if using LLM-based similarity