sources/uDB.py - main - VolcanoByte/vbNotes

sources/uDB.py
429 строк15 KB
Eugene Gavrilyuk
mass changes
03 янв 2026, 18:03
03 янв 2026, 18:0304d697b
"""
Управление базой данных.
"""

import datetime as dt
import os
import sqlite3

from unicodedata import normalize, combining

from PyQt6.QtSql import QSqlDatabase, QSqlQuery

from uNote import Note
from uSQL import (SQL_NOTE_TAG_DELETE, SQL_NOTE_TAG_INSERT,
                  SQL_NOTE_TAG_SELECT_TAGS, SQL_NOTE_TAG_SELECT_NAMES,
                  SQL_NOTE_TAG_SELECT, SQL_TAG_SELECT_BY_ID,
                  SQL_TAG_SELECT_BY_NAME, SQL_TAG_DELETE_BY_NAME,
                  SQL_NOTE_TAG_DELETE_ALL_TAGS, SQL_NOTE_DELETE,
                  SQL_TAG_ADD, SQL_TAG_DELETE_BY_ID, SQL_TAG_DELETE_EMPTY,
                  SQL_CREATE_TABLE_TAGS, SQL_CREATE_TABLE_NOTE_TAG,
                  SQL_NOTE_SELECT, SQL_NOTE_TAG_SELECT_NOTES,
                  SQL_TAG_CALC_COUNT, SQL_TAG_CLEAR_COUNT,
                  SQL_GET_REC_COUNT, SQL_PROPERTIES_ADD_ROW,
                  SQL_PROPERTIES_GET, SQL_PROPERTIES_SET,
                  SQL_CREATE_TABLE_NOTES, SQL_CREATE_TABLE_PROPERTIES,
                  SQL_TAG_SET_UPPER_NAME, SQL_NOTE_SELECT_ALL)
from uConsts import DEFAULT_DB_FULL_NAME, StoreConsts
from uGlobal import store


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


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

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

        return fields

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

    @staticmethod
    def queryNotesTags(noteIds: set):
        """Возвращает запрос с выбором тегов указанных заметок"""
        _NotesIds = ', '.join(map(str, noteIds))
        return SQL_NOTE_TAG_SELECT_TAGS % _NotesIds

    @staticmethod
    def queryTagsNotes(tagIds: set):
        """Возвращает запрос с выбором заметок с указанными тегами"""
        _tagsIds = ', '.join(map(str, tagIds))
        _query = SQL_NOTE_TAG_SELECT_NOTES % (_tagsIds, len(tagIds))
        return _query

    @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 ''

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

    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_NOTES)
        self.query.exec(SQL_CREATE_TABLE_TAGS)
        self.query.exec(SQL_CREATE_TABLE_NOTE_TAG)
        self.query.exec(SQL_CREATE_TABLE_PROPERTIES)

    def preparePropertiesTable(self):
        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()

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

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

                    self.query.next()

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

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

    def setProperty(self, name: str, value):
        """Установка свойства"""
        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()

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

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

    def addNote(self, note: Note) -> int:
        """Добавление заметки в базу."""
        if "fldNoteUpdatedAt" not in note.data:
            note.data["fldNoteUpdatedAt"] = getNow()
        if "fldNoteCreatedAt" not in note.data:
            note.data["fldNoteCreatedAt"] = getNow()

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

        noteId = self.query.lastInsertId()

        # Добавляем теги
        self.addNoteTags(noteId, note.tags)

        self.setTimestamp()

        return noteId

    def updateNote(self, noteId: int, note: Note):
        """Обновление заметки"""
        if "fldNoteUpdatedAt" not in note.data:
            note.data["fldNoteUpdatedAt"] = getNow()

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

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

        # Обновляем теги
        self.clearNoteTags(noteId)
        self.addNoteTags(noteId, note.tags)

        self.setTimestamp()

    def getNote(self, noteId) -> Note | None:
        """Возвращает данные заметки с указанным id
        или None если таковой не найден"""
        if isinstance(noteId, str):
            _id = int(noteId)
        else:
            _id = noteId

        self.query.prepare(SQL_NOTE_SELECT)
        self.query.bindValue(':noteId', _id)
        self.query.exec()

        if self.query.first():
            note = Note()

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

            # Читаем теги
            note.tags = self.getNoteTagNames(noteId)

            return note

    def deleteNote(self, noteId: int):
        """Удаляет указанную заметку"""
        self.query.prepare(SQL_NOTE_DELETE)
        self.query.bindValue(':noteId', noteId)
        self.query.exec()

        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())

            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()

    def addNoteTag(self, noteId: int, tag: str | int):
        """Добавляет тег к заметке, если такого тега еще нет"""
        # Проверяем наличие фильма
        if not self.getNote(noteId):
            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_NOTE_TAG_SELECT)
        self.query.bindValue(':noteId', noteId)
        self.query.bindValue(':tagId', tagId)
        self.query.exec()

        if not self.query.first():
            self.query.prepare(SQL_NOTE_TAG_INSERT)
            self.query.bindValue(':noteId', noteId)
            self.query.bindValue(':tagId', tagId)
            self.query.exec()

            self.setTimestamp()

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

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

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

        self.query.prepare(SQL_NOTE_TAG_DELETE)
        self.query.bindValue(':noteId', noteId)
        self.query.bindValue(':tagId', tagId)
        self.query.exec()

        self.setTimestamp()

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

    def getNotesTags(self, noteIds: set) -> set:
        """Возвращает теги указанных заметок"""
        _movieIds = ', '.join(map(str, noteIds))
        if self.query.exec(SQL_NOTE_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 fldNoteID FROM tblNotes 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)