thingman

  1"""
  2Управление местоположений вещей: CRUD-операции с SQLite.
  3
  4Предоставляет класс ThingManager для работы с записями о вещях и их местах,
  5включая добавление, обновление, поиск и удаление локаций.
  6Хранение осуществляется в SQLite.
  7"""
  8
  9import os
 10import sqlite3
 11from datetime import datetime
 12from typing import Dict, List, Optional, cast
 13
 14from langfuse import observe
 15from loguru import logger
 16
 17DATA_DIR = os.getenv("DATA_DIR", "./data")
 18DB_FILE = os.path.join(DATA_DIR, "locations.db")
 19
 20
 21def _convert_location(row: sqlite3.Row) -> Dict:
 22    """
 23    Преобразует строку из SQLite (sqlite3.Row) в словарь.
 24
 25    Args:
 26        row: Строка результата запроса SQLite
 27
 28    Returns:
 29        Dict: Словарь с данными локации
 30    """
 31    return dict(row)
 32
 33
 34def _common_prefix_len(a: str, b: str) -> int:
 35    """
 36    Длина общего префикса двух строк.
 37
 38    Считает количество совпадающих символов с начала строк (для учета
 39    словоформ: 'болгарку' и 'болгарка' дают 7).
 40
 41    Args:
 42        a: Первая строка
 43        b: Вторая строка
 44
 45    Returns:
 46        int: Длина общего префикса
 47    """
 48    for i, (ca, cb) in enumerate(zip(a, b)):
 49        if ca != cb:
 50            return i
 51    return min(len(a), len(b))
 52
 53
 54def _token_in_thing(token: str, thing: str) -> bool:
 55    """
 56    Проверяет, входит ли токен запроса в название вещи.
 57
 58    Ищет вхождение подстроки, совпадение на уровне отдельного слова
 59    и морфологическую близость (длинный общий префикс — например,
 60    'болгарку' vs 'болгарка').
 61
 62    Args:
 63        token: Токен поискового запроса
 64        thing: Название вещи
 65
 66    Returns:
 67        bool: True, если токен релевантен для названия вещи
 68    """
 69    if token in thing:
 70        return True
 71    for word in thing.split():
 72        if token in word or word in token:
 73            return True
 74        common = _common_prefix_len(token, word)
 75        if common >= 3 and common >= 0.6 * min(len(token), len(word)):
 76            return True
 77    return False
 78
 79
 80class ThingManager:
 81    """
 82    Менеджер вещей с хранением в SQLite.
 83
 84    Предоставляет полный набор CRUD-операций для управления записями о вещях
 85    пользователей. Поддерживает поиск по thing и возврат всех мест.
 86    """
 87
 88    def __init__(self, db_path: str = DB_FILE):
 89        """
 90        Инициализирует ThingManager, создавая/подключая БД SQLite.
 91
 92        При первом запуске создаёт необходимые таблицы и индексы.
 93
 94        Args:
 95            db_path: Путь к файлу базы данных SQLite
 96        """
 97        self.db_path = db_path
 98        os.makedirs(DATA_DIR, exist_ok=True)
 99        self._init_db()
100
101    def _get_conn(self) -> sqlite3.Connection:
102        """
103        Создаёт и возвращает подключение к SQLite.
104
105        Настраивает row_factory для доступа по имени колонки,
106        включает WAL-режим журнала и таймаут ожидания блокировки.
107
108        Returns:
109            sqlite3.Connection: Подключение к БД
110        """
111        conn = sqlite3.connect(self.db_path)
112        conn.row_factory = sqlite3.Row
113        conn.execute("PRAGMA journal_mode=WAL")
114        conn.execute("PRAGMA busy_timeout=5000")
115        return conn
116
117    def _init_db(self):
118        """
119        Инициализирует структуру базы данных.
120
121        Создаёт таблицу locations (если не существует) с колонками:
122        id, user_id, thing, place, created_at.
123        Также создаёт индексы для ускорения запросов.
124        """
125        with self._get_conn() as conn:
126            conn.executescript("""
127                CREATE TABLE IF NOT EXISTS locations (
128                    id INTEGER PRIMARY KEY AUTOINCREMENT,
129                    user_id TEXT NOT NULL,
130                    thing TEXT NOT NULL,
131                    place TEXT NOT NULL,
132                    created_at TEXT NOT NULL
133                );
134                CREATE INDEX IF NOT EXISTS idx_loc_thing ON locations(thing);
135                CREATE INDEX IF NOT EXISTS idx_loc_user_thing ON locations(user_id, thing);
136            """)
137
138    @observe(name="Add location")
139    def add_location(self, user_id: str, thing: str, place: str) -> int:
140        """
141        Добавляет новую запись о местоположении вещи.
142
143        Args:
144            user_id: Идентификатор пользователя
145            thing: Название вещи
146            place: Название места
147
148        Returns:
149            int: ID созданной записи
150        """
151        with self._get_conn() as conn:
152            cur = conn.execute(
153                "INSERT INTO locations (user_id, thing, place, created_at) VALUES (?, ?, ?, ?)",
154                (user_id, thing.lower(), place.lower(), datetime.now().isoformat()),
155            )
156            loc_id = cast(int, cur.lastrowid)
157            logger.debug(f"Добавлена запись {loc_id}: '{thing}' → '{place}'")
158            return loc_id
159
160    @observe(name="Update location")
161    def update_location(self, user_id: str, thing: str, place: str) -> bool:
162        """
163        Обновляет место для существующей вещи.
164
165        Args:
166            user_id: Идентификатор пользователя
167            thing: Название вещи
168            place: Новое название места
169
170        Returns:
171            bool: True если запись найдена и обновлена, False в противном случае
172        """
173        with self._get_conn() as conn:
174            cur = conn.execute(
175                "UPDATE locations SET place = ?, created_at = ? WHERE user_id = ? AND thing = ?",
176                (place.lower(), datetime.now().isoformat(), user_id, thing.lower()),
177            )
178            if cur.rowcount > 0:
179                logger.debug(f"Обновлена место для '{thing}': '{place}'")
180                return True
181        return False
182
183    @observe(name="Find thing locations")
184    def find_thing(self, user_id: str, thing: str) -> List[Dict]:
185        """
186        Возвращает все места где вещь была записена.
187
188        Args:
189            user_id: Идентификатор пользователя
190            thing: Название вещи для поиска
191
192        Returns:
193            List[Dict]: Список записей о местоположении вещи, отсортированных по ID (новые сверху)
194        """
195        with self._get_conn() as conn:
196            rows = conn.execute(
197                "SELECT * FROM locations WHERE user_id = ? AND thing = ? ORDER BY id DESC",
198                (user_id, thing.lower()),
199            ).fetchall()
200            return [_convert_location(row) for row in rows]
201
202    @observe(name="Search things")
203    def search_thing(self, user_id: str, query: str, limit: int = 5) -> List[Dict]:
204        """
205        Нечёткий поиск вещей по названию.
206
207        Разбивает запрос на токены и ищет записи, где токен входит в название
208        вещи как подстрока либо морфологически близок к слову (общий префикс —
209        учёт словоформ вроде 'болгарку' vs 'болгарка'). Ранжирует по количеству
210        совпавших токенов и возвращает уникальные вещи (по самому свежему месту).
211
212        Args:
213            user_id: Идентификатор пользователя
214            query: Поисковый запрос (произвольная фраза)
215            limit: Максимальное количество результатов (по умолчанию 5)
216
217        Returns:
218            List[Dict]: Найденные записи (thing, place, id), отсортированные
219                по релевантности — сначала больше совпавших токенов, затем свежие
220        """
221        tokens = [t for t in query.strip().lower().split() if t]
222        if not tokens:
223            return []
224
225        params = [user_id]
226        clauses = []
227        for tok in tokens:
228            clauses.append("thing LIKE ?")
229            params.append(f"%{tok}%")
230            if len(tok) >= 5:
231                # префиксное вхождение ловит словоформы ('болгарку' → 'болгарка')
232                clauses.append("thing LIKE ?")
233                params.append(f"%{tok[:4]}%")
234
235        sql = (
236            "SELECT * FROM locations WHERE user_id = ? AND ("
237            + " OR ".join(clauses)
238            + ") ORDER BY id DESC"
239        )
240
241        with self._get_conn() as conn:
242            rows = conn.execute(sql, params).fetchall()
243
244        best: Dict[str, tuple] = {}
245        for row in rows:
246            loc = _convert_location(row)
247            thing = loc["thing"]
248            score = sum(1 for tok in tokens if _token_in_thing(tok, thing))
249            if score == 0:
250                continue
251            if thing not in best or score > best[thing][0]:
252                best[thing] = (score, loc["place"], loc["id"])
253
254        ranked = sorted(best.items(), key=lambda kv: (kv[1][0], kv[1][2]), reverse=True)
255        return [
256            {"thing": thing, "place": place, "id": rec_id}
257            for thing, (score, place, rec_id) in ranked[:limit]
258        ]
259
260    @observe(name="Get all things")
261    def get_all_things(self, user_id: str) -> List[Dict]:
262        """
263        Возвращает все вещи пользователя.
264
265        Args:
266            user_id: Идентификатор пользователя
267
268        Returns:
269            List[Dict]: Список всех записей, отсортированных по ID
270        """
271        with self._get_conn() as conn:
272            rows = conn.execute(
273                "SELECT * FROM locations WHERE user_id = ? ORDER BY id", (user_id,)
274            ).fetchall()
275            return [_convert_location(row) for row in rows]
276
277    @observe(name="Delete location")
278    def delete_location(self, user_id: str, thing: str) -> bool:
279        """
280        Удаляет все записи о вещи по её названию.
281
282        Args:
283            user_id: Идентификатор пользователя
284            thing: Название вещи для удаления
285
286        Returns:
287            bool: True если записи найдены и удалены, False в противном случае
288        """
289        with self._get_conn() as conn:
290            cur = conn.execute(
291                "DELETE FROM locations WHERE user_id = ? AND thing = ?",
292                (user_id, thing.lower()),
293            )
294            if cur.rowcount > 0:
295                logger.debug(f"Удалены записи для '{thing}'")
296                return True
297        return False
298
299    def get_location_count(self, user_id: Optional[str] = None) -> int:
300        """
301        Возвращает количество записей пользователя.
302
303        Args:
304            user_id: Идентификатор пользователя. Если None — считает все записи.
305
306        Returns:
307            int: Количество записей
308        """
309        if user_id is None:
310            query = "SELECT COUNT(*) FROM locations"
311            params: tuple = ()
312        else:
313            query = "SELECT COUNT(*) FROM locations WHERE user_id = ?"
314            params = (user_id,)
315        with self._get_conn() as conn:
316            row = conn.execute(query, params).fetchone()
317            return row[0]
318
319    @observe(name="Format location list")
320    def format_location_list(self, locations: List[Dict], max_chars: int = 1024) -> str:
321        """
322        Форматирует список местоположений в строку с учётом ограничения длины.
323
324        Args:
325            locations: Список записей (словарей)
326            max_chars: Максимальная длина строки (по умолчанию 1024)
327
328        Returns:
329            str: Отформатированный список местоположений
330        """
331        if not locations:
332            return "Записей нет."
333
334        raw_lines = [f"{loc['thing']} → {loc['place']}" for loc in locations]
335        raw_lines.reverse()  # новые сверху
336
337        total = len(raw_lines)
338        tail_template = "\nи еще {} записей"
339        best_lines = []
340
341        for line in raw_lines:
342            tail = (
343                tail_template.format(total - len(best_lines) - 1)
344                if len(best_lines) < total
345                else ""
346            )
347            candidate = "\n".join(best_lines + [line]) + tail
348            if len(candidate) > max_chars:
349                break
350            best_lines.append(line)
351
352        shown = len(best_lines)
353        if shown == total:
354            if shown == 1:
355                return f"Запись: {best_lines[0]}"
356            return "Ваши записей: \n" + "\n".join(best_lines)
357
358        tail = tail_template.format(total - shown)
359        if shown == 0:
360            return f"Не показано ни одной записи. {tail.strip()}"
361
362        return "\n".join(best_lines) + tail
363
364    @observe(name="Get all thing names")
365    def get_all_thing_names(self, user_id: str) -> List[str]:
366        """
367        Возвращает список всех названий вещей пользователя.
368
369        Args:
370            user_id: Идентификатор пользователя
371
372        Returns:
373            List[str]: Список названий вещей
374        """
375        with self._get_conn() as conn:
376            rows = conn.execute(
377                "SELECT DISTINCT thing FROM locations WHERE user_id = ?", (user_id,)
378            ).fetchall()
379            return [row["thing"] for row in rows]
380
381    @observe(name="Get all records")
382    def get_all(self, user_id: Optional[str] = None) -> List[Dict]:
383        """
384        Возвращает все записи о вещах (всех пользователей или одного).
385
386        Args:
387            user_id: Если задан — только записи этого пользователя,
388                иначе записи всех пользователей.
389
390        Returns:
391            List[Dict]: Список записей, отсортированных по ID
392        """
393        with self._get_conn() as conn:
394            if user_id is None:
395                rows = conn.execute("SELECT * FROM locations ORDER BY id").fetchall()
396            else:
397                rows = conn.execute(
398                    "SELECT * FROM locations WHERE user_id = ? ORDER BY id",
399                    (user_id,),
400                ).fetchall()
401            return [_convert_location(row) for row in rows]