chore: remove legacy entry scripts

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

"""분석 실행 API 엔드포인트"""

import math

import traceback

import numpy as np

import pandas as pd

from fastapi import APIRouter, HTTPException

from api.schemas.analysis import AnalysisRunRequest, AnalysisRunResponse

from fastapi.responses import JSONResponse

from api.schemas.analysis import AnalysisRunRequest

from api.services import analysis_service

router = APIRouter()

service = analysis_service

@router.post("/run", response_model=AnalysisRunResponse)

def make_json_safe(value):

    if hasattr(value, "model_dump"):

        return make_json_safe(value.model_dump())

    if isinstance(value, pd.DataFrame):

        return value.replace({np.nan: None}).to_dict(orient="records")

    if isinstance(value, pd.Series):

        return value.replace({np.nan: None}).to_dict()

    if isinstance(value, dict):

        return {key: make_json_safe(item) for key, item in value.items()}

    if isinstance(value, list):

        return [make_json_safe(item) for item in value]

    if isinstance(value, tuple):

        return [make_json_safe(item) for item in value]

    if isinstance(value, np.integer):

        return int(value)

    if isinstance(value, np.floating):

        number = float(value)

        if math.isnan(number):

            return None

        return number

    if isinstance(value, np.ndarray):

        return make_json_safe(value.tolist())

    return value

@router.post("/run")

async def run_analysis(request: AnalysisRunRequest):

    """

    품질 분석 실행

    """

    try:

        result = await service.execute(request)

        return result

        return JSONResponse(content=make_json_safe(result.model_dump()))

    except Exception as e:

        traceback.print_exc()

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

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,

    )

"""결과 조회 API 엔드포인트"""

import math

import numpy as np

import pandas as pd

from fastapi import APIRouter, HTTPException

from api.schemas.analysis import AnalysisSummaryResponse, MetricDetailResponse

from fastapi.responses import JSONResponse

from api.schemas.analysis import AnalysisSummaryResponse

from api.services import analysis_service

router = APIRouter()

service = analysis_service

def make_json_safe(value):

    if hasattr(value, "model_dump"):

        return make_json_safe(value.model_dump())

    if isinstance(value, pd.DataFrame):

        return value.replace({np.nan: None}).to_dict(orient="records")

    if isinstance(value, pd.Series):

        return value.replace({np.nan: None}).to_dict()

    if isinstance(value, dict):

        return {key: make_json_safe(item) for key, item in value.items()}

    if isinstance(value, list):

        return [make_json_safe(item) for item in value]

    if isinstance(value, tuple):

        return [make_json_safe(item) for item in value]

    if isinstance(value, np.integer):

        return int(value)

    if isinstance(value, np.floating):

        number = float(value)

        if math.isnan(number):

            return None

        return number

    if isinstance(value, np.ndarray):

        return make_json_safe(value.tolist())

    try:

        if pd.isna(value):

            return None

    except (TypeError, ValueError):

        pass

    return value

@router.get("/{analysis_id}/summary", response_model=AnalysisSummaryResponse)

async def get_analysis_summary(analysis_id: str):

    """

    return result

@router.get("/{analysis_id}/details/{metric}", response_model=MetricDetailResponse)

@router.get("/{analysis_id}/details/{metric}")

async def get_metric_detail(analysis_id: str, metric: str):

    """

    개별 지표 상세 결과 조회

    result = service.get_metric_detail(analysis_id, metric)

    if not result:

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

    return result

    return JSONResponse(content=make_json_safe(result))

@router.get("/{analysis_id}/rules/{metric}")

    result = service.get_rule_results(analysis_id, metric)

    if not result:

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

    return result

    return JSONResponse(content=make_json_safe(result))

@router.get("/{analysis_id}/samples/{metric}")

    result = service.get_samples(analysis_id, metric, limit)

    if not result:

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

    return result

    return JSONResponse(content=make_json_safe(result))