251013 [ Day 64 ] - Project (18)

시작하며

오늘은 어제에 이어 Oracle Cloud를 이용해 DB를 저장해보려 했는데, Oracle Cloud 회원가입 이슈로 인해 Turso라는 데이터베이스 서버를 사용하기로 했다. 최대 8GB의 저장 용량을 Free Tier로 사용할 수 있고, SQLite의 클라우드 버전이라고 표현해도 될 만큼 비슷해서 사용하기도 편했다.

database

# Turso 데이터베이스 연결 및 초기화 (HTTP API 사용)

import os
import requests
from contextlib import contextmanager
from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

# Turso 데이터베이스 설정
TURSO_DATABASE_URL = os.getenv("TURSO_DATABASE_URL")
TURSO_AUTH_TOKEN = os.getenv("TURSO_AUTH_TOKEN")

# libsql:// URL을 HTTPS URL로 변환
def get_https_url(libsql_url):
    """libsql://host -> https://host"""
    return libsql_url.replace("libsql://", "https://")

# Turso HTTP API 클라이언트
class TursoClient:
    """Turso HTTP API 클라이언트 (SQLite 호환)"""

    def __init__(self, url, auth_token):
        self.base_url = get_https_url(url)
        self.auth_token = auth_token
        self.headers = {
            "Authorization": f"Bearer {auth_token}",
            "Content-Type": "application/json"
        }
        self.last_result = None

    def execute(self, sql, params=None):
        """SQL 실행"""
        # Turso HTTP API 올바른 형식
        payload = {
            "requests": [
                {
                    "type": "execute",
                    "stmt": {
                        "sql": sql
                    }
                }
            ]
        }

        # 파라미터가 있으면 Turso 형식으로 변환하여 추가
        if params:
            # Turso args 형식: [{"type": "...", "value": "..."}]
            # 모든 value는 문자열로 전달 (타입만 지정)
            turso_args = []
            for param in params:
                if param is None:
                    turso_args.append({"type": "null"})
                elif isinstance(param, bool):
                    # bool은 int의 서브클래스이므로 먼저 체크
                    turso_args.append({"type": "integer", "value": str(int(param))})
                elif isinstance(param, int):
                    turso_args.append({"type": "integer", "value": str(param)})
                elif isinstance(param, float):
                    # float는 숫자 타입으로 전달
                    turso_args.append({"type": "float", "value": param})
                else:
                    turso_args.append({"type": "text", "value": str(param)})
            payload["requests"][0]["stmt"]["args"] = turso_args

        try:
            response = requests.post(
                f"{self.base_url}/v2/pipeline",
                headers=self.headers,
                json=payload,
                timeout=10
            )

            if response.status_code != 200:
                raise Exception(f"Turso API error {response.status_code}: {response.text}")

            result = response.json()
            self.last_result = result
            return self

        except requests.exceptions.RequestException as e:
            raise Exception(f"Turso 연결 오류: {str(e)}")

    def fetchone(self):
        """첫 번째 행 가져오기"""
        if not self.last_result:
            return None

        # Turso 응답 형식: {"results": [{"response": {"result": {"rows": [...]}}}]}
        results = self.last_result.get("results", [])
        if not results:
            return None

        response = results[0].get("response", {})
        result = response.get("result", {})
        rows = result.get("rows", [])

        if not rows:
            return None

        # 첫 번째 행 처리
        row = rows[0]

        # Turso는 행을 dict 배열로 반환: [{"type": "text", "value": "foo"}]
        if isinstance(row, list) and len(row) > 0 and isinstance(row[0], dict):
            return tuple(col.get("value") for col in row)
        elif isinstance(row, list):
            return tuple(row)
        else:
            return tuple(row.values())

    def fetchall(self):
        """모든 행 가져오기"""
        if not self.last_result:
            return []

        results = self.last_result.get("results", [])
        if not results:
            return []

        response = results[0].get("response", {})
        result = response.get("result", {})
        rows = result.get("rows", [])

        # 각 행을 튜플로 변환
        processed_rows = []
        for row in rows:
            # Turso는 행을 dict 배열로 반환: [{"type": "text", "value": "foo"}]
            if isinstance(row, list) and len(row) > 0 and isinstance(row[0], dict):
                processed_rows.append(tuple(col.get("value") for col in row))
            elif isinstance(row, list):
                processed_rows.append(tuple(row))
            else:
                processed_rows.append(tuple(row.values()))
        return processed_rows

    @property
    def lastrowid(self):
        """마지막 삽입된 행의 ID"""
        if not self.last_result:
            return None

        results = self.last_result.get("results", [])
        if not results:
            return None

        response = results[0].get("response", {})
        result = response.get("result", {})
        return result.get("last_insert_rowid")

# Turso 클라이언트 생성
def create_turso_client():
    """Turso 클라이언트 생성"""
    if not TURSO_DATABASE_URL or not TURSO_AUTH_TOKEN:
        raise ValueError("TURSO_DATABASE_URL and TURSO_AUTH_TOKEN must be set in .env file")

    print(f"🔗 Turso 연결 (HTTP API): {TURSO_DATABASE_URL}")

    return TursoClient(TURSO_DATABASE_URL, TURSO_AUTH_TOKEN)

def init_db():
    """데이터베이스 초기화 및 테이블 생성"""
    client = create_turso_client()

    # 분석 기록 테이블
    client.execute("""
        CREATE TABLE IF NOT EXISTS analysis_records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            total_items INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # 탐지된 항목 테이블
    client.execute("""
        CREATE TABLE IF NOT EXISTS detected_items (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            analysis_id INTEGER NOT NULL,
            category TEXT NOT NULL,
            confidence REAL NOT NULL,
            FOREIGN KEY (analysis_id) REFERENCES analysis_records(id)
        )
    """)

    # 피드백 테이블 (잘못된 분류 보고)
    client.execute("""
        CREATE TABLE IF NOT EXISTS feedback (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            predicted_class TEXT NOT NULL,
            actual_class TEXT NOT NULL,
            confidence REAL NOT NULL,
            timestamp TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    print(f"✅ Turso 데이터베이스 초기화 완료: {TURSO_DATABASE_URL}")

@contextmanager
def get_db():
    """데이터베이스 연결 컨텍스트 매니저"""
    client = create_turso_client()
    try:
        yield client
    finally:
        # libsql-client는 자동으로 연결을 관리하므로 명시적 close 불필요
        pass

dummy data

"""
더미 통계 데이터 생성 스크립트
2개월 동안의 분석 데이터를 생성하여 Turso 데이터베이스에 삽입
"""

import sys
import os
import random
from datetime import datetime, timedelta

# app 모듈을 import할 수 있도록 경로 추가
sys.path.append(os.path.dirname(__file__))
from app.database import create_turso_client

# 카테고리 및 각 카테고리별 생성 개수 (100~200 랜덤)
CATEGORIES = {
    "캔": random.randint(100, 200),
    "유리": random.randint(100, 200),
    "종이": random.randint(100, 200),
    "플라스틱": random.randint(100, 200),
    "스티로폼": random.randint(100, 200),
    "비닐": random.randint(100, 200),
}

# 2개월 전부터 오늘까지
START_DATE = datetime.now() - timedelta(days=60)
END_DATE = datetime.now()

def random_datetime(start, end):
    """시작일과 종료일 사이의 랜덤한 날짜/시간 생성"""
    delta = end - start
    random_seconds = random.randint(0, int(delta.total_seconds()))
    return start + timedelta(seconds=random_seconds)

def random_confidence():
    """0.75~0.99 사이의 랜덤한 신뢰도 생성"""
    return round(random.uniform(0.75, 0.99), 2)

def generate_dummy_data():
    """더미 데이터 생성 및 Turso DB 삽입"""

    # Turso 클라이언트 생성
    client = create_turso_client()

    print("🚀 더미 데이터 생성 시작...")
    print(f"📅 기간: {START_DATE.strftime('%Y-%m-%d')} ~ {END_DATE.strftime('%Y-%m-%d')}")
    print(f"📊 카테고리별 개수:")
    for category, count in CATEGORIES.items():
        print(f"   - {category}: {count}개")

    total_items = sum(CATEGORIES.values())
    print(f"✅ 총 {total_items}개의 탐지 항목 생성 예정\n")

    # 모든 탐지 항목을 담을 리스트
    all_detections = []

    # 각 카테고리별로 탐지 항목 생성
    for category, count in CATEGORIES.items():
        for _ in range(count):
            timestamp = random_datetime(START_DATE, END_DATE)
            confidence = random_confidence()
            all_detections.append((timestamp, category, confidence))

    # 시간순으로 정렬
    all_detections.sort(key=lambda x: x[0])

    # 분석 세션 생성 (3~5개의 탐지를 하나의 분석으로 그룹화)
    analysis_id = 1
    i = 0

    while i < len(all_detections):
        # 이번 분석 세션에 포함될 탐지 개수 (1~5개 랜덤)
        detections_in_session = random.randint(1, 5)
        session_timestamp = all_detections[i][0]

        # analysis_records 테이블에 분석 세션 삽입
        client.execute(
            "INSERT INTO analysis_records (timestamp, total_items, created_at) VALUES (?, ?, ?)",
            [session_timestamp.strftime('%Y-%m-%d %H:%M:%S'), detections_in_session, session_timestamp.strftime('%Y-%m-%d %H:%M:%S')]
        )

        # 이번 세션의 탐지들을 detected_items 테이블에 삽입
        for j in range(detections_in_session):
            if i + j >= len(all_detections):
                break

            _, category, confidence = all_detections[i + j]
            client.execute(
                "INSERT INTO detected_items (analysis_id, category, confidence) VALUES (?, ?, ?)",
                [analysis_id, category, confidence]
            )

        i += detections_in_session
        analysis_id += 1

    # 결과 확인
    client.execute("SELECT COUNT(*) FROM analysis_records")
    total_analyses = client.fetchone()[0]

    client.execute("SELECT COUNT(*) FROM detected_items")
    total_detections = client.fetchone()[0]

    client.execute("""
        SELECT category, COUNT(*)
        FROM detected_items
        GROUP BY category
        ORDER BY COUNT(*) DESC
    """)
    category_stats = client.fetchall()

    print("=" * 50)
    print("✨ 더미 데이터 생성 완료!")
    print("=" * 50)
    print(f"📌 총 분석 세션: {total_analyses}개")
    print(f"📌 총 탐지 항목: {total_detections}개")
    print(f"\n📊 카테고리별 통계:")
    for category, count in category_stats:
        print(f"   - {category}: {count}개")
    print("\n✅ Turso 데이터베이스에 데이터가 저장되었습니다.")
    print("🌐 통계 페이지(stats.html)를 열어서 확인하세요!\n")

if __name__ == "__main__":
    generate_dummy_data()

마치며

더미 데이터를 랜덤으로 생성해 Turso 서버에 저장한 뒤, 통계 탭과 소개 탭을 만들었다. 통계 탭에는 카테고리별 통계와 기간별 통계를 포함했으며, 분류 결과에 피드백 버튼을 추가해 오분류에 대한 피드백을 저장하고, 이를 기반으로 오분류 차트를 추가할 예정이다.
이제 내일 테스트 배포를 해보고, 최종적으로 수정을 한 다음 PPT와 발표 준비만 하면 이번 프로젝트도 끝이 난다.