SQLConnect.py

import os
import sqlite3
from typing import Optional, List, Dict, Any, Tuple
from contextlib import closing
import json, logging

### DB 저장 경로 지정
DB_PATH = "data/tts.db"

### Read, Debug

def showModels(): #DEBUG용
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    with closing (db_connect.cursor()) as db_cursor:
        db_cursor.execute("""
            SELECT id, model_name, voice_code, status, created_at
            FROM tts_models
            ORDER BY id DESC
                          """)
        rows = db_cursor.fetchall()
        return [dict(r) for r in rows]
    
def showMatch(): #DEBUG용
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    with closing (db_connect.cursor()) as db_cursor:
        db_cursor.execute("""
            SELECT story_id, model_name, user_id, voice_match, created_at, updated_at
            FROM tts_match
                          """)
        rows = db_cursor.fetchall()
        return [dict(r) for r in rows]
    
def getVoiceCode(model_name:str) -> str:
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    with closing (db_connect.cursor()) as db_cursor:
        db_cursor.execute(
                "SELECT voice_code FROM tts_models WHERE model_name = ? LIMIT 1",
                (model_name,),
            )
        row = db_cursor.fetchone()
        return row["voice_code"] if row else None
    
def getStoryMatch(story_id:str, model_name:str, user_id:str) -> Optional[Dict[str, Any]]:
    if not story_id or not model_name or not user_id:
        raise ValueError("story_id, model_name, user_id가 필요합니다.")
    
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    try:
        with closing(db_connect.cursor()) as db_cursor:
            db_cursor.execute("""
            SELECT voice_match
                FROM tts_match
                WHERE story_id=? AND model_name=? AND user_id=?
                LIMIT 1
            """, (story_id, model_name, user_id))

            row = db_cursor.fetchone()
            if not row:
                return None
            return json.loads(row["voice_match"])
    except Exception:
        logging.exception("DB 조회 실패")
        return None


def checkModel(model_name:str): #검색된 모델이 없을시 None 반환 있을시 Status 반환
    if not model_name:
        raise ValueError("model_name이 비어 있습니다.")
    
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    with closing(db_connect.cursor()) as db_cursor:
        db_cursor.execute(
            "SELECT status FROM tts_models WHERE model_name = ? LIMIT 1",
            (model_name,),
        )
        row = db_cursor.fetchone()
    db_connect.close()

    if row:
        return row["status"]
    else:
        return None
        
### DB 기본 설정
def defaultSetting():
    os.makedirs("data", exist_ok=True)
    db_connect = sqlite3.connect(DB_PATH)
    db_connect.row_factory = sqlite3.Row
    with closing (db_connect.cursor()) as db_cursor:
        db_cursor.execute("""
            SELECT voice_code
            FROM tts_models
            ORDER BY id DESC
                          """)
        rows = db_cursor.fetchall()
        return [dict(r) for r in rows]

    db_cursor.executescript("""
        CREATE TABLE IF NOT EXISTS tts_models (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            model_name TEXT NOT NULL UNIQUE,
            voice_code TEXT NOT NULL,
            status TEXT NOT NULL CHECK(status IN ('Done','Process','Fail')),
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS tts_match (
            story_id    TEXT NOT NULL,
            model_name  TEXT NOT NULL,
            user_id     TEXT NOT NULL,
            voice_match TEXT NOT NULL CHECK (json_valid(voice_match)),
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (story_id, model_name, user_id)
        );

        CREATE TRIGGER IF NOT EXISTS trg_tts_match_updated_at
        AFTER UPDATE ON tts_match
        FOR EACH ROW
        BEGIN
            UPDATE tts_match SET updated_at = CURRENT_TIMESTAMP
            WHERE story_id=OLD.story_id AND model_name=OLD.model_name AND user_id=OLD.user_id;
        END;
        """)
    db_connect.commit()
    db_connect.close()



### MODEL CREATE , UPDATE

def createModel(model_name:str, voice_code:str, status:str): #모델 생성, 존재할시 status 변경
    if status not in ("Done", "Process", "Fail"):
        raise ValueError("status 상태가 부적절합니다.")
    with sqlite3.connect(DB_PATH) as db_connect:
        db_connect.execute(
            """
            INSERT INTO tts_models(model_name, voice_code, status) VALUES(?,?,?)
            ON CONFLICT (model_name)
            DO UPDATE SET
                voice_code = excluded.voice_code,
                status = excluded.status
            """,
            (model_name, voice_code, status)
        )

def debug_createModel(model_name:str, voice_code:str, status:str):
    with sqlite3.connect(DB_PATH) as db_connect:
        db_connect.execute(
            """
            INSERT INTO tts_models(model_name, voice_code, status) VALUES(?,?,?)
            ON CONFLICT (model_name)
            DO UPDATE SET
                voice_code = excluded.voice_code,
                status = excluded.status
            """, (model_name, voice_code, status)
        )

### VOICE CREATE
def createMatch(story_id:str, model_name:str, user_id:str, voice_match:dict) -> bool:
    payload = json.dumps(voice_match, ensure_ascii=False)

    try:
        with sqlite3.connect(DB_PATH) as db_connect:
            db_connect.execute(
            """
            INSERT INTO tts_match(story_id, model_name, user_id, voice_match) VALUES(?,?,?,?)                    
            ON CONFLICT (story_id, model_name, user_id)
            DO UPDATE SET
                voice_match = excluded.voice_match,
               updated_at  = CURRENT_TIMESTAMP
            """, (story_id, model_name, user_id, payload)
            )
            return True
    except Exception as e:
        return False
    

def debug_createMatch(story_id:str, model_name:str, user_id:str, voice_math:dict):

    payload = json.dumps(voice_math, ensure_ascii=False)

    with sqlite3.connect(DB_PATH) as db_connect:
        db_connect.execute(
            """
            INSERT INTO tts_match(story_id, model_name, user_id, voice_match) VALUES(?,?,?,?)
            ON CONFLICT (story_id, model_name, user_id)
            DO UPDATE SET
                voice_match = excluded.voice_match,
                updated_at  = CURRENT_TIMESTAMP
            """,
            (story_id, model_name, user_id, payload)
        )

 

###  MODEL DELETE

def deleteModel(model_name) -> bool:
    if not model_name:
        raise ValueError("model_name이 비어 있습니다.")

    with sqlite3.connect(DB_PATH) as db_connect:
        db_cursor = db_connect.execute(
            "DELETE FROM tts_models WHERE model_name =?",
            (model_name,),
        )
        return db_cursor.rowcount > 0