Начальная индексация в Typesense

pip install pymysql typesense beautifulsoup4

CREATE TABLE pages (
    id INT PRIMARY KEY AUTO_INCREMENT,
    slug VARCHAR(255) NOT NULL,
    title VARCHAR(255) NOT NULL,
    content MEDIUMTEXT NOT NULL,
    tags VARCHAR(255) NULL,
    published_at DATETIME NOT NULL,
    popularity INT NOT NULL DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

# db.py
import pymysql

def get_connection():
    """
    Возвращает подключение к MySQL.
    Меняй параметры под свою среду.
    """
    return pymysql.connect(
        host="127.0.0.1",
        user="your_user",
        password="your_password",
        db="your_database",
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor, # будем получать dict'ы
    )

# ts_client.py
import typesense

TYPESENSE_API_KEY = "SUPER_SECRET_ADMIN_KEY" # лучше брать из env

client = typesense.Client({
    "nodes": [
        {
            "host": "127.0.0.1", # или Tailscale IP, если сервер отдельный (разберём позже)
            "port": "8108",
            "protocol": "http",
        }
    ],
    "api_key": TYPESENSE_API_KEY,
    "connection_timeout_seconds": 2,
})

# text_utils.py
import re
from bs4 import BeautifulSoup

def html_to_text(html: str) -> str:
    """
    Превращает HTML в простой текст:
    - убираем теги;
    - вычищаем лишние пробелы и переводы строк.
    """
    if not html:
        return ""
    soup = BeautifulSoup(html, "html.parser")
    text = soup.get_text(separator=" ")
    # сжимаем последовательности whitespace в один пробел
    text = re.sub(r"\s+", " ", text)
    return text.strip()

def make_snippet(text: str, max_chars: int = 300) -> str:
    """
    Создаём сниппет для отображения в результатах поиска.
    Не обязательно, но удобно: Typesense может искать по полю,
    а мы показываем краткое превью.
    """
    if len(text) <= max_chars:
        return text
    # обрезаем по словам, чтобы не рвать посередине
    snippet = text[:max_chars]
    last_space = snippet.rfind(" ")
    if last_space > 0:
        snippet = snippet[:last_space]
    return snippet + "…"

def parse_tags(tags_str: str | None) -> list[str]:
    """
    Превращаем строку 'python, flask, поиск' в список ['python', 'flask', 'поиск'].
    """
    if not tags_str:
        return []
    return [tag.strip() for tag in tags_str.split(",") if tag.strip()]

# mapper.py
from datetime import datetime
from text_utils import html_to_text, make_snippet, parse_tags

def row_to_typesense_doc(row: dict) -> dict:
    """
    Конвертирует строку из MySQL в документ для коллекции `pages`.
    Ожидается, что row содержит ключи: id, slug, title, content, tags,
    published_at (datetime), popularity (int).
    """
    raw_text = html_to_text(row["content"])
    snippet = make_snippet(raw_text, max_chars=400)
    tags = parse_tags(row.get("tags"))

    published_at = row["published_at"]
    if isinstance(published_at, datetime):
        # UNIX timestamp (секунды)
        published_ts = int(published_at.timestamp())
    else:
        # на всякий случай — если пришла строка
        # предполагаем формат 'YYYY-MM-DD HH:MM:SS'
        dt = datetime.fromisoformat(str(published_at))
        published_ts = int(dt.timestamp())

    return {
        "id": row["id"],
        "title": row["title"],
        "slug": row["slug"],
        # в индекс кладём сниппет; при желании можно хранить и полный текст в отдельном поле
        "content": snippet,
        "tags": tags,
        "published_at": published_ts,
        "popularity": row["popularity"],
    }

SELECT * FROM pages WHERE id > ? ORDER BY id ASC LIMIT ?

# initial_index.py
import time
from typing import Optional

from db import get_connection
from ts_client import client as ts_client
from mapper import row_to_typesense_doc


BATCH_SIZE = 1000 # сколько страниц загружаем за один проход
COLLECTION_NAME = "pages"


def fetch_batch(last_id: int, limit: int) -> list[dict]:
    """
    Берём очередной батч страниц из MySQL с id > last_id.
    """
    conn = get_connection()
    try:
        with conn.cursor() as cursor:
            sql = """
                SELECT id, slug, title, content, tags, published_at, popularity
                FROM pages
                WHERE id > %s
                ORDER BY id ASC
                LIMIT %s
            """
            cursor.execute(sql, (last_id, limit))
            rows = cursor.fetchall()
            return rows
    finally:
        conn.close()


def index_batch(docs: list[dict]) -> None:
    """
    Отправляем батч документов в Typesense.
    Используем action=upsert, чтобы можно было запускать скрипт повторно.
    """
    if not docs:
        return

    # Python-клиент Typesense умеет принимать список dict'ов.
    # Он сам конвертирует в JSON Lines и отправляет на /documents/import.
    # action=upsert: вставить новый или обновить существующий документ по id.
    result = ts_client.collections[COLLECTION_NAME].documents.import_(
        docs,
        {"action": "upsert"}, # 'create' / 'update' / 'upsert' / 'emplace'
        batch_size=len(docs), # можно оставить по умолчанию, но так нагляднее
    )

    # result обычно — список словарей с полями success/error для каждой строки
    # имеет смысл проверять, не было ли ошибок
    errors = [r for r in result if not r.get("success")]
    if errors:
        # На проде лучше логировать в файл/системный лог
        print(f"Ошибки при импорте {len(errors)} документов:")
        for e in errors[:5]: # не флудим
            print(e)


def initial_index_all():
    """
    Основной цикл первичной индексации всех страниц.
    """
    last_id: int = 0
    total = 0

    while True:
        rows = fetch_batch(last_id, BATCH_SIZE)
        if not rows:
            break # ничего больше нет, выходим

        docs = [row_to_typesense_doc(row) for row in rows]
        index_batch(docs)

        last_id = rows[-1]["id"]
        total += len(rows)
        print(f"Проиндексировано страниц: {total}")

        # Небольшая пауза, чтобы не душить базу и Typesense на слабом сервере
        time.sleep(0.1)

    print(f"Готово. Всего проиндексировано {total} страниц.")


if __name__ == "__main__":
    initial_index_all()

# reindex_single_page.py
from db import get_connection
from ts_client import client as ts_client
from mapper import row_to_typesense_doc

COLLECTION_NAME = "pages"

def reindex_page(page_id: int):
    conn = get_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                """
                SELECT id, slug, title, content, tags, published_at, popularity
                FROM pages
                WHERE id = %s
                """,
                (page_id,),
            )
            row = cursor.fetchone()
    finally:
        conn.close()

    if not row:
        raise ValueError(f"Страница id={page_id} не найдена в MySQL")

    doc = row_to_typesense_doc(row)
    # upsert одного документа
    ts_client.collections[COLLECTION_NAME].documents.upsert(doc)
    print(f"Страница {page_id} переиндексирована.")


if __name__ == "__main__":
    reindex_page(123) # пример