Сообщить об ошибке.

Логирование поисковых запросов и простая аналитика

Содержание:

Добавим логирование запросов из /search и JSON-API, сохраним в MySQL сведения о строке запроса, числе совпадений и времени ответа. Потом сделаем простую админ-страницу аналитики: топ запросов, запросы без результатов и рекомендации, что с этим делать.

Зачем вообще логировать поиск

Поисковый лог - это бесплатная аналитика по мозгу пользователя:

  • какие запросы задают чаще всего;
  • по каким запросам ничего не находится (контентные дыры или плохая морфология);
  • по каким запросам слишком много результатов (нужно уточнять фасеты, синонимы);
  • как быстро отвечает поиск (можно ловить деградации).

Дальше эти данные:

  • подсказывают, какие синонимы и overrides заводить (Часть 4);
  • помогают решать, какие разделы сайта развивать;
  • служат сигналом, что Typesense или база не справляются по скорости.

Таблица логов в MySQL

Сделаем простую таблицу search_log в той же базе, что и pages:

CREATE TABLE search_log (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    q VARCHAR(255) NOT NULL,               -- строка запроса
    hits INT UNSIGNED NOT NULL,            -- сколько документов вернул Typesense
    took_ms INT UNSIGNED NOT NULL,         -- время выполнения запроса к Typesense (мс)
    source ENUM('page', 'api', 'suggest') NOT NULL DEFAULT 'page',
                                           -- откуда запрос: HTML-страница, JSON-API, подсказки
    ip VARCHAR(45) NULL,                   -- IPv4/IPv6 (не обязательно заполнять)
    user_agent VARCHAR(255) NULL,          -- укороченный UA
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (id),
    KEY idx_created_at (created_at),
    KEY idx_q_created_at (q, created_at),
    KEY idx_hits (hits)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Пояснения:

  • hits - общее число найденных документов (found из ответа Typesense).
  • took_ms - время запроса к Typesense (считаем во Flask).
  • source - удобно фильтровать аналитику отдельно по HTML-поиску и API.
  • индексы:
    • idx_q_created_at - для выборки по запросу за период;
    • idx_created_at - для отчётов "за последние N дней";
    • idx_hits - для поиска "запросов без результатов".

Если не хочешь хранить IP/UA - смело выкидывай эти поля, особенно если есть требования по приватности.

Утилита логирования: search_logging.py

Сделаем небольшой модуль, чтобы не плодить SQL по всему коду.

# search_logging.py
from typing import Optional
import time

from db import get_connection # из предыдущих частей у нас есть этот helper :contentReference[oaicite:3]{index=3}


def log_search(
    q: str,
    hits: int,
    took_ms: int,
    source: str = "page",
    ip: Optional[str] = None,
    user_agent: Optional[str] = None,
) -> None:
    """
    Сохраняет одну строку в таблицу search_log.
    Предполагаем, что q уже обрезан по длине и лишние пробелы убраны.
    """
    if not q:
        return

    # На всякий случай слегка ограничим длину тут
    if len(q) > 255:
        q = q[:255]

    conn = get_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO search_log (q, hits, took_ms, source, ip, user_agent)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (q, hits, took_ms, source, ip, user_agent[:255] if user_agent else None),
            )
        conn.commit()
    finally:
        conn.close()


def measure(func):
    """
    Простейший декоратор, который меряет время работы функции в миллисекундах
    и возвращает (result, took_ms).
    """
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        took_ms = int((time.perf_counter() - start) * 1000)
        return result, took_ms

    return wrapper

measure пригодится, чтобы аккуратно замерять время запроса к Typesense.

Логирование в маршруте /search

У нас уже есть маршрут /search, который ходит в Typesense и рендерит search.html. Добавим туда измерение времени и логирование.

# app.py (фрагмент: обновлённый /search)
from flask import Flask, request, render_template
from ts_client_search import client as ts_client # search-only ключ, Ч.5 :contentReference[oaicite:5]{index=5}
from search_logging import log_search, measure

app = Flask(__name__)
COLLECTION_NAME = "pages"


@measure
def typesense_search(params: dict):
    """
    Обёртка над вызовом Typesense, чтобы замерить took_ms.
    """
    return ts_client.collections[COLLECTION_NAME].documents.search(params)


@app.route("/search")
def search():
    q = (request.args.get("q") or "").strip()
    page_arg = request.args.get("page", "1")
    tag = (request.args.get("tag") or "").strip()
    sort = (request.args.get("sort") or "").strip()

    try:
        page = int(page_arg)
        if page < 1:
            page = 1
    except ValueError:
        page = 1

    per_page = 10

    search_params = {
        "q": q or "*",
        "query_by": "title,content",
        "page": page,
        "per_page": per_page,
        "highlight_full_fields": "title,content",
        "highlight_start_tag": "<mark>",
        "highlight_end_tag": "</mark>",
        "facet_by": "tags",
        "max_facet_values": 50,
    }

    filter_clauses = []
    if tag:
        filter_clauses.append(f"tags:=[{tag}]")
    if filter_clauses:
        search_params["filter_by"] = " && ".join(filter_clauses)

    if sort == "new":
        search_params["sort_by"] = "published_at:desc"
    elif sort == "popular":
        search_params["sort_by"] = "popularity:desc"

    # --- сам запрос к Typesense + измерение времени ---
    (result, took_ms) = typesense_search(search_params)

    hits = []
    for hit in result.get("hits", []):
        doc = hit.get("document", {})
        highlights = hit.get("highlights", [])
        hits.append({
            "id": doc.get("id"),
            "slug": doc.get("slug"),
            "title": doc.get("title"),
            "snippet": doc.get("content"),
            "highlights": highlights,
        })

    found = result.get("found", 0)
    facet_counts = result.get("facet_counts", [])
    pages_total = (found + per_page - 1) // per_page if found else 0

    facets = {}
    for facet in facet_counts:
        field_name = facet.get("field_name")
        counts = facet.get("counts", [])
        facets[field_name] = [
            {"value": c.get("value"), "count": c.get("count")}
            for c in counts
        ]

    # --- логирование запроса ---
    try:
        log_search(
            q=q or "*",
            hits=found,
            took_ms=took_ms,
            source="page",
            ip=request.remote_addr,
            user_agent=request.headers.get("User-Agent", "")[:255],
        )
    except Exception as e:
        # В проде лучше залогировать, но не падать из-за проблем логирования
        app.logger.warning("Failed to log search: %s", e)

    return render_template(
        "search.html",
        q=q,
        tag=tag,
        sort=sort,
        hits=hits,
        found=found,
        page=page,
        pages_total=pages_total,
        per_page=per_page,
        facets=facets,
        took_ms=took_ms,
    )

Теперь каждый запрос /search складывается в search_log с числом найденных документов и временем ответа.

Логирование из JSON-API (/api/search, /api/suggest)

У нас есть лёгкие JSON-endpoint’ы для живого поиска и подсказок. Добавим туда логирование так же аккуратно.

/api/search

# app_api.py (фрагмент)
from flask import Flask, request, jsonify
from ts_client_search import client as ts_client
from search_logging import log_search, measure

app = Flask(__name__)
COLLECTION_NAME = "pages"


@measure
def typesense_api_search(params: dict):
    return ts_client.collections[COLLECTION_NAME].documents.search(params)


@app.route("/api/search")
def api_search():
    q = (request.args.get("q") or "").strip()
    if not q:
        return jsonify({"hits": [], "found": 0})

    try:
        limit = int(request.args.get("limit", "5"))
    except ValueError:
        limit = 5
    limit = max(1, min(limit, 20))

    search_params = {
        "q": q,
        "query_by": "title,content",
        "per_page": limit,
        "page": 1,
        "include_fields": "id,slug,title",
        "query_by_weights": "3,1",
        "prefix": "true",
    }

    (result, took_ms) = typesense_api_search(search_params)

    hits = []
    for hit in result.get("hits", []):
        doc = hit.get("document", {})
        hits.append({
            "id": doc.get("id"),
            "slug": doc.get("slug"),
            "title": doc.get("title"),
        })

    found = result.get("found", 0)

    # Логируем, но без IP/UA — если API зовут с фронта, они у нас есть:
    try:
        log_search(
            q=q,
            hits=found,
            took_ms=took_ms,
            source="api",
            ip=request.remote_addr,
            user_agent=request.headers.get("User-Agent", "")[:255],
        )
    except Exception:
        pass

    return jsonify({"hits": hits, "found": found, "took_ms": took_ms})

/api/suggest (подсказки)

Подсказок обычно очень много (каждое нажатие клавиши), поэтому:

  • либо вообще не логировать каждую подсказку;
  • либо логировать только раз в N запросов;
  • либо логировать только финальный запрос, когда пользователь нажал Enter (что мы уже делаем через /search).

Для простоты можно или не логировать suggest вовсе, или сделать "тонкий" лог: только ip/q без hits.

import random
from search_logging import log_search, measure

SUGGEST_LOG_SAMPLE_RATE = 0.1 # логируем ~10% запросов подсказок


@measure
def typesense_suggest(params: dict):
    return ts_client.collections[COLLECTION_NAME].documents.search(params)


@app.route("/api/suggest")
def api_suggest():
    q = (request.args.get("q") or "").strip()
    if not q or len(q) < 2:
        return jsonify({"suggestions": []})

    try:
        limit = int(request.args.get("limit", "8"))
    except ValueError:
        limit = 8
    limit = max(1, min(limit, 20))

    search_params = {
        "q": q,
        "query_by": "title",
        "per_page": limit,
        "page": 1,
        "include_fields": "id,slug,title",
        "query_by_weights": "1",
        "prefix": "true",
        "num_typos": 1,
        "min_len_1typo": 4,
        "min_len_2typo": 8,
    }

    (result, took_ms) = typesense_suggest(search_params)

    suggestions = []
    for hit in result.get("hits", []):
        doc = hit.get("document", {})
        suggestions.append({
            "id": doc.get("id"),
            "slug": doc.get("slug"),
            "title": doc.get("title"),
        })

    found = result.get("found", 0)

    # Сэмплируем логирование подсказок
    if random.random() < SUGGEST_LOG_SAMPLE_RATE:
        try:
            log_search(
                q=q,
                hits=found,
                took_ms=took_ms,
                source="suggest",
                ip=request.remote_addr,
                user_agent=request.headers.get("User-Agent", "")[:255],
            )
        except Exception:
            pass

    return jsonify({"suggestions": suggestions, "found": found})

Простая аналитика: SQL-запросы

Теперь, когда лог есть, можно быстро собрать полезные отчёты.

Топ-50 запросов за последние 7 дней

SELECT q, COUNT(*) AS cnt
FROM search_log
WHERE created_at >= NOW() - INTERVAL 7 DAY
  AND source = 'page'
GROUP BY q
ORDER BY cnt DESC
LIMIT 50;

Использование:

  • это список запросов, под которые имеет смысл делать отдельные статьи, подборки или overrides.

Запросы без результатов

SELECT q, COUNT(*) AS cnt
FROM search_log
WHERE created_at >= NOW() - INTERVAL 30 DAY
  AND hits = 0
  AND source = 'page'
GROUP BY q
HAVING cnt >= 3
ORDER BY cnt DESC;

Интерпретация:

  • часто повторяющиеся запросы с hits = 0 => контентные дыры или неправильная морфология/синонимы;
  • под такие запросы можно:
    • написать отдельную страницу;
    • добавить синонимы/overrides (Часть 4);
    • проверить индексацию (страницы есть в БД, но нет в Typesense?).

Среднее число результатов и "слишком жирные" запросы

SELECT
    q,
    COUNT(*) AS cnt,
    AVG(hits) AS avg_hits,
    MAX(hits) AS max_hits
FROM search_log
WHERE created_at >= NOW() - INTERVAL 7 DAY
  AND source = 'page'
GROUP BY q
HAVING avg_hits > 100
ORDER BY avg_hits DESC
LIMIT 50;

Если по запросу в среднем >100 результатов, пользователю может быть сложно выбрать - хороший повод:

  • улучшить подсказки/фасеты;
  • добавить фильтры по тегам/датам (мы уже умеем это делать).

Медленные запросы

SELECT q, AVG(took_ms) AS avg_ms, MAX(took_ms) AS max_ms, COUNT(*) AS cnt
FROM search_log
WHERE created_at >= NOW() - INTERVAL 7 DAY
GROUP BY q
HAVING avg_ms > 300
ORDER BY avg_ms DESC
LIMIT 50;

Так можно найти:

  • запросы, которые стабильно тормозят (например, слишком широкие фильтры);
  • моменты просадки (если вдруг индекс раздулся или сервер не тянет).

Мини-дашборд /admin/search-analytics

Добавим простую админ-страницу (без полноценной авторизации, её оставим на совесть):

# admin_views.py
from flask import Blueprint, render_template
from db import get_connection

bp_admin = Blueprint("admin", __name__, url_prefix="/admin")


@bp_admin.route("/search-analytics")
def search_analytics():
    conn = get_connection()
    try:
        with conn.cursor() as cursor:
            # Топ запросов за 7 дней
            cursor.execute(
                """
                SELECT q, COUNT(*) AS cnt
                FROM search_log
                WHERE created_at >= NOW() - INTERVAL 7 DAY
                  AND source = 'page'
                GROUP BY q
                ORDER BY cnt DESC
                LIMIT 20
                """
            )
            top_queries = cursor.fetchall()

            # Запросы без результатов
            cursor.execute(
                """
                SELECT q, COUNT(*) AS cnt
                FROM search_log
                WHERE created_at >= NOW() - INTERVAL 30 DAY
                  AND hits = 0
                  AND source = 'page'
                GROUP BY q
                HAVING cnt >= 3
                ORDER BY cnt DESC
                LIMIT 20
                """
            )
            zero_queries = cursor.fetchall()

            # Среднее время ответа
            cursor.execute(
                """
                SELECT
                    DATE(created_at) AS d,
                    AVG(took_ms) AS avg_ms,
                    COUNT(*) AS cnt
                FROM search_log
                WHERE created_at >= NOW() - INTERVAL 14 DAY
                GROUP BY d
                ORDER BY d DESC
                """
            )
            daily_speed = cursor.fetchall()
    finally:
        conn.close()

    return render_template(
        "admin/search_analytics.html",
        top_queries=top_queries,
        zero_queries=zero_queries,
        daily_speed=daily_speed,
    )

Регистрация bluepring’а в основном app.py:

# app.py
from admin_views import bp_admin

app.register_blueprint(bp_admin)

Шаблон templates/admin/search_analytics.html

Простой Bootstrap-вид:

{% extends "base.html" %}

{% block content %}
<div class="container mt-4">
    <h1>Поисковая аналитика</h1>

    <h3 class="mt-4">Топ запросов за 7 дней</h3>
    <table class="table table-sm table-striped">
        <thead>
        <tr>
            <th>Запрос</th>
            <th>Количество</th>
        </tr>
        </thead>
        <tbody>
        {% for row in top_queries %}
            <tr>
                <td>{{ row.q }}</td>
                <td>{{ row.cnt }}</td>
            </tr>
        {% endfor %}
        </tbody>
    </table>

    <h3 class="mt-4">Запросы без результатов (≥3 раза за 30 дней)</h3>
    <table class="table table-sm table-striped">
        <thead>
        <tr>
            <th>Запрос</th>
            <th>Количество</th>
        </tr>
        </thead>
        <tbody>
        {% for row in zero_queries %}
            <tr>
                <td>{{ row.q }}</td>
                <td>{{ row.cnt }}</td>
            </tr>
        {% endfor %}
        </tbody>
    </table>

    <h3 class="mt-4">Среднее время ответа по дням</h3>
    <table class="table table-sm table-striped">
        <thead>
        <tr>
            <th>Дата</th>
            <th>Среднее время, мс</th>
            <th>Запросов</th>
        </tr>
        </thead>
        <tbody>
        {% for row in daily_speed %}
            <tr>
                <td>{{ row.d }}</td>
                <td>{{ row.avg_ms|round(1) }}</td>
                <td>{{ row.cnt }}</td>
            </tr>
        {% endfor %}
        </tbody>
    </table>
</div>
{% endblock %}

На слабом сервере:

  • держи отчёты максимально простыми (не строить сложные графики прямо в запросах);
  • по необходимости можно лимитировать период (например, последние 30 дней).

Чистка логов и аккуратность

Лог может расти бесконечно, так что:

  • раз в день/неделю можно чистить старые строки:
    DELETE FROM search_log
WHERE created_at < NOW() - INTERVAL 90 DAY;
  • если трафик большой - подумать про партиционирование или отдельную базу под аналитику (но для сайта на 100k страниц это, скорее всего, не нужно).

И важно:

  • не логируй чувствительные данные (логинов, email’ов, телефонов) - только сам запрос.

Что дальше делать с этими данными

Когда есть поисковая аналитика, можно:

  1. По популярным запросам:
    • делать подборки/лендинги;
    • вешать overrides, чтобы нужные страницы всегда были выше.
  2. По запросам без результатов:
    • заводить синонимы;
    • переиндексировать недостающие страницы;
    • писать новый контент.
  3. По медленным запросам:
    • подрезать per_page, фасеты и подсветку;
    • при необходимости - выносить Typesense на отдельный сервер (мы уже обсуждали это в Ч.5).