feat: add web dashboard workflow and report history

.venv/

venv/

.DS_Store

# macOS

.DS_Store

# Python

__pycache__/

*.py[cod]

# Local database

*.db

# Office temporary files

~$*

from __future__ import annotations

import os

from collections.abc import Generator

from sqlalchemy import create_engine

from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./etsi_dq_history.db")

connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}

engine = create_engine(

    DATABASE_URL,

    connect_args=connect_args,

)

SessionLocal = sessionmaker(

    autocommit=False,

    autoflush=False,

    bind=engine,

)

Base = declarative_base()

def get_db() -> Generator:

    db = SessionLocal()

    try:

        yield db

    finally:

        db.close()

def init_db() -> None:

    # Import models here so SQLAlchemy registers them before create_all().

    from api.model.analysis_history import AnalysisHistory  # noqa: F401

    Base.metadata.create_all(bind=engine)

 No newline at end of file

"""

ETSI DQ - FastAPI Backend

=========================

프로젝트 구조:

etsi_dq/                  ← 기존 코드 (그대로 유지)

├── metrics/

│   ├── completeness.py

│   ├── accuracy.py

│   ├── consistency.py

│   └── ...

├── pipeline.py

├── schemas.py

├── profiler.py

├── io.py

└── ...

api/                      ← 새로 추가하는 백엔드 계층

├── __init__.py

├── main.py               ← FastAPI 앱 진입점 (이 파일)

├── routers/

│   ├── __init__.py

│   ├── datasets.py       ← 데이터셋 업로드/조회 API

│   ├── metrics.py        ← 지표 설정/pre-check API

│   ├── analysis.py       ← 분석 실행 API

│   └── results.py        ← 결과 조회 API

├── schemas/

│   ├── __init__.py

│   ├── datasets.py       ← 데이터셋 관련 요청/응답 모델

│   ├── metrics.py        ← 지표 설정 관련 모델

│   ├── analysis.py       ← 분석 요청/응답 모델

│   └── results.py        ← 결과 응답 모델

└── services/

    ├── __init__.py

    ├── dataset_service.py  ← 기존 io.py 함수를 감싸는 서비스

    ├── metric_service.py   ← 기존 precheck/설정 로직 감싸는 서비스

    └── analysis_service.py ← 기존 pipeline.py의 run_analysis 감싸는 서비스

실행 방법:

    uvicorn api.main:app --reload --port 8000

API 문서 자동 생성:

    http://localhost:8000/docs  (Swagger UI)

    http://localhost:8000/redoc (ReDoc)

"""

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware

from .database import init_db

from .routers import datasets, metrics, analysis, results, reports

app = FastAPI(

    title="ETSI Data Quality API",

    description="ETSI 데이터 품질 평가 시스템 백엔드 API",

    version="0.1.0",

)

@app.on_event("startup")

def on_startup():

    init_db()

# CORS 설정 - Streamlit(8501) 또는 React(3000)에서 접근 허용

app.add_middleware(

    CORSMiddleware,

    allow_origins=[

        "http://localhost:8501",   # Streamlit 기본 포트

        "http://localhost:3000",   # React 개발 서버 포트

        "http://localhost:5173",   # Vite 개발 서버 포트

    ],

    allow_credentials=True,

    allow_methods=["*"],

    allow_headers=["*"],

)

# 라우터 등록

app.include_router(datasets.router, prefix="/api/datasets", tags=["datasets"])

app.include_router(metrics.router,  prefix="/api/metrics",  tags=["metrics"])

app.include_router(analysis.router, prefix="/api/analysis", tags=["analysis"])

app.include_router(results.router,  prefix="/api/results",  tags=["results"])

app.include_router(reports.router,  prefix="/api/reports",  tags=["reports"])

@app.get("/")

def root():

    return {"message": "ETSI Data Quality API", "version": "0.1.0"}

@app.get("/health")

def health_check():

    return {"status": "ok"}

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