paperless-ngx/src/documents/search/_schema.py

from __future__ import annotations

import json
import logging
import shutil
from typing import TYPE_CHECKING
from typing import Final
from typing import cast

import tantivy
from django.conf import settings
from whoosh_compat import FieldKind

from documents.search._fields import PUBLIC_FIELDS

if TYPE_CHECKING:
    from pathlib import Path

logger = logging.getLogger("paperless.search")

# v1 - Initial tantivy schema format
# v2 - build_schema() derived from PUBLIC_FIELDS, changing the field declaration
#      order, and the write-only correspondent/document_type/storage_path/tag id
#      columns dropped. tantivy compares schemas by ordered field list, so an
#      index built by v1 rejects every write against the v2 schema.
SCHEMA_VERSION: Final[int] = 2


def build_schema() -> tantivy.Schema:
    """
    Build the Tantivy schema for the paperless document index.

    Creates a comprehensive schema supporting full-text search, filtering,
    sorting, and autocomplete functionality. Includes fields for document
    content, metadata, permissions, custom fields, and notes.

    Returns:
        Configured Tantivy schema ready for index creation
    """
    sb = tantivy.SchemaBuilder()

    sb.add_unsigned_field("id", stored=True, indexed=True, fast=True)

    for field in PUBLIC_FIELDS:
        if field.kind is FieldKind.TEXT:
            sb.add_text_field(field.name, stored=True, tokenizer_name="paperless_text")
        elif field.kind is FieldKind.KEYWORD:
            sb.add_text_field(field.name, stored=True, tokenizer_name="raw")
        elif field.kind is FieldKind.U64:
            sb.add_unsigned_field(
                field.name,
                stored=True,
                indexed=True,
                fast=field.fast,
            )
        elif field.kind in (FieldKind.DATE, FieldKind.DATETIME):
            sb.add_date_field(
                field.name,
                stored=True,
                indexed=True,
                fast=field.fast,
            )
        elif field.kind is FieldKind.JSON:
            sb.add_json_field(field.name, stored=True, tokenizer_name="paperless_text")
            if field.name == "notes":
                # Plain-text companion for snippet generation — tantivy's
                # SnippetGenerator does not support JSON fields. Schema-only,
                # no query-syntax meaning, not in PUBLIC_FIELDS.
                sb.add_text_field(
                    "notes_text",
                    stored=True,
                    tokenizer_name="paperless_text",
                )

    # Shadow sort fields - fast, not stored/indexed
    for field in ("title_sort", "correspondent_sort", "type_sort"):
        sb.add_text_field(
            field,
            stored=False,
            tokenizer_name="simple_analyzer",
            fast=True,
        )

    # CJK support - not stored, indexed only
    sb.add_text_field("bigram_content", stored=False, tokenizer_name="bigram_analyzer")
    sb.add_text_field("bigram_title", stored=False, tokenizer_name="bigram_analyzer")
    sb.add_text_field(
        "bigram_correspondent",
        stored=False,
        tokenizer_name="bigram_analyzer",
    )
    sb.add_text_field(
        "bigram_document_type",
        stored=False,
        tokenizer_name="bigram_analyzer",
    )
    sb.add_text_field("bigram_tag", stored=False, tokenizer_name="bigram_analyzer")

    # Simple substring search support for title/content - not stored, indexed only
    sb.add_text_field(
        "simple_title",
        stored=False,
        tokenizer_name="simple_search_analyzer",
    )
    sb.add_text_field(
        "simple_content",
        stored=False,
        tokenizer_name="simple_search_analyzer",
    )

    # Autocomplete prefix scan via terms_with_prefix, which walks the field's
    # term dictionary - so the field must be indexed (term dict), not stored.
    # The stored value is never read back, so storing it only wastes space.
    sb.add_text_field("autocomplete_word", stored=False, tokenizer_name="raw")

    # Permission filter columns, read by build_permission_filter.
    for field in ("owner_id", "viewer_id", "viewer_group_id"):
        sb.add_unsigned_field(field, stored=False, indexed=True, fast=True)

    return sb.build()


def needs_rebuild(index_dir: Path) -> bool:
    """
    Check if the search index needs rebuilding.

    Reads .index_settings.json to compare the stored schema version and
    search language against the current configuration. Returns True if the
    file is missing, unparsable, or either value mismatches.

    Args:
        index_dir: Path to the search index directory

    Returns:
        True if the index needs rebuilding, False if it's up to date
    """
    settings_file = index_dir / ".index_settings.json"
    if not settings_file.exists():
        return True
    try:
        data = json.loads(settings_file.read_text())
        if data.get("schema_version") != SCHEMA_VERSION:
            logger.info("Search index schema version mismatch - rebuilding.")
            return True
        if "language" not in data or data["language"] != settings.SEARCH_LANGUAGE:
            logger.info("Search index language changed - rebuilding.")
            return True
    except ValueError:
        return True
    return False


def wipe_index(index_dir: Path) -> None:
    """
    Delete all contents of the index directory to prepare for rebuild.

    Recursively removes all files and subdirectories within the index
    directory while preserving the directory itself.

    Args:
        index_dir: Path to the search index directory to clear
    """
    for child in index_dir.iterdir():
        if child.is_dir():
            shutil.rmtree(child)
        else:
            child.unlink()


def _write_sentinels(index_dir: Path) -> None:
    """Write .index_settings.json so the next index open can skip rebuilding."""
    settings_file = index_dir / ".index_settings.json"
    settings_file.write_text(
        json.dumps(
            {
                "schema_version": SCHEMA_VERSION,
                "language": settings.SEARCH_LANGUAGE,
            },
        ),
    )


def open_or_rebuild_index(index_dir: Path | None = None) -> tantivy.Index:
    """
    Open the Tantivy index, creating or rebuilding as needed.

    Checks if the index needs rebuilding due to schema version or language
    changes. If rebuilding is needed, wipes the directory and creates a fresh
    index with the current schema and configuration.

    Args:
        index_dir: Path to index directory (defaults to settings.INDEX_DIR)

    Returns:
        Opened Tantivy index (caller must register custom tokenizers)
    """
    if index_dir is None:
        index_dir = cast("Path", settings.INDEX_DIR)
    if not index_dir.exists():
        return tantivy.Index(build_schema())
    if needs_rebuild(index_dir):
        wipe_index(index_dir)
        idx = tantivy.Index(build_schema(), path=str(index_dir))
        _write_sentinels(index_dir)
        return idx
    return tantivy.Index.open(str(index_dir))