feat: add web dashboard workflow and report history

from __future__ import annotations

from datetime import datetime

from typing import Any

from sqlalchemy import Column, DateTime, Integer, String

from sqlalchemy.types import JSON

from api.database import Base

class AnalysisHistory(Base):

    """Persisted analysis result history for report preview/download.

    The existing analysis service keeps results in memory. This table stores the

    same result payload in the database so previous reports can be loaded again

    after a server restart.

    """

    __tablename__ = "analysis_histories"

    id = Column(Integer, primary_key=True, index=True)

    analysis_id = Column(String(64), unique=True, index=True, nullable=False)

    dataset_id = Column(String(128), index=True, nullable=False)

    report_name = Column(String(255), nullable=True)

    dataset_filename = Column(String(255), nullable=True)

    results_json = Column(JSON, nullable=False)

    config_json = Column(JSON, nullable=True)

    overall_score = Column(String(32), nullable=True)

    weighting_method = Column(String(64), nullable=True)

    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

    def to_stored_result(self) -> dict[str, Any]:

        """Convert DB row back into the in-memory result shape."""

        return {

            "dataset_id": self.dataset_id,

            "results": self.results_json or {},

            "config": self.config_json or {},

        }

    def to_report_item(self) -> dict[str, Any]:

        """Compact representation used by the Reports history API."""

        results = self.results_json or {}

        return {

            "id": self.id,

            "analysis_id": self.analysis_id,

            "report_name": self.report_name or f"Report {self.id}",

            "dataset_id": self.dataset_id,

            "dataset_filename": self.dataset_filename,

            "overall_score": self.overall_score,

            "weighting_method": self.weighting_method,

            "metric_count": len(results),

            "created_at": self.created_at.isoformat() if self.created_at else None,

        }

 No newline at end of file

from __future__ import annotations

from io import BytesIO

from typing import Any

from fastapi import APIRouter, HTTPException

from pydantic import BaseModel, Field

from fastapi.encoders import jsonable_encoder

from fastapi.responses import Response, StreamingResponse

from api.services import analysis_service

router = APIRouter()

service = analysis_service

class ReportRenameRequest(BaseModel):

    report_name: str = Field(..., min_length=1, max_length=255)

def _get_summary_or_404(analysis_id: str) -> dict[str, Any]:

    try:

        summary = service.get_summary(analysis_id)

    except KeyError:

        raise HTTPException(status_code=404, detail="Analysis result not found")

    except Exception as exc:

        raise HTTPException(status_code=500, detail=str(exc))

    if not summary:

        raise HTTPException(status_code=404, detail="Analysis result not found")

    encoded_summary = jsonable_encoder(summary)

    if not isinstance(encoded_summary, dict):

        raise HTTPException(status_code=500, detail="Invalid summary response format")

    return encoded_summary

def _fmt_score(value: Any) -> str:

    try:

        return f"{float(value):.2f}"

    except (TypeError, ValueError):

        return "-"

def _fmt_percent_from_weight(value: Any) -> str:

    try:

        return f"{float(value) * 100:.2f}%"

    except (TypeError, ValueError):

        return "-"

def _excel_value(value: Any) -> Any:

    if value is None:

        return ""

    if isinstance(value, (str, int, float, bool)):

        return value

    return str(value)

def _pdf_escape(value: Any) -> str:

    text = str(value).replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")

    return text

def _build_report_lines(summary: dict[str, Any]) -> list[str]:

    analysis_id = summary.get("analysis_id", "-")

    data_profile = summary.get("data_profile") or {}

    results = summary.get("results") or []

    effective_weights = summary.get("effective_weights") or {}

    excluded_metrics = summary.get("excluded_metrics") or []

    lines = [

        "Data Quality Assessment Report",

        f"Analysis ID: {analysis_id}",

        f"Overall Score: {_fmt_score(summary.get('overall_score'))}",

        f"Weighting Method: {summary.get('weighting_method', '-')}",

        "",

        "Dataset Profile",

        f"Rows: {data_profile.get('row_count', '-')}",

        f"Columns: {data_profile.get('column_count', '-')}",

        f"Missing Total: {data_profile.get('missing_total', '-')}",

        "",

        "Metric Results",

    ]

    if results:

        for result in results:

            metric = result.get("metric", "-")

            score = _fmt_score(result.get("score"))

            status = result.get("status", "-")

            weight = _fmt_percent_from_weight(effective_weights.get(metric))

            lines.append(f"- {metric}: score={score}, status={status}, effective_weight={weight}")

    else:

        lines.append("- No metric results available.")

    lines.extend(["", "Excluded Metrics"])

    if excluded_metrics:

        lines.extend([f"- {metric}" for metric in excluded_metrics])

    else:

        lines.append("- None")

    return lines

def _build_simple_pdf(lines: list[str]) -> bytes:

    # A minimal single-page PDF generator for lightweight report export.

    y = 760

    text_commands = ["BT", "/F1 12 Tf", "14 TL"]

    for index, line in enumerate(lines[:45]):

        safe_line = _pdf_escape(line)

        if index == 0:

            text_commands.append(f"72 {y} Td ({safe_line}) Tj")

        else:

            text_commands.append(f"0 -16 Td ({safe_line}) Tj")

    text_commands.append("ET")

    stream = "\n".join(text_commands).encode("latin-1", errors="replace")

    objects = [

        b"<< /Type /Catalog /Pages 2 0 R >>",

        b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",

        b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",

        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",

        b"<< /Length " + str(len(stream)).encode("ascii") + b" >>\nstream\n" + stream + b"\nendstream",

    ]

    pdf = BytesIO()

    pdf.write(b"%PDF-1.4\n")

    offsets = [0]

    for idx, obj in enumerate(objects, start=1):

        offsets.append(pdf.tell())

        pdf.write(f"{idx} 0 obj\n".encode("ascii"))

        pdf.write(obj)

        pdf.write(b"\nendobj\n")

    xref_offset = pdf.tell()

    pdf.write(f"xref\n0 {len(objects) + 1}\n".encode("ascii"))

    pdf.write(b"0000000000 65535 f \n")

    for offset in offsets[1:]:

        pdf.write(f"{offset:010d} 00000 n \n".encode("ascii"))

    pdf.write(

        f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF".encode("ascii")

    )

    return pdf.getvalue()

def _append_dict_rows(sheet, title: str, data: dict[str, Any]) -> None:

    sheet.append([title])

    sheet.append(["Field", "Value"])

    for key, value in data.items():

        sheet.append([_excel_value(key), _excel_value(value)])

    sheet.append([])

@router.get("")

def list_reports() -> list[dict[str, Any]]:

    try:

        return service.list_reports()

    except Exception as exc:

        raise HTTPException(status_code=500, detail=str(exc))

@router.patch("/{analysis_id}")

def rename_report(analysis_id: str, payload: ReportRenameRequest) -> dict[str, Any]:

    try:

        updated = service.rename_report(analysis_id, payload.report_name)

    except ValueError as exc:

        raise HTTPException(status_code=400, detail=str(exc))

    except Exception as exc:

        raise HTTPException(status_code=500, detail=str(exc))

    if updated is None:

        raise HTTPException(status_code=404, detail="Report not found")

    return updated

@router.delete("/{analysis_id}")

def delete_report(analysis_id: str) -> dict[str, str]:

    try:

        deleted = service.delete_report(analysis_id)

    except Exception as exc:

        raise HTTPException(status_code=500, detail=str(exc))

    if not deleted:

        raise HTTPException(status_code=404, detail="Report not found")

    return {"message": "Report deleted"}

@router.get("/{analysis_id}/pdf")

def download_report_pdf(analysis_id: str) -> Response:

    summary = _get_summary_or_404(analysis_id)

    lines = _build_report_lines(summary)

    pdf_bytes = _build_simple_pdf(lines)

    headers = {

        "Content-Disposition": f'attachment; filename="quality-report-{analysis_id}.pdf"'

    }

    return Response(content=pdf_bytes, media_type="application/pdf", headers=headers)

@router.get("/{analysis_id}/excel")

def download_report_excel(analysis_id: str) -> StreamingResponse:

    try:

        from openpyxl import Workbook

    except ImportError:

        raise HTTPException(status_code=500, detail="openpyxl is required for Excel export")

    summary = _get_summary_or_404(analysis_id)

    data_profile = summary.get("data_profile") or {}

    results = summary.get("results") or []

    metric_weights = summary.get("metric_weights") or {}

    effective_weights = summary.get("effective_weights") or {}

    excluded_metrics = summary.get("excluded_metrics") or []

    workbook = Workbook()

    summary_sheet = workbook.active

    summary_sheet.title = "Summary"

    summary_sheet.append(["Field", "Value"])

    summary_sheet.append(["Analysis ID", _excel_value(summary.get("analysis_id", analysis_id))])

    summary_sheet.append(["Overall Score", _excel_value(summary.get("overall_score"))])

    summary_sheet.append(["Weighting Method", _excel_value(summary.get("weighting_method"))])

    summary_sheet.append(["Excluded Metrics", ", ".join(excluded_metrics) if excluded_metrics else "None"])

    metrics_sheet = workbook.create_sheet("Metric Results")

    metrics_sheet.append(["Metric", "Score", "Status", "Reason", "Configured Weight", "Effective Weight"])

    for result in results:

        metric = result.get("metric", "-")

        metrics_sheet.append([

            _excel_value(metric),

            _excel_value(result.get("score")),

            _excel_value(result.get("status")),

            _excel_value(result.get("reason")),

            _excel_value(metric_weights.get(metric)),

            _excel_value(effective_weights.get(metric)),

        ])

    profile_sheet = workbook.create_sheet("Data Profile")

    _append_dict_rows(profile_sheet, "Data Profile", data_profile)

    dtype_counts = data_profile.get("dtype_counts") or {}

    if dtype_counts:

        profile_sheet.append(["Dtype Counts"])

        profile_sheet.append(["Dtype", "Count"])

        for dtype, count in dtype_counts.items():

            profile_sheet.append([_excel_value(dtype), _excel_value(count)])

        profile_sheet.append([])

    missing_by_column = data_profile.get("missing_by_column") or {}

    if missing_by_column:

        profile_sheet.append(["Missing By Column"])

        profile_sheet.append(["Column", "Missing Count"])

        for column, count in missing_by_column.items():

            profile_sheet.append([_excel_value(column), _excel_value(count)])

    weights_sheet = workbook.create_sheet("Weights")

    weights_sheet.append(["Metric", "Configured Weight", "Effective Weight"])

    all_weight_metrics = sorted(set(metric_weights.keys()) | set(effective_weights.keys()))

    for metric in all_weight_metrics:

        weights_sheet.append([

            _excel_value(metric),

            _excel_value(metric_weights.get(metric)),

            _excel_value(effective_weights.get(metric)),

        ])

    output = BytesIO()

    workbook.save(output)

    output.seek(0)

    headers = {

        "Content-Disposition": f'attachment; filename="quality-report-{analysis_id}.xlsx"'

    }

    return StreamingResponse(

        output,

        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",

        headers=headers,

    )

    """분석 결과 전체 요약"""

    """분석 결과 전체 요약"""

    analysis_id: str

    analysis_id: str

    dataset_id: str

    dataset_id: str

    data_profile: dict[str, Any] = {}

    overall_score: Optional[float] = None

    weighting_method: Optional[str] = None

    metric_weights: dict[str, float] = {}

    effective_weights: dict[str, float] = {}

    excluded_metrics: list[str] = []

    results: list[MetricSummary]

    results: list[MetricSummary]