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() 时,框架会执行一次完整的索引重建:
-
数据库准备:创建一个临时的 SQLite 数据库,并根据你的模式配置 FTS5 表
-
文档收集:使用配置的字段映射和过滤器查询所有指定的 DocType
-
文档处理:对于每个文档:
-
根据
INDEXABLE_DOCTYPES配置提取和映射字段 -
使用 BeautifulSoup 清理 HTML 内容以提取纯文本
-
如果覆盖了
prepare_document(),则应用自定义的文档准备逻辑 -
验证必填字段(标题、内容)是否存在
-
-
批量插入:将处理后的文档分批插入 FTS5 索引以提高性能
-
词汇表构建:从所有索引文本中构建拼写纠正字典
-
原子替换:以原子方式用新数据库替换现有索引数据库
单个文档索引
使用 index_doc() 或 remove_doc() 进行实时更新:
-
单个文档处理:使用相同的字段映射逻辑检索并处理一个文档
-
增量更新:通过插入、更新或删除特定文档来更新现有的 FTS5 索引
-
词汇表更新:使用文档中的新术语更新拼写字典
2. 搜索过程
当用户使用 search() 执行搜索时,框架会执行以下步骤:
-
权限过滤:调用
get_search_filters()来确定当前用户可以访问哪些文档 -
查询预处理:
-
验证搜索查询不为空
-
将用户提供的过滤器与权限过滤器合并
-
-
拼写纠正:
-
对照词汇字典分析查询词条
-
使用三元组相似度来为拼写错误的单词提供纠正建议
-
使用纠正后的词条扩展原始查询
-
-
FTS5 查询执行:
-
构建一个兼容 FTS5 的查询字符串
-
对 SQLite 数据库执行全文搜索
-
应用元数据过滤器(状态、所有者、项目等)
-
检索带有 BM25 分数的原始结果
-
-
结果处理:
-
自定义评分:应用评分管道来计算最终的相关性分数
-
基础 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"} -
必填字段:
title和content字段必须存在或显式映射(例如{"title": "subject"}) -
自动添加的字段:
doctype和name会自动包含 -
修改字段:如果在任何文档类型配置中使用,则会自动添加。用于近期加权 – 如果您想使用不同的时间戳字段(如
creation或last_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
See our code examples for details.
| Method | POST |
, , , 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']
框架自动完成的工作:
-
迁移后索引构建:运行 bench migrate 后自动构建搜索索引
-
定期索引验证:每 15 分钟检查一次索引是否存在,如果缺失则重新构建
-
实时文档更新:在文档生命周期事件(插入、更新、删除)中自动调用 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 装饰器将函数标记为评分函数。
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']
迁移后索引构建:运行 bench migrate 后自动构建搜索索引
定期索引验证:每 15 分钟检查一次索引是否存在,如果缺失则重新构建
实时文档更新:在文档生命周期事件(插入、更新、删除)中自动调用 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"],
}
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 装饰器将函数标记为评分函数。