notes_database.py - master - tari.gpt/ServitorBot

notes_database.py
829 строк28 KB
tarrri
feat: add database foundation for monthly insights digest
03 дек 2025, 11:58
03 дек 2025, 11:58c08e7e6
import sqlite3
from datetime import datetime
import re


def init_notes_db():
    """Инициализация базы данных для заметок"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        # Таблица заметок
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS notes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                content TEXT NOT NULL,
                raw_content TEXT,
                forwarded_message_id INTEGER,
                forwarded_link TEXT,
                category TEXT DEFAULT 'general',
                reflection_date TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users(user_id)
            )
        ''')

        # Таблица тегов
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS tags (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                name TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(user_id, name),
                FOREIGN KEY (user_id) REFERENCES users(user_id)
            )
        ''')

        # Связь заметок и тегов
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS note_tags (
                note_id INTEGER NOT NULL,
                tag_id INTEGER NOT NULL,
                PRIMARY KEY (note_id, tag_id),
                FOREIGN KEY (note_id) REFERENCES notes(id) ON DELETE CASCADE,
                FOREIGN KEY (tag_id) REFERENCES tags(id) ON DELETE CASCADE
            )
        ''')

        # Таблица файлов (для будущего расширения)
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS note_files (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                note_id INTEGER NOT NULL,
                file_type TEXT NOT NULL,
                file_id TEXT NOT NULL,
                file_name TEXT,
                file_size INTEGER,
                mime_type TEXT,
                duration INTEGER,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (note_id) REFERENCES notes(id) ON DELETE CASCADE
            )
        ''')

        # Индексы для оптимизации
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_notes_user_id ON notes(user_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_notes_category ON notes(category)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_tags_user_id ON tags(user_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_note_tags_note_id ON note_tags(note_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_note_tags_tag_id ON note_tags(tag_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_note_files_note_id ON note_files(note_id)')

        conn.commit()
        conn.close()

        print("Таблицы заметок успешно инициализированы")
        return True
    except sqlite3.Error as e:
        print(f"Ошибка при инициализации БД заметок: {e}")
        return False


# ============================================================================
# CRUD функции для заметок
# ============================================================================

def add_note(user_id, content, raw_content=None, category='general',
             forwarded_message_id=None, forwarded_link=None, reflection_date=None):
    """
    Добавляет заметку в базу данных

    Args:
        user_id: ID пользователя Telegram
        content: Содержимое заметки (до 4096 символов)
        raw_content: Оригинальный текст (если был AI)
        category: Категория ('general' или 'reflection')
        forwarded_message_id: ID пересланного сообщения
        forwarded_link: URL ссылка
        reflection_date: Дата рефлексии (для категории reflection)

    Returns:
        note_id если успешно, None в случае ошибки
    """
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        cursor.execute('''
            INSERT INTO notes (user_id, content, raw_content, category,
                             forwarded_message_id, forwarded_link, reflection_date)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (user_id, content, raw_content, category,
              forwarded_message_id, forwarded_link, reflection_date))

        note_id = cursor.lastrowid
        conn.commit()
        conn.close()

        return note_id
    except sqlite3.Error as e:
        print(f"Ошибка при добавлении заметки: {e}")
        return None


def get_note_by_id(note_id, user_id):
    """
    Получает заметку по ID

    Args:
        note_id: ID заметки
        user_id: ID пользователя (для проверки прав доступа)

    Returns:
        Словарь с данными заметки + список тегов или None
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        # Получаем заметку
        cursor.execute('''
            SELECT id, user_id, content, raw_content, forwarded_message_id,
                   forwarded_link, category, reflection_date,
                   created_at, updated_at
            FROM notes
            WHERE id = ? AND user_id = ?
        ''', (note_id, user_id))

        row = cursor.fetchone()

        if not row:
            conn.close()
            return None

        # Получаем теги заметки
        cursor.execute('''
            SELECT t.name
            FROM tags t
            JOIN note_tags nt ON t.id = nt.tag_id
            WHERE nt.note_id = ?
        ''', (note_id,))

        tags = [tag_row['name'] for tag_row in cursor.fetchall()]
        conn.close()

        return {
            'id': row['id'],
            'user_id': row['user_id'],
            'content': row['content'],
            'raw_content': row['raw_content'],
            'forwarded_message_id': row['forwarded_message_id'],
            'forwarded_link': row['forwarded_link'],
            'category': row['category'],
            'reflection_date': row['reflection_date'],
            'created_at': row['created_at'],
            'updated_at': row['updated_at'],
            'tags': tags
        }
    except sqlite3.Error as e:
        print(f"Ошибка при получении заметки: {e}")
        return None


def update_note_content(note_id, user_id, new_content):
    """
    Обновляет содержимое заметки

    Args:
        note_id: ID заметки
        user_id: ID пользователя (для проверки прав доступа)
        new_content: Новое содержимое

    Returns:
        True если успешно, False в случае ошибки
    """
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        # Проверяем существование и права доступа
        cursor.execute('SELECT id FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))

        if cursor.fetchone() is None:
            print(f"Заметка {note_id} не найдена или не принадлежит пользователю {user_id}")
            conn.close()
            return False

        # Обновляем содержимое и время обновления
        cursor.execute('''
            UPDATE notes
            SET content = ?, updated_at = CURRENT_TIMESTAMP
            WHERE id = ? AND user_id = ?
        ''', (new_content, note_id, user_id))

        conn.commit()
        conn.close()
        return True
    except sqlite3.Error as e:
        print(f"Ошибка при обновлении заметки: {e}")
        return False


def delete_note(note_id, user_id):
    """
    Удаляет заметку (CASCADE удалит теги)

    Args:
        note_id: ID заметки
        user_id: ID пользователя (для проверки прав доступа)

    Returns:
        True если успешно, False в случае ошибки
    """
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        # Проверяем существование и права доступа
        cursor.execute('SELECT id FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))

        if cursor.fetchone() is None:
            print(f"Заметка {note_id} не найдена или не принадлежит пользователю {user_id}")
            conn.close()
            return False

        # Удаляем заметку (связи удалятся автоматически через CASCADE)
        cursor.execute('DELETE FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))

        conn.commit()
        conn.close()
        return True
    except sqlite3.Error as e:
        print(f"Ошибка при удалении заметки: {e}")
        return False


def get_all_notes(user_id, limit=5, offset=0):
    """
    Получает все заметки пользователя с пагинацией

    Args:
        user_id: ID пользователя
        limit: Количество заметок на странице (по умолчанию 5)
        offset: Смещение (по умолчанию 0)

    Returns:
        Список словарей с заметками + теги для каждой
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        # Получаем заметки (исключая рефлексии)
        cursor.execute('''
            SELECT id, content, category, reflection_date, created_at, updated_at
            FROM notes
            WHERE user_id = ? AND (category IS NULL OR category != 'reflection')
            ORDER BY created_at DESC
            LIMIT ? OFFSET ?
        ''', (user_id, limit, offset))

        rows = cursor.fetchall()
        notes = []

        for row in rows:
            # Получаем теги для каждой заметки
            cursor.execute('''
                SELECT t.name
                FROM tags t
                JOIN note_tags nt ON t.id = nt.tag_id
                WHERE nt.note_id = ?
            ''', (row['id'],))

            tags = [tag_row['name'] for tag_row in cursor.fetchall()]

            notes.append({
                'id': row['id'],
                'content': row['content'],
                'category': row['category'],
                'reflection_date': row['reflection_date'],
                'created_at': row['created_at'],
                'updated_at': row['updated_at'],
                'tags': tags
            })

        conn.close()
        return notes
    except sqlite3.Error as e:
        print(f"Ошибка при получении списка заметок: {e}")
        return []


def get_total_notes_count(user_id):
    """Возвращает общее количество заметок пользователя (исключая рефлексии)"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        cursor.execute('''
            SELECT COUNT(*) as count
            FROM notes
            WHERE user_id = ? AND (category IS NULL OR category != 'reflection')
        ''', (user_id,))
        result = cursor.fetchone()
        conn.close()

        return result[0] if result else 0
    except sqlite3.Error as e:
        print(f"Ошибка при подсчёте заметок: {e}")
        return 0


# ============================================================================
# Функции для работы с тегами
# ============================================================================

def add_tag(user_id, tag_name):
    """
    Создаёт тег или возвращает существующий

    Args:
        user_id: ID пользователя
        tag_name: Название тега (будет приведено к нижнему регистру)

    Returns:
        tag_id
    """
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        tag_name = tag_name.lower().strip()

        # Пытаемся вставить или игнорируем, если существует
        cursor.execute('''
            INSERT OR IGNORE INTO tags (user_id, name)
            VALUES (?, ?)
        ''', (user_id, tag_name))

        # Получаем ID тега (либо только что созданного, либо существующего)
        cursor.execute('''
            SELECT id FROM tags WHERE user_id = ? AND name = ?
        ''', (user_id, tag_name))

        tag_id = cursor.fetchone()[0]

        conn.commit()
        conn.close()

        return tag_id
    except sqlite3.Error as e:
        print(f"Ошибка при добавлении тега: {e}")
        return None


def get_user_tags_with_count(user_id):
    """
    Получает список тегов с количеством заметок

    Args:
        user_id: ID пользователя

    Returns:
        Список словарей [{'id': 1, 'name': 'работа', 'count': 15}, ...]
        Сортировка: по количеству DESC
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        cursor.execute('''
            SELECT t.id, t.name, COUNT(nt.note_id) as count
            FROM tags t
            LEFT JOIN note_tags nt ON t.id = nt.tag_id
            WHERE t.user_id = ?
            GROUP BY t.id, t.name
            ORDER BY count DESC, t.name ASC
        ''', (user_id,))

        rows = cursor.fetchall()
        conn.close()

        return [{'id': row['id'], 'name': row['name'], 'count': row['count']} for row in rows]
    except sqlite3.Error as e:
        print(f"Ошибка при получении тегов: {e}")
        return []


def link_note_tags(note_id, tag_ids):
    """
    Привязывает заметку к тегам

    Args:
        note_id: ID заметки
        tag_ids: Список ID тегов

    Returns:
        True если успешно, False в случае ошибки
    """
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        # Удаляем старые связи
        cursor.execute('DELETE FROM note_tags WHERE note_id = ?', (note_id,))

        # Добавляем новые связи
        for tag_id in tag_ids:
            cursor.execute('''
                INSERT OR IGNORE INTO note_tags (note_id, tag_id)
                VALUES (?, ?)
            ''', (note_id, tag_id))

        conn.commit()
        conn.close()
        return True
    except sqlite3.Error as e:
        print(f"Ошибка при связывании тегов: {e}")
        return False


def get_notes_by_tag(user_id, tag_id, limit=5, offset=0):
    """
    Получает заметки по тегу с пагинацией

    Args:
        user_id: ID пользователя
        tag_id: ID тега
        limit: Количество заметок
        offset: Смещение

    Returns:
        Список словарей с заметками
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        cursor.execute('''
            SELECT n.id, n.content, n.category, n.reflection_date, n.created_at, n.updated_at
            FROM notes n
            JOIN note_tags nt ON n.id = nt.note_id
            WHERE n.user_id = ? AND nt.tag_id = ?
              AND (n.category IS NULL OR n.category != 'reflection')
            ORDER BY n.created_at DESC
            LIMIT ? OFFSET ?
        ''', (user_id, tag_id, limit, offset))

        rows = cursor.fetchall()
        notes = []

        for row in rows:
            # Получаем все теги для каждой заметки
            cursor.execute('''
                SELECT t.name
                FROM tags t
                JOIN note_tags nt ON t.id = nt.tag_id
                WHERE nt.note_id = ?
            ''', (row['id'],))

            tags = [tag_row['name'] for tag_row in cursor.fetchall()]

            notes.append({
                'id': row['id'],
                'content': row['content'],
                'category': row['category'],
                'reflection_date': row['reflection_date'],
                'created_at': row['created_at'],
                'updated_at': row['updated_at'],
                'tags': tags
            })

        conn.close()
        return notes
    except sqlite3.Error as e:
        print(f"Ошибка при получении заметок по тегу: {e}")
        return []


def get_tag_notes_count(user_id, tag_id):
    """Возвращает количество заметок с тегом (исключая рефлексии)"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        cursor.execute('''
            SELECT COUNT(*) as count
            FROM notes n
            JOIN note_tags nt ON n.id = nt.note_id
            WHERE n.user_id = ? AND nt.tag_id = ?
              AND (n.category IS NULL OR n.category != 'reflection')
        ''', (user_id, tag_id))

        result = cursor.fetchone()
        conn.close()

        return result[0] if result else 0
    except sqlite3.Error as e:
        print(f"Ошибка при подсчёте заметок с тегом: {e}")
        return 0


def get_untagged_notes_count(user_id):
    """Возвращает количество заметок без тегов (исключая рефлексии)"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        cursor.execute('''
            SELECT COUNT(*) as count
            FROM notes n
            WHERE n.user_id = ?
            AND (n.category IS NULL OR n.category != 'reflection')
            AND NOT EXISTS (
                SELECT 1 FROM note_tags nt WHERE nt.note_id = n.id
            )
        ''', (user_id,))

        result = cursor.fetchone()
        conn.close()

        return result[0] if result else 0
    except sqlite3.Error as e:
        print(f"Ошибка при подсчёте заметок без тегов: {e}")
        return 0


def get_untagged_notes(user_id, limit=5, offset=0):
    """Получает заметки без тегов (исключая рефлексии)"""
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        cursor.execute('''
            SELECT n.id, n.content, n.category, n.reflection_date, n.created_at, n.updated_at
            FROM notes n
            WHERE n.user_id = ?
            AND (n.category IS NULL OR n.category != 'reflection')
            AND NOT EXISTS (
                SELECT 1 FROM note_tags nt WHERE nt.note_id = n.id
            )
            ORDER BY n.created_at DESC
            LIMIT ? OFFSET ?
        ''', (user_id, limit, offset))

        rows = cursor.fetchall()
        conn.close()

        notes = []
        for row in rows:
            notes.append({
                'id': row['id'],
                'content': row['content'],
                'category': row['category'],
                'reflection_date': row['reflection_date'],
                'created_at': row['created_at'],
                'updated_at': row['updated_at'],
                'tags': []
            })

        return notes
    except sqlite3.Error as e:
        print(f"Ошибка при получении заметок без тегов: {e}")
        return []


# ============================================================================
# Функции поиска
# ============================================================================

def search_notes(user_id, query, limit=5, offset=0):
    """
    Полнотекстовый поиск по заметкам

    Args:
        user_id: ID пользователя
        query: Поисковый запрос
        limit: Количество результатов
        offset: Смещение

    Returns:
        Список словарей с заметками, содержащими запрос
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        search_pattern = f'%{query}%'

        cursor.execute('''
            SELECT n.id, n.content, n.category, n.reflection_date, n.created_at, n.updated_at
            FROM notes n
            WHERE n.user_id = ? AND n.content LIKE ?
              AND (n.category IS NULL OR n.category != 'reflection')
            ORDER BY n.created_at DESC
            LIMIT ? OFFSET ?
        ''', (user_id, search_pattern, limit, offset))

        rows = cursor.fetchall()
        notes = []

        for row in rows:
            # Получаем теги
            cursor.execute('''
                SELECT t.name
                FROM tags t
                JOIN note_tags nt ON t.id = nt.tag_id
                WHERE nt.note_id = ?
            ''', (row['id'],))

            tags = [tag_row['name'] for tag_row in cursor.fetchall()]

            notes.append({
                'id': row['id'],
                'content': row['content'],
                'category': row['category'],
                'reflection_date': row['reflection_date'],
                'created_at': row['created_at'],
                'updated_at': row['updated_at'],
                'tags': tags
            })

        conn.close()
        return notes
    except sqlite3.Error as e:
        print(f"Ошибка при поиске заметок: {e}")
        return []


def get_search_count(user_id, query):
    """Возвращает количество найденных заметок (исключая рефлексии)"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        search_pattern = f'%{query}%'

        cursor.execute('''
            SELECT COUNT(*) as count
            FROM notes
            WHERE user_id = ? AND content LIKE ?
              AND (category IS NULL OR category != 'reflection')
        ''', (user_id, search_pattern))

        result = cursor.fetchone()
        conn.close()

        return result[0] if result else 0
    except sqlite3.Error as e:
        print(f"Ошибка при подсчёте результатов поиска: {e}")
        return 0


# ============================================================================
# Вспомогательные функции для файлов (для будущего расширения)
# ============================================================================

def add_note_file(note_id, file_type, file_id, file_name=None, file_size=None,
                  mime_type=None, duration=None):
    """Добавляет файл к заметке"""
    try:
        conn = sqlite3.connect('tasks.db')
        cursor = conn.cursor()

        cursor.execute('''
            INSERT INTO note_files (note_id, file_type, file_id, file_name,
                                  file_size, mime_type, duration)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (note_id, file_type, file_id, file_name, file_size, mime_type, duration))

        conn.commit()
        conn.close()
        return True
    except sqlite3.Error as e:
        print(f"Ошибка при добавлении файла: {e}")
        return False


def get_note_files(note_id):
    """Получает список файлов заметки"""
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        cursor.execute('''
            SELECT id, file_type, file_id, file_name, file_size, mime_type, duration
            FROM note_files
            WHERE note_id = ?
            ORDER BY created_at ASC
        ''', (note_id,))

        rows = cursor.fetchall()
        conn.close()

        files = []
        for row in rows:
            files.append({
                'id': row['id'],
                'file_type': row['file_type'],
                'file_id': row['file_id'],
                'file_name': row['file_name'],
                'file_size': row['file_size'],
                'mime_type': row['mime_type'],
                'duration': row['duration']
            })

        return files
    except sqlite3.Error as e:
        print(f"Ошибка при получении файлов: {e}")
        return []


def extract_insights_from_reflection(content):
    """
    Извлекает инсайты из структурированного контента рефлексии.

    Поддерживает паттерны:
    - "💡 Скрытый инсайт:"
    - "💡 HIDDEN INSIGHT:" (legacy English)
    - "💡 Инсайт:"

    Args:
        content: Структурированный текст рефлексии

    Returns:
        str: Извлеченный инсайт или пустая строка если не найден
    """
    if not content:
        return ""

    # Паттерны для поиска инсайтов (регистронезависимые)
    patterns = [
        r'💡\s*Скрытый\s+инсайт:\s*(.+?)(?=\n\n|\n💡|\Z)',
        r'💡\s*HIDDEN\s+INSIGHT:\s*(.+?)(?=\n\n|\n💡|\Z)',
        r'💡\s*Инсайт:\s*(.+?)(?=\n\n|\n💡|\Z)'
    ]

    for pattern in patterns:
        match = re.search(pattern, content, re.IGNORECASE | re.DOTALL)
        if match:
            insight = match.group(1).strip()
            # Удаляем лишние переносы строк
            insight = re.sub(r'\n+', ' ', insight)
            return insight

    return ""


def get_monthly_reflections_with_insights(user_id, year, month):
    """
    Получает все рефлексии с инсайтами за указанный месяц.

    Args:
        user_id: ID пользователя
        year: Год (например, 2024)
        month: Месяц (1-12)

    Returns:
        list: Список словарей с полями:
            - reflection_date: дата рефлексии
            - content: полный контент рефлексии
            - insight: извлеченный инсайт
    """
    try:
        conn = sqlite3.connect('tasks.db')
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        # SQL запрос для получения рефлексий за месяц
        # reflection_date в формате YYYY-MM-DD, используем LIKE для фильтрации
        date_pattern = f"{year}-{month:02d}-%"

        cursor.execute('''
            SELECT reflection_date, content
            FROM notes
            WHERE user_id = ?
              AND category = 'reflection'
              AND reflection_date LIKE ?
            ORDER BY reflection_date ASC
        ''', (user_id, date_pattern))

        reflections = []
        for row in cursor.fetchall():
            insight = extract_insights_from_reflection(row['content'])

            # Включаем только рефлексии с инсайтами
            if insight:
                reflections.append({
                    'reflection_date': row['reflection_date'],
                    'content': row['content'],
                    'insight': insight
                })

        conn.close()
        return reflections

    except sqlite3.Error as e:
        print(f"Ошибка при получении рефлексий за {year}-{month:02d}: {e}")
        return []