services/cache_service.py - master - mishzorikhin/VoiceDecodeAPI

services/cache_service.py
102 строки4 KB
mish
first commit
04 янв 2025, 18:25
04 янв 2025, 18:254327bcd
import json
import logging
import sqlite3
from typing import Any, Optional

# Путь к базе данных SQLite
DB_PATH = "cache.db"

# Настройка логгера
logger = logging.getLogger(__name__)


def init_db(db_path: str = DB_PATH) -> None:
    """
    Инициализирует базу данных, создавая таблицу кэша, если она не существует.

    :param db_path: Путь к файлу базы данных SQLite.
    """
    logger.info("Инициализация базы данных SQLite.")
    try:
        with sqlite3.connect(db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS cache (
                    file_hash TEXT PRIMARY KEY,
                    result TEXT NOT NULL
                )
            """)
            conn.commit()
        logger.info("База данных успешно инициализирована.")
    except sqlite3.Error as e:
        logger.error(f"Ошибка при инициализации базы данных: {e}")
        raise


def get_from_cache(file_hash: str, db_path: str = DB_PATH) -> Optional[Any]:
    """
    Получает данные из кэша по хешу файла.

    :param file_hash: SHA-256 хеш файла.
    :param db_path: Путь к файлу базы данных SQLite.
    :return: Распарсенный результат из кэша или None, если запись не найдена.
    """
    logger.info(f"Попытка получить данные из кэша для хеша: {file_hash}")
    try:
        with sqlite3.connect(db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT result FROM cache WHERE file_hash = ?", (file_hash,))
            row = cursor.fetchone()
            if row:
                result = json.loads(row[0])
                logger.debug(f"Данные успешно получены из кэша для хеша: {file_hash}")
                return result
            else:
                logger.info(f"Нет данных в кэше для хеша: {file_hash}")
                return None
    except sqlite3.Error as e:
        logger.error(f"Ошибка при получении данных из кэша: {e}")
        return None
    except json.JSONDecodeError as e:
        logger.error(f"Ошибка при декодировании JSON из кэша: {e}")
        return None


def save_to_cache(file_hash: str, result: Any, db_path: str = DB_PATH) -> None:
    """
    Сохраняет данные в кэш с привязкой к хешу файла.

    :param file_hash: SHA-256 хеш файла.
    :param result: Результат транскрипции или другие данные для кэширования.
    :param db_path: Путь к файлу базы данных SQLite.
    """
    logger.info(f"Сохранение данных в кэш для хеша: {file_hash}")
    try:
        with sqlite3.connect(db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT OR REPLACE INTO cache (file_hash, result) 
                VALUES (?, ?)
            """, (file_hash, json.dumps(result)))
            conn.commit()
        logger.debug(f"Данные успешно сохранены в кэш для хеша: {file_hash}")
    except sqlite3.Error as e:
        logger.error(f"Ошибка при сохранении данных в кэш: {e}")
        raise
    except TypeError as e:
        logger.error(f"Ошибка при сериализации данных для кэша: {e}")
        raise

def clear_cache():
    """
    Удаляет все записи из таблицы кэша.
    """
    try:
        with sqlite3.connect(DB_PATH) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM cache")
            conn.commit()
            logger.info("Кэш успешно очищен.")
    except sqlite3.Error as e:
        logger.error(f"Ошибка при очистке кэша: {e}")
        raise