sources/uDB.py - master - VolcanoByte/vbCinema

sources/uDB.py
464 строки17 KB
Eugene Gavrilyuk
Update sync
09 июл 2026, 14:09
09 июл 2026, 14:0954058c1
"""
Управление базой данных.
"""

import datetime as dt
import os
import sqlite3

from unicodedata import normalize, combining

from PyQt6.QtSql import QSqlDatabase, QSqlQuery

from uMovie import Movie
from uConsts import DEFAULT_DB_FULL_NAME, StoreConsts
from uGlobal import store

from uSQL import (SQL_MOVIE_TAG_SELECT, SQL_MOVIE_TAG_SELECT_NAMES,
                  SQL_MOVIE_TAG_DELETE, SQL_MOVIE_TAG_SELECT_TAGS,
                  SQL_MOVIE_TAG_INSERT, SQL_MOVIE_TAG_DELETE_ALL_TAGS,
                  SQL_MOVIE_TAG_SELECT_MOVIES, SQL_TAG_SELECT_BY_ID,
                  SQL_TAG_SELECT_BY_NAME, SQL_TAG_ADD, SQL_MOVIE_SELECT_ALL,
                  SQL_TAG_CLEAR_COUNT, SQL_TAG_CALC_COUNT,
                  SQL_TAG_DELETE_BY_NAME, SQL_TAG_DELETE_BY_ID,
                  SQL_TAG_DELETE_EMPTY, SQL_CREATE_TABLE_TAGS,
                  SQL_CREATE_TABLE_MOVIE_TAG, SQL_CREATE_TABLE_MOVIES,
                  SQL_MOVIE_SELECT, SQL_MOVIE_DELETE, SQL_MOVIE_WATCHED_AT,
                  SQL_TAG_SET_COUNT_UPPER_NAME, SQL_GET_REC_COUNT,
                  SQL_PROPERTIES_GET, SQL_CREATE_TABLE_PROPERTIES,
                  SQL_PROPERTIES_SET, SQL_PROPERTIES_ADD_ROW,
                  TABLE_MOVIES_FIELD, SQL_MOVIE_SET_WANTED,
                  SQL_MOVIE_SET_WATCHED)


def getNow() -> float:
    return dt.datetime.now().timestamp()


class DB:
    """Класс для управления базой фильмов"""

    def __init__(self):
        self.filename = None
        self.syncId = None
        self.native = None
        self.conn = QSqlDatabase.addDatabase('QSQLITE')
        self.query = QSqlQuery()

    @staticmethod
    def upper_cyr(text):
        if text:
            return ''.join(
                normalize('NFKC', char.upper())
                if combining(char) else char.upper()
                for char in text)
        else:
            return ''

    @staticmethod
    def getFieldIndexes(query: QSqlQuery) -> dict:
        """Возвращает словарь имя поля -> индекс поля"""
        fields = dict()
        for i in range(query.record().count()):
            fields[query.record().field(i).name()] = i

        return fields

    @staticmethod
    def getQuerySet(query: QSqlQuery, fieldNo: int) -> set:
        """Функция нужна для устранения дублирования"""
        _res = set()
        if query.first():
            while query.isValid():
                _res.add(query.record().field(fieldNo).value())
                query.next()

        return _res

    @staticmethod
    def queryMoviesTags(movieIds: set):
        """Возвращает запрос с выбором тегов указанных фильмов"""
        _movieIds = ', '.join(map(str, movieIds))
        return SQL_MOVIE_TAG_SELECT_TAGS % _movieIds

    @staticmethod
    def queryAllMovies():
        """Возвращает запрос с выбором всех фильмов"""
        _query = SQL_MOVIE_SELECT_ALL % TABLE_MOVIES_FIELD
        return _query

    @staticmethod
    def queryTagsMovies(tagIds: set):
        """Возвращает запрос с выбором фильмов с указанными тегами"""
        _tagIds = ', '.join(map(str, tagIds))
        _query = SQL_MOVIE_TAG_SELECT_MOVIES % (
        TABLE_MOVIES_FIELD, _tagIds, len(tagIds))
        return _query

    def getRecCount(self, table: str) -> int:
        """Возвращает количество записей в указанной таблице"""
        self.query.exec(SQL_GET_REC_COUNT % table)
        self.query.first()
        return self.query.record().field(0).value()

    def open(self, filename: str):
        self.filename = filename

        self.native = sqlite3.connect(self.filename)
        self.native.create_function("UPPER_CYR", 1, self.upper_cyr)

        self.conn.setDatabaseName(self.filename)
        if not self.conn.open():
            print(self.conn.lastError())
            exit(1)
        self.conn.exec('PRAGMA foreign_keys = ON')

        self.createTables()
        self.preparePropertiesTable()
        self.prepareTags()

    def reopen(self):
        self.native = sqlite3.connect(self.filename)
        # todo: Возможно следующая строка излишняя
        self.native.create_function("UPPER_CYR", 1, self.upper_cyr)

        self.open(self.filename)

    def close(self):
        """Закрытие базы"""
        self.native.close()
        self.conn.close()

    def createTables(self):
        """Создание таблиц"""
        self.query.exec(SQL_CREATE_TABLE_MOVIES)
        self.query.exec(SQL_CREATE_TABLE_TAGS)
        self.query.exec(SQL_CREATE_TABLE_MOVIE_TAG)
        self.query.exec(SQL_CREATE_TABLE_PROPERTIES)

    def preparePropertiesTable(self):
        """Если база новая, то добавляем в таблицу свойств
        tblProperties идентификатор для синхронизации базы"""
        if self.getRecCount('tblProperties') == 0:
            name = os.path.splitext(os.path.basename(self.filename))
            self.query.prepare(SQL_PROPERTIES_ADD_ROW)
            self.query.bindValue(':syncId', name[0])
            self.query.exec()

        self.syncId = self.getSyncId()
        self.setTimestamp()

    def prepareTags(self):
        """Подготовка тегов: подсчёт количества, добавление имени в верхнем
        регистре, удаление пустых тегов."""
        if self.getRecCount('tblTags'):
            self.conn.transaction()
            try:
                #  Обнуляем количества тегов
                self.query.exec(SQL_TAG_CLEAR_COUNT)

                #  Собираем количества
                tags = dict()
                tagName, tagCount = range(2)
                self.query.exec(SQL_TAG_CALC_COUNT)
                self.query.first()
                while self.query.isValid():
                    tags[self.query.value(tagName)] = (
                        self.query.value(tagCount))

                    self.query.next()

                #  Заносим собранные количества и имя в верхнем регистре
                self.query.prepare(SQL_TAG_SET_COUNT_UPPER_NAME)
                for tagName, tagCount in tags.items():
                    self.query.bindValue(':fldTagCount', tagCount)
                    self.query.bindValue(':fldTagName', tagName)
                    self.query.bindValue(':fldTagUpperName', tagName.upper())
                    self.query.exec()

                # Удаляем пустые теги
                self.query.exec(SQL_TAG_DELETE_EMPTY)
            finally:
                self.conn.commit()

    def setProperty(self, name: str, value: str):
        """Установка свойства"""
        self.query.exec(SQL_PROPERTIES_SET % (name, value))

    def property(self, propertyName: str) -> str:
        """Возвращает свойство"""
        self.query.exec(SQL_PROPERTIES_GET % propertyName)
        self.query.first()
        if self.query.isValid():
            return self.query.record().field(0).value()
        else:
            return ''

    def setTimestamp(self):
        """Установка метки времени изменения базы данных"""
        self.setProperty('fldPropertyTimestamp', getNow())

    def timestamp(self) -> dt.datetime | None:
        """Получение метки времени изменения базы данных"""
        return self.property('fldPropertyTimestamp')

    def setSyncId(self, syncId: str):
        """Установка идентификатора синхронизации базы"""
        self.setProperty('fldPropertySyncId', syncId)
        self.syncId = syncId

    def getSyncId(self) -> str:
        """Получение идентификатора синхронизации базы"""
        return self.property('fldPropertySyncId')

    def addMovie(self, movie: Movie) -> int:
        """Добавление фильма в базу."""
        if "fldMovieAddedAt" not in movie.data:
            movie.data["fldMovieAddedAt"] = getNow()

        _sql = "INSERT INTO tblMovies ({:s}) VALUES ({:s})".format(
            ', '.join(f'{key}' for key in movie.data.keys()),
            ', '.join(f':{key}' for key in movie.data.keys()))
        self.query.prepare(_sql)
        for key, value in movie.data.items():
            self.query.bindValue(f':{key}', value)
        if not self.query.exec():
            print(self.query.lastError().text())

        movieId = self.query.lastInsertId()

        # Добавляем теги
        self.addMovieTags(movieId, movie.tags)

        self.setTimestamp()

        return movieId

    def updateMovie(self, movieId: int, movie: Movie):
        """Обновление фильма"""

        #  Подготавливаем запрос, для последующего связывания:
        _sql = "UPDATE tblMovies SET {:s} WHERE fldMovieID = {:n}".format(
            ', '.join(f'{key} = :{key}' for key in
                      movie.data.keys()), movieId)

        self.query.prepare(_sql)
        for key, value in movie.data.items():
            self.query.bindValue(f':{key}', value)
        if not self.query.exec():
            print(self.query.lastError().text())

        # Обновляем теги
        self.clearMovieTags(movieId)
        self.addMovieTags(movieId, movie.tags)

        self.setTimestamp()

    def setMovieWatchedNow(self, movieId: int):
        self.query.prepare(SQL_MOVIE_WATCHED_AT)
        self.query.bindValue(':movieId', movieId)
        self.query.bindValue(':watchedAt', getNow())
        self.query.exec()

        # self.setTimestamp()

    def setWanted(self, movieId: int, isWanted: bool):
        self.query.prepare(SQL_MOVIE_SET_WANTED)
        self.query.bindValue(':movieId', movieId)
        self.query.bindValue(':wanted', isWanted)
        self.query.exec()

        self.setTimestamp()

    def setWatched(self, movieId: int, isWatched: bool):
        self.query.prepare(SQL_MOVIE_SET_WATCHED)
        self.query.bindValue(':movieId', movieId)
        self.query.bindValue(':watched', isWatched)
        self.query.exec()

        self.setTimestamp()

    def getMovie(self, movieId) -> Movie | None:
        """Возвращает данные фильма с указанным id
        или None если таковой не найден"""
        if isinstance(movieId, str):
            _id = int(movieId)
        else:
            _id = movieId
        self.query.prepare(SQL_MOVIE_SELECT)
        self.query.bindValue(':movieId', _id)
        self.query.exec()

        if self.query.first():
            movie = Movie()

            # Читаем данные
            for n in range(self.query.record().count()):
                _field = self.query.record().field(n).name()
                _value = self.query.record().field(n).value()
                movie.data[_field] = _value

            # Читаем теги
            movie.tags = self.getMovieTagNames(movieId)

            return movie

    def deleteMovie(self, movieId: int):
        """Удаляет указанный фильм"""
        self.query.prepare(SQL_MOVIE_DELETE)
        self.query.bindValue(':movieId', movieId)
        self.query.exec()

        self.prepareTags()

        self.setTimestamp()

    def addTag(self, tagName: str) -> int:
        """Если тег новый, то добавляет. В любом случае возвращает id тега"""
        if _res := self.getTagId(tagName):
            return _res
        else:
            self.setTimestamp()

            self.query.prepare(SQL_TAG_ADD)
            self.query.bindValue(':name', tagName)
            self.query.bindValue(':upperName', tagName.upper())

            if not self.query.exec():
                print(self.query.lastError().text())

            self.setTimestamp()

            return self.query.lastInsertId()

    def addTags(self, tags: set) -> set:
        """Добавляет теги в базу, если новые.
        В любом случае возвращает множество id тегов"""
        tagIds = set()
        for tag in tags:
            tagIds.add(self.addTag(tag))
        return tagIds

    def getTagName(self, tagId: int) -> str:
        """Возвращает тег по id"""
        self.query.prepare(SQL_TAG_SELECT_BY_ID)
        self.query.bindValue(':id', tagId)
        self.query.exec()

        return self.query.record().field(0).value() if self.query.first() \
            else None

    def getTagId(self, tagName: str) -> int:
        """Возвращает id тега"""
        self.query.prepare(SQL_TAG_SELECT_BY_NAME)
        self.query.bindValue(':upperName', tagName.upper())
        self.query.exec()

        return self.query.record().field(0).value() if self.query.first() \
            else None

    def deleteTag(self, tag: str | int):
        """Удаляет тег (по имени или id)"""
        if isinstance(tag, str):
            self.query.prepare(SQL_TAG_DELETE_BY_NAME)
            self.query.bindValue(':upperName', tag.upper())
        else:
            self.query.prepare(SQL_TAG_DELETE_BY_ID)
            self.query.bindValue(':id', tag)
        self.query.exec()

        self.setTimestamp()

    def addMovieTag(self, movieId: int, tag: str | int):
        """Добавляет тег к фильму, если такого тега еще нет"""
        # Проверяем наличие фильма
        if not self.getMovie(movieId):
            return

        # Проверяем, не пустой ли тег
        if not tag:
            return

        if isinstance(tag, str):
            tagId = self.addTag(tag)
        else:
            #  Если передан id тега, то проверяем его наличие
            if self.getTagName(tag):
                tagId = tag
            else:
                return

        self.query.prepare(SQL_MOVIE_TAG_SELECT)
        self.query.bindValue(':movieId', movieId)
        self.query.bindValue(':tagId', tagId)
        self.query.exec()

        if not self.query.first():
            self.query.prepare(SQL_MOVIE_TAG_INSERT)
            self.query.bindValue(':movieId', movieId)
            self.query.bindValue(':tagId', tagId)
            self.query.exec()

            self.setTimestamp()

    def addMovieTags(self, movieId: int, tags: set):
        """Добавляет теги к заметке"""
        self.conn.transaction()
        try:
            for tag in tags:
                self.addMovieTag(movieId, tag)
        finally:
            self.conn.commit()

    def getMovieTagNames(self, movieId: int) -> set:
        """Возвращает набор имен тегов заметки"""
        self.query.prepare(SQL_MOVIE_TAG_SELECT_NAMES)
        self.query.bindValue(':movieId', movieId)
        if self.query.exec():
            return self.getQuerySet(self.query, 0)

    def deleteMovieTag(self, movieId: int, tag):
        """Удаляет тег у заметки"""
        if isinstance(tag, str):
            tagId = self.getTagId(tag)
        else:
            tagId = tag

        self.query.prepare(SQL_MOVIE_TAG_DELETE)
        self.query.bindValue(':movieId', movieId)
        self.query.bindValue(':tagId', tagId)
        self.query.exec()

        self.setTimestamp()

    def clearMovieTags(self, movieId: int):
        """Удаляет все теги у заметки"""
        self.query.prepare(SQL_MOVIE_TAG_DELETE_ALL_TAGS)
        self.query.bindValue(':movieId', movieId)
        self.query.exec()

    def getMoviesTags(self, movieIds: set) -> set:
        """Возвращает теги указанных заметок"""
        _movieIds = ', '.join(map(str, movieIds))
        if self.query.exec(SQL_MOVIE_TAG_SELECT_TAGS % _movieIds):
            return self.getQuerySet(self.query, 1)

    def find(self, what: str, where: set) -> str:
        qWhere = ''
        for field in where:
            qWhere += ('UPPER_CYR(' + field + ') LIKE "%' +
                       what.upper() + '%" OR ')
        query = 'SELECT fldMovieID FROM tblMovies WHERE ' + qWhere[:-4]
        _q = self.native.execute(query)
        _r = _q.fetchall()
        return '(' + ', '.join(str(r[0]) for r in _r) + ')'


if store.contains(StoreConsts.DBFileName) and os.path.exists(
        store.value(StoreConsts.DBFileName)):
    db_filename = store.value(StoreConsts.DBFileName)
else:
    db_filename = DEFAULT_DB_FULL_NAME

db = DB()
db.open(db_filename)