SQLite 搜索

SQLite 搜索是 Frappe 应用程序的一个全文搜索框架,利用 SQLite 的 FTS5(全文搜索)引擎提供高级搜索功能。它提供了拼写纠正、基于时间的时效性评分、自定义排名、权限感知过滤和可扩展的评分管道等功能。

目录

  • 快速开始

  • 工作原理

  • 配置

  • 功能与自定义

  • API 参考

快速开始

1. 创建搜索类

通过继承 SQLiteSearch 来创建搜索实现:


# my_app/search.py

from frappe.search.sqlite_search import SQLiteSearch

class MyAppSearch(SQLiteSearch):

    # Database file name

    INDEX_NAME = "my_app_search.db"

    # Define the search schema

    INDEX_SCHEMA = {

        "metadata_fields": ["project", "owner", "status"],

        "tokenizer": "unicode61 remove_diacritics 2 tokenchars '-_'",

    }

    # Define which doctypes to index and their field mappings

    INDEXABLE_DOCTYPES = {

        "Task": {

            "fields": ["name", {"title": "subject"}, {"content": "description"}, "modified", "project", "owner", "status"],

        },

        "Issue": {

            "fields": ["name", "title", "description", {"modified": "last_updated"}, "project", "owner"],

            "filters": {"status": ("!=", "Closed")},  # Only index non-closed issues

        },

    }

    def get_search_filters(self):

        """Return permission filters for current user"""

        # Get projects accessible to current user

        accessible_projects = frappe.get_all(

            "Project",

            filters={"owner": frappe.session.user},

            pluck="name"

        )

        if not accessible_projects:

            return {"project": []}  # No access

        return {"project": accessible_projects}

2. 注册搜索类

将你的搜索类添加到 hooks.py 文件中:


# my_app/hooks.py

sqlite_search = ['my_app.search.MyAppSearch']

3. 创建 API 端点

创建一个白名单方法来暴露搜索功能:


# my_app/api.py

import frappe

from my_app.search import MyAppSearch

@frappe.whitelist()

def search(query, filters=None):

    search = MyAppSearch()

    result = search.search(query, filters=filters)

    return result

4. 构建索引

通过编程方式或控制台构建搜索索引:


from my_app.search import MyAppSearch

search = MyAppSearch()

search.build_index()

工作原理

1. 索引过程

完整索引构建

当你调用 build_index() 时,框架会执行一次完整的索引重建:

  1. 数据库准备:创建一个临时的 SQLite 数据库,并根据你的模式配置 FTS5 表

  2. 文档收集:使用配置的字段映射和过滤器查询所有指定的 DocType

  3. 文档处理:对于每个文档:

    • 根据 INDEXABLE_DOCTYPES 配置提取和映射字段

    • 使用 BeautifulSoup 清理 HTML 内容以提取纯文本

    • 如果覆盖了 prepare_document(),则应用自定义的文档准备逻辑

    • 验证必填字段(标题、内容)是否存在

  4. 批量插入:将处理后的文档分批插入 FTS5 索引以提高性能

  5. 词汇表构建:从所有索引文本中构建拼写纠正字典

  6. 原子替换:以原子方式用新数据库替换现有索引数据库

单个文档索引

使用 index_doc()remove_doc() 进行实时更新:

  1. 单个文档处理:使用相同的字段映射逻辑检索并处理一个文档

  2. 增量更新:通过插入、更新或删除特定文档来更新现有的 FTS5 索引

  3. 词汇表更新:使用文档中的新术语更新拼写字典

2. 搜索过程

当用户使用 search() 执行搜索时,框架会执行以下步骤:

  1. 权限过滤:调用 get_search_filters() 来确定当前用户可以访问哪些文档

  2. 查询预处理

    • 验证搜索查询不为空

    • 将用户提供的过滤器与权限过滤器合并

  3. 拼写纠正

    • 对照词汇字典分析查询词条

    • 使用三元组相似度来为拼写错误的单词提供纠正建议

    • 使用纠正后的词条扩展原始查询

  4. FTS5 查询执行

    • 构建一个兼容 FTS5 的查询字符串

    • 对 SQLite 数据库执行全文搜索

    • 应用元数据过滤器(状态、所有者、项目等)

    • 检索带有 BM25 分数的原始结果

  5. 结果处理

    • 自定义评分:应用评分管道来计算最终的相关性分数

      • 基础 BM25 分数处理

      • 标题匹配加权(精确匹配和部分匹配)

      • 基于文档年龄的时效性加权

      • 自定义评分函数(特定 DocType、基于优先级等)

    • 排名:按最终分数对结果进行排序并分配排名位置

    • 内容格式化:生成内容摘要并高亮匹配词条

配置

INDEX_SCHEMA

定义搜索索引的结构:


INDEX_SCHEMA = {

    # Text fields that will be searchable (defaults to ["title", "content"])

    "text_fields": ["title", "content"],

    # Metadata fields stored alongside text content for filtering

    "metadata_fields": ["project", "owner", "status", "priority"],

    # FTS5 tokenizer configuration

    "tokenizer": "unicode61 remove_diacritics 2 tokenchars '-_@.'"

}

INDEXABLE_DOCTYPES

指定要索引的 DocType 以及如何映射其字段:


INDEXABLE_DOCTYPES = {

    "Task": {

        # Field mapping

        "fields": [

            "name",

            {"title": "subject"},        # Maps subject field to title

            {"content": "description"},  # Maps description field to content

            {"modified": "creation"},    # Use creation instead of modified for recency boost

            "project",

            "owner"

        ],

        # Optional filters to limit which records are indexed

        "filters": {

            "status": ("!=", "Cancelled"),

            "docstatus": ("!=", 2)

        }

    }

}

字段映射规则

  • 字符串字段:直接映射 "field_name"

  • 别名字段:字典映射 {"schema_field": "doctype_field"}

  • 必填字段titlecontent 字段必须存在或显式映射(例如 {"title": "subject"}

  • 自动添加的字段doctypename 会自动包含

  • 修改字段:如果在任何文档类型配置中使用,则会自动添加。用于近期加权 – 如果您想使用不同的时间戳字段(如 creationlast_updated),请使用 {"modified": "creation"} 将其映射到 modified

功能与自定义

权限过滤

实现 get_search_filters() 来控制访问权限:


def get_search_filters(self):

    """Return filters based on user permissions"""

    user = frappe.session.user

    if user == "Administrator":

        return {}  # No restrictions

    # Example: User can only see their own and public documents

    return {

        "owner": user,

        "status": ["Active", "Published"]

    }

自定义评分

创建自定义评分函数以影响搜索相关性:


class MyAppSearch(SQLiteSearch):

    ...

    @SQLiteSearch.scoring_function

    def *get*priority_boost(self, row, query, query_words):

        """Boost high-priority items"""

        priority = row.get("priority", "Medium")

        if priority == "High":

            return 1.5

        if priority == "Medium":

            return 1.1

        return 1.0

近期加权

框架使用 modified 字段自动提供基于时间的近期加权:


# The modified field is used for calculating document age

# Recent documents get higher scores:

# - Last 24 hours: 1.8x boost

# - Last 7 days: 1.5x boost

# - Last 30 days: 1.2x boost

# - Last 90 days: 1.1x boost

# - Older documents: gradually decreasing boost

# If your doctype uses a different timestamp field, map it to modified:

INDEXABLE_DOCTYPES = {

    "GP Discussion": {

        "fields": ["name", "title", "content", {"modified": "last_post_at"}, "project"],

    },

    "Article": {

        "fields": ["name", "title", "content", {"modified": "published_date"}, "category"],

    }

}

文档准备

覆盖 prepare_document() 以进行自定义文档处理:


def prepare_document(self, doc):

    """Custom document preparation"""

    document = super().prepare_document(doc)

    if not document:

        return None

    # Add computed fields

    if doc.doctype == "Task":

        # Combine multiple fields into content

        content_parts = [

            doc.description or "",

            doc.notes or "",

            "\n".join([comment.content for comment in doc.get("comments", [])])

        ]

        document["content"] = "\n".join(filter(None, content_parts))

        # set fields that might be stored in another table

        document["category"] = get_category_for_task(doc)

    return document

拼写纠正

框架内置了使用三元组相似度的拼写纠正功能:


# Spelling correction happens automatically

search_result = search.search("projetc managment")  # Will find "project management"

# Access correction information

print(search_result["summary"]["corrected_words"])

# Output: {"projetc": "project", "managment": "management"}

内容处理

HTML 内容会自动使用 BeautifulSoup 进行清理和处理:


# Complex HTML content like this:

html_content = """

<div class="article">

<h1>API Documentation</h1>

<p>Learn how to integrate with our <a href="/api">REST API</a>.</p>

    <img src="/images/api-flow.png" alt="API workflow diagram">

<ul>

<li><strong>Authentication:</strong> Use <code>Bearer tokens

  • Rate limiting: 1000 requests/hour
  • See our code examples for details.
    Method POST
    analytics.track('page_view'); .hidden { display: none; }
    """ # Is automatically converted to clean, searchable plain text: """ API Documentation Learn how to integrate with our REST API. Authentication: Use Bearer tokens Rate limiting: 1000 requests/hour See our code examples for details. Method POST """ # The cleaning process: # 1. Removes all HTML tags (
    ,

    , , , etc.) # 2. Strips out scripts, styles, and non-content elements # 3. Extracts link text while removing href URLs # 4. Normalizes whitespace and line breaks
    
    results = search.search("project update", title_only=True)
    

    高级过滤

    
    accessible_projects = ['PROJ001', 'PROJ002', ...]
    
    filters = {
    
        "project": accessible_projects,     # Multiple values (IN clause)
    
        "owner": current_user,              # Single value (= clause)
    
    }
    
    results = search.search("bug fix", filters=filters)
    

    自动索引处理

    当您注册搜索类时,框架会自动处理索引的构建和维护:

    
    # hooks.py
    
    sqlite_search = ['my_app.search.MyAppSearch']
    

    框架自动完成的工作:

    1. 迁移后索引构建:运行 bench migrate 后自动构建搜索索引

    2. 定期索引验证:每 15 分钟检查一次索引是否存在,如果缺失则重新构建

    3. 实时文档更新:在文档生命周期事件(插入、更新、删除)中自动调用 index_doc()remove_doc(),适用于您在 INDEXABLE_DOCTYPES 中定义的所有文档类型

    手动索引处理

    如果您希望手动控制索引的生命周期,可以通过不在 sqlite_search 钩子中注册搜索类来退出自动索引处理。

    
    from my_app.search import MyAppSearch
    
    def build_index_in_background():
    
        """Manually trigger background index building"""
    
        search = MyAppSearch()
    
        if search.is_search_enabled() and not search.index_exists():
    
            frappe.enqueue("my_app.search.build_index", queue="long")
    
    # hooks.py
    
    scheduler_events = {
    
        # Custom scheduler (if you want different timing)
    
        "daily": ["my_app.search.build_index_if_not_exists"],
    
    }
    

    API 参考

    search(query, title_only=False, filters=None)

    返回格式化结果的主要搜索方法。

    参数:

    • query(字符串):搜索查询文本

    • title_only(布尔值):仅在标题字段中搜索

    • filters(字典):要应用的附加过滤器

    返回值:

    
    {
    
        "results": [
    
            {
    
                "doctype": "Task",
    
                "name": "TASK-001",
    
                "title": "Fix login bug",
    
                "content": "User cannot login after password reset...",
    
                "score": 0.85,
    
                "original_rank": 3, # original bm25 rank
    
                "rank": 1, # modified rank after custom scoring pipeline
    
                # ... other metadata fields
    
            }
    
        ],
    
        "summary": {
    
            "duration": 0.023,
    
            "total_matches": 15,
    
            "returned_matches": 15,
    
            "corrected_words": {"loggin": "login"},
    
            "corrected_query": "Fix login bug",
    
            "title_only": False,
    
            "filtered_matches": 15,
    
            "applied_filters": {"status": ["Open"]}
    
        }
    
    }
    

    build_index()

    从头构建完整的搜索索引。

    index_doc(doctype, docname)

    索引单个文档。

    remove_doc(doctype, docname)

    从索引中移除单个文档。

    is_search_enabled()

    检查搜索是否已启用(覆盖以添加禁用逻辑)。

    index_exists()

    检查搜索索引是否存在。

    get_search_filters()

    必须由子类实现。返回当前用户的过滤器。

    返回值:

    
    {
    
        "field_name": "value",           # Single value
    
        "field_name": ["val1", "val2"],  # Multiple values
    
    }
    

    scoring_function()

    使用 @SQLiteSearch.scoring_function 装饰器将函数标记为评分函数。