245 lines
No EOL
9.3 KiB
Python
245 lines
No EOL
9.3 KiB
Python
"""Модуль для работы с базой данных SQLite"""
|
||
import sqlite3
|
||
import os
|
||
from datetime import datetime
|
||
from app.config import Config
|
||
|
||
|
||
class Database:
|
||
def __init__(self):
|
||
self.db_path = Config.DB_PATH
|
||
self.init_db()
|
||
|
||
def get_connection(self):
|
||
"""Получить соединение с БД"""
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
def init_db(self):
|
||
"""Инициализация базы данных"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# Таблица настроек
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS settings
|
||
(
|
||
key
|
||
TEXT
|
||
PRIMARY
|
||
KEY,
|
||
value
|
||
TEXT
|
||
NOT
|
||
NULL
|
||
)
|
||
''')
|
||
|
||
# Таблица заметок
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS notes
|
||
(
|
||
id
|
||
INTEGER
|
||
PRIMARY
|
||
KEY
|
||
AUTOINCREMENT,
|
||
title
|
||
TEXT,
|
||
content
|
||
TEXT,
|
||
created_at
|
||
TIMESTAMP
|
||
DEFAULT
|
||
CURRENT_TIMESTAMP,
|
||
updated_at
|
||
TIMESTAMP
|
||
DEFAULT
|
||
CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
conn.commit()
|
||
|
||
def get_setting(self, key, default=None):
|
||
"""Получить значение настройки"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT value FROM settings WHERE key = ?', (key,))
|
||
row = cursor.fetchone()
|
||
return row['value'] if row else default
|
||
|
||
def set_setting(self, key, value):
|
||
"""Установить значение настройки"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO settings (key, value)
|
||
VALUES (?, ?)
|
||
''', (key, value))
|
||
conn.commit()
|
||
|
||
def get_notes_by_date(self, year, month, day=None):
|
||
"""Получить заметки по дате"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
if day:
|
||
cursor.execute('''
|
||
SELECT *
|
||
FROM notes
|
||
WHERE strftime('%Y', created_at) = ?
|
||
AND strftime('%m', created_at) = ?
|
||
AND strftime('%d', created_at) = ?
|
||
ORDER BY created_at DESC
|
||
''', (str(year), str(month).zfill(2), str(day).zfill(2)))
|
||
else:
|
||
cursor.execute('''
|
||
SELECT *
|
||
FROM notes
|
||
WHERE strftime('%Y', created_at) = ?
|
||
AND strftime('%m', created_at) = ?
|
||
ORDER BY created_at DESC
|
||
''', (str(year), str(month).zfill(2)))
|
||
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_notes_with_content_by_date(self, date_str):
|
||
"""Получить заметки с контентом по дате (YYYY-MM-DD)"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT *
|
||
FROM notes
|
||
WHERE date (created_at) = ?
|
||
ORDER BY created_at DESC
|
||
''', (date_str,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_note_by_id(self, note_id):
|
||
"""Получить заметку по ID"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM notes WHERE id = ?', (note_id,))
|
||
row = cursor.fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def create_note(self, title=None, content='', created_at=None):
|
||
"""Создать новую заметку"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
if created_at:
|
||
cursor.execute('''
|
||
INSERT INTO notes (title, content, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?)
|
||
''', (title, content, created_at, created_at))
|
||
else:
|
||
cursor.execute('''
|
||
INSERT INTO notes (title, content)
|
||
VALUES (?, ?)
|
||
''', (title, content))
|
||
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def update_note(self, note_id, title=None, content=None):
|
||
"""Обновить заметку"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
updates = []
|
||
params = []
|
||
|
||
if title is not None:
|
||
updates.append('title = ?')
|
||
params.append(title)
|
||
|
||
if content is not None:
|
||
updates.append('content = ?')
|
||
params.append(content)
|
||
|
||
if updates:
|
||
updates.append('updated_at = CURRENT_TIMESTAMP')
|
||
params.append(note_id)
|
||
|
||
cursor.execute(f'''
|
||
UPDATE notes
|
||
SET {', '.join(updates)}
|
||
WHERE id = ?
|
||
''', params)
|
||
conn.commit()
|
||
|
||
def delete_note(self, note_id):
|
||
"""Удалить заметку"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM notes WHERE id = ?', (note_id,))
|
||
conn.commit()
|
||
|
||
def get_all_notes(self):
|
||
"""Получить все заметки"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM notes ORDER BY created_at DESC')
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def export_data(self):
|
||
"""Экспортировать все данные"""
|
||
return {
|
||
'notes': self.get_all_notes(),
|
||
'settings': self.get_all_settings()
|
||
}
|
||
|
||
def get_all_settings(self):
|
||
"""Получить все настройки"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM settings')
|
||
return {row['key']: row['value'] for row in cursor.fetchall()}
|
||
|
||
def import_data(self, data):
|
||
"""Импортировать данные"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# Очищаем существующие данные
|
||
cursor.execute('DELETE FROM notes')
|
||
cursor.execute('DELETE FROM settings')
|
||
|
||
# Импортируем заметки
|
||
for note in data.get('notes', []):
|
||
cursor.execute('''
|
||
INSERT INTO notes (id, title, content, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
''', (
|
||
note['id'],
|
||
note.get('title'),
|
||
note.get('content', ''),
|
||
note.get('created_at'),
|
||
note.get('updated_at')
|
||
))
|
||
|
||
# Импортируем настройки
|
||
for key, value in data.get('settings', {}).items():
|
||
cursor.execute('''
|
||
INSERT INTO settings (key, value)
|
||
VALUES (?, ?)
|
||
''', (key, value))
|
||
|
||
conn.commit()
|
||
|
||
def get_dates_with_notes(self, year, month):
|
||
"""Получить дни месяца, в которых есть заметки"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT DISTINCT strftime('%d', created_at) as day
|
||
FROM notes
|
||
WHERE strftime('%Y'
|
||
, created_at) = ?
|
||
AND strftime('%m'
|
||
, created_at) = ?
|
||
''', (str(year), str(month).zfill(2)))
|
||
return [int(row['day']) for row in cursor.fetchall()] |