本地全文搜索工具

Python 倒排索引 + 命令行交互，在文档目录里秒级检索关键词。

场景：本地文档检索技术栈：Python 来源：GitHub 参考

在线演示

文档集（每行一篇，格式：标题 :: 正文）

查询词（支持中文二字切分与英文单词）

返回条数

0文档数

0词项数

0命中文档

0 ms检索耗时

等待操作：载入示例语料或粘贴自己的文档集，先“构建索引”，再输入查询词“搜索”。

项目简介

当文件散落在多个目录、用资源管理器翻到眼花时，一个本地全文搜索工具就能解决。本页实现基于倒排索引的迷你搜索引擎：首次扫描建索引，之后关键词查询 O(1) 定位文档；纯 Python 标准库，无第三方依赖。

代码示例：倒排索引构建

import re
import json
from pathlib import Path

TOKEN = re.compile(r"[\u4e00-\u9fff]|[a-zA-Z0-9_]+")

def tokenize(text: str) -> list[str]:
    # 中文按单字索引、英文按词索引，兼顾检索与内存
    return [t.lower() for t in TOKEN.findall(text)]

def build_index(root: Path) -> dict[str, list[int]]:
    index: dict[str, list[int]] = {}
    docs: list[str] = []
    for p in sorted(root.rglob("*")):
        if not p.is_file() or p.suffix.lower() in {".pyc", ".png", ".jpg"}:
            continue
        try:
            text = p.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        doc_id = len(docs)
        docs.append(str(p))
        seen: set[str] = set()
        for tok in tokenize(text):
            if tok not in seen:
                seen.add(tok)
                index.setdefault(tok, []).append(doc_id)
    return index, docs

def save(index: dict, docs: list[str], out: Path) -> None:
    out.write_text(json.dumps({"index": index, "docs": docs}, ensure_ascii=False), encoding="utf-8")

代码示例：查询与命令行入口

import sys

def search(index: dict, docs: list[str], query: str) -> list[str]:
    toks = tokenize(query)
    if not toks:
        return []
    hits: set[int] | None = None
    for t in toks:
        ids = set(index.get(t, []))
        hits = ids if hits is None else (hits & ids)  # 全部关键词命中
    return [docs[i] for i in sorted(hits)] if hits else []

def main() -> int:
    if len(sys.argv) < 2:
        print("usage: local_search.py <keyword> [root_dir]")
        return 1
    root = Path(sys.argv[2]) if len(sys.argv) > 2 else Path.cwd()
    index_path = root / ".search_index.json"
    if not index_path.exists():
        print("building index...")
        index, docs = build_index(root)
        save(index, docs, index_path)
    data = json.loads(index_path.read_text(encoding="utf-8"))
    for hit in search(data["index"], data["docs"], sys.argv[1])[:20]:
        print(hit)
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

操作步骤

首次运行 python local_search.py 关键词目录 会自动扫描目录并生成 .search_index.json。
之后同目录查询直接命中索引，秒级返回文件路径列表。
文档更新后可重建索引文件，或自行实现增量更新（记录 mtime）。

完整可运行工程

可直接运行零第三方依赖Python 3.8+

完整工程（非片段）：mini_search.py 实现「分词 → 倒排表 → TF-IDF 打分 → 排序输出」，页面演示与命令行共用同一套分词与打分公式，因此演示结果可直接用 CLI 复现。

local-search/
  ├── mini_search.py     # 分词 / 建索引 / 检索
  ├── test_mini_search.py # 最小用例回归
  ├── requirements.txt
  └── README.md

mini_search.py

核心实现：tokenize / Index.build / Index.search / 索引 JSON 序列化，含 --selftest。

python

#!/usr/bin/env python3
"""本地倒排索引检索（零第三方依赖，仅标准库）

用法：
    python mini_search.py --docs ./docs --query "备份 缓存" --top 5
    python mini_search.py --docs notes.md --query index --json
    python mini_search.py --docs ./docs --index-save index.json
    python mini_search.py --index-load index.json --query "倒排"
    python mini_search.py --selftest
"""
import argparse
import json
import math
import re
import sys
from pathlib import Path

EN = re.compile(r"[a-z0-9_]+")
HAN = re.compile(r"[\u4e00-\u9fff]+")
SUFFIX = {".txt", ".md", ".markdown", ".log", ".csv", ".json"}


def tokenize(text):
    """英文/数字按单词切分；中文按 2-gram（单字短语保留自身）。"""
    low = text.lower()
    out = EN.findall(low)
    for seq in HAN.findall(low):
        if len(seq) == 1:
            out.append(seq)
        else:
            out.extend(seq[i:i + 2] for i in range(len(seq) - 1))
    return out


class Index:
    def __init__(self):
        self.docs = []        # [{title, body, path}]
        self.postings = {}    # term -> {doc_id: tf}
        self.avg_len = 0.0

    def build(self, docs):
        self.docs, self.postings, total = [], {}, 0
        for i, d in enumerate(docs):
            body = d.get("body", "")
            toks = tokenize(d.get("title", "") + " " + body)
            total += len(toks)
            tf = {}
            for t in toks:
                tf[t] = tf.get(t, 0) + 1
            self.docs.append({"id": i, "title": d.get("title") or "文档 %d" % (i + 1),
                              "body": body, "path": d.get("path", ""), "len": len(toks)})
            for t, c in tf.items():
                self.postings.setdefault(t, {})[i] = c
        self.avg_len = total / len(self.docs) if self.docs else 0.0
        return self

    def search(self, query, top=5):
        terms = tokenize(query)
        scores, matched = {}, {}
        for t in terms:
            post = self.postings.get(t)
            if not post:
                continue
            idf = math.log(1 + len(self.docs) / len(post))
            for doc_id, tf in post.items():
                scores[doc_id] = scores.get(doc_id, 0.0) + (tf / (tf + 1.2)) * idf
                matched.setdefault(doc_id, []).append(t)
        out = []
        for doc_id, s in scores.items():
            d = self.docs[doc_id]
            norm = 1 / math.sqrt(1 + d["len"] / (self.avg_len or 1))
            out.append({"doc": d["title"], "path": d["path"], "score": round(s * norm, 4),
                        "matched": sorted(set(matched[doc_id])), "snippet": d["body"][:160]})
        out.sort(key=lambda x: (-x["score"], x["doc"]))
        return out[:top]

    def to_json(self):
        return json.dumps({"docs": self.docs, "postings": self.postings, "avg_len": self.avg_len},
                          ensure_ascii=False)

    @classmethod
    def from_json(cls, text):
        obj = json.loads(text)
        idx = cls()
        idx.docs = obj["docs"]
        idx.postings = {k: {int(i): c for i, c in v.items()} for k, v in obj["postings"].items()}
        idx.avg_len = obj.get("avg_len", 0.0)
        return idx


def load_docs(path):
    p = Path(path)
    if not p.exists():
        raise SystemExit("路径不存在：%s" % p)
    files = [p] if p.is_file() else sorted(f for f in p.rglob("*") if f.is_file() and f.suffix.lower() in SUFFIX)
    if not files:
        raise SystemExit("目录中没有可索引的文本文件（支持 %s）。" % ", ".join(sorted(SUFFIX)))
    docs = []
    for f in files:
        body = f.read_text(encoding="utf-8", errors="replace")
        title = f.stem if f.is_file() else f.name
        first_line = next((l.strip("# ").strip() for l in body.splitlines() if l.strip()), "")
        docs.append({"title": title, "body": body, "path": str(f), "first_line": first_line})
    return docs


DEMO_DOCS = [
    {"title": "备份轮转", "body": "用 SQLite 记录备份元数据，按保留策略归档与删除，支持 dry-run。"},
    {"title": "倒排索引", "body": "倒排表把词项映射到文档，检索时按 TF-IDF 累加打分。"},
    {"title": "主题令牌", "body": "用 CSS 变量统一管理颜色，浅色与深色主题只切换令牌值。"},
]


def selftest():
    idx = Index().build(DEMO_DOCS)
    hits = idx.search("备份", top=3)
    assert hits and hits[0]["doc"] == "备份轮转", hits
    hits2 = idx.search("TF-IDF", top=3)
    assert hits2 and hits2[0]["doc"] == "倒排索引", hits2
    assert idx.search("完全不存在的词", top=3) == []
    assert tokenize("备份轮转") == ["备份", "份轮", "轮转"]
    assert tokenize("Index Build") == ["index", "build"]
    idx2 = Index.from_json(idx.to_json())
    assert idx2.search("令牌", top=1)[0]["doc"] == "主题令牌"
    print("selftest OK：中英分词 / 检索排序 / 空结果 / 索引序列化 用例通过")


def main(argv=None):
    ap = argparse.ArgumentParser(description="本地倒排索引检索（TF-IDF）")
    ap.add_argument("--docs", help="文本文件或目录（递归索引 .txt/.md/.log/.csv/.json）")
    ap.add_argument("--query", help="查询词")
    ap.add_argument("--top", type=int, default=5)
    ap.add_argument("--index-save", help="把索引写入 JSON 文件")
    ap.add_argument("--index-load", help="从 JSON 文件加载索引")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args(argv)

    if a.selftest:
        selftest()
        return 0

    if a.index_load:
        idx = Index.from_json(Path(a.index_load).read_text(encoding="utf-8"))
    elif a.docs:
        idx = Index().build(load_docs(a.docs))
    else:
        ap.error("请给出 --docs 或 --index-load")

    if a.index_save:
        Path(a.index_save).write_text(idx.to_json(), encoding="utf-8")
        print("索引已保存：%s（%d 篇文档 / %d 个词项）" % (a.index_save, len(idx.docs), len(idx.postings)))

    if not a.query:
        if a.index_save:
            return 0
        ap.error("请给出 --query，或仅用 --index-save 建库")

    hits = idx.search(a.query, a.top)
    if a.json:
        print(json.dumps({"query": a.query, "hits": hits}, ensure_ascii=False, indent=2))
    else:
        print("查询「%s」命中 %d 篇：" % (a.query, len(hits)))
        for i, h in enumerate(hits, 1):
            print("  %d. [%.4f] %s%s" % (i, h["score"], h["doc"], ("  (%s)" % h["path"]) if h["path"] else ""))
            print("     命中词项：%s" % ", ".join(h["matched"]))
            print("     片段：%s" % h["snippet"].replace("\n", " ")[:120])
    return 0


if __name__ == "__main__":
    sys.exit(main())

test_mini_search.py

unittest 回归：中英分词、排序正确性、空结果、索引往返、目录加载。

python

"""mini_search.py 最小用例回归（标准库 unittest）"""
import json
import tempfile
import unittest
from pathlib import Path

from mini_search import DEMO_DOCS, Index, load_docs, tokenize


class TestMiniSearch(unittest.TestCase):
    def setUp(self):
        self.idx = Index().build(DEMO_DOCS)

    def test_tokenize_chinese_bigram(self):
        self.assertEqual(tokenize("备份轮转"), ["备份", "份轮", "轮转"])

    def test_tokenize_english(self):
        self.assertEqual(tokenize("Index Build"), ["index", "build"])

    def test_search_ranks_relevant_doc(self):
        self.assertEqual(self.idx.search("备份", 3)[0]["doc"], "备份轮转")
        self.assertEqual(self.idx.search("TF-IDF", 3)[0]["doc"], "倒排索引")

    def test_search_empty(self):
        self.assertEqual(self.idx.search("完全不存在的词", 3), [])

    def test_index_roundtrip(self):
        idx2 = Index.from_json(self.idx.to_json())
        self.assertEqual(idx2.search("令牌", 1)[0]["doc"], "主题令牌")

    def test_load_docs_from_temp_dir(self):
        with tempfile.TemporaryDirectory() as td:
            Path(td, "a.md").write_text("# 缓存策略\nLRU 与失效策略。", encoding="utf-8")
            Path(td, "b.txt").write_text("倒排索引与打分。", encoding="utf-8")
            docs = load_docs(td)
            self.assertEqual(len(docs), 2)
            idx = Index().build(docs)
            self.assertTrue(idx.search("缓存"))


if __name__ == "__main__":
    unittest.main(verbosity=2)

requirements.txt

依赖清单：仅标准库。

text

# 本工程仅使用 Python 标准库（argparse / json / math / re / pathlib / unittest）。
# 不需要安装任何第三方包；此文件保留是为了统一 `pip install -r requirements.txt` 流程。

README.md

参数、分词策略、打分公式与调参位置。

markdown

# 本地倒排索引检索（local-search）

把一组本地文本文件建成倒排索引，用 TF-IDF 打分检索，**零第三方依赖**。
不是把整个文件按 `grep` 扫一遍——先建索引再查，重复查询的开销与语料规模无关。

## 文件说明

| 文件 | 作用 |
|---|---|
| `mini_search.py` | 分词 / 建索引 / 检索核心 + CLI 入口 |
| `test_mini_search.py` | unittest 最小用例回归 |
| `requirements.txt` | 依赖清单（本工程无第三方依赖） |

## 运行

```bash
python -m pip install -r requirements.txt          # 无第三方依赖，可跳过
python mini_search.py --docs ./docs --query "备份 缓存" --top 5
python mini_search.py --docs notes.md --query index --json
python mini_search.py --docs ./docs --index-save index.json     # 建库后复用
python mini_search.py --index-load index.json --query "倒排"
python mini_search.py --selftest
python test_mini_search.py
```

## 分词策略

| 语言 | 切分方式 | 示例 |
|---|---|---|
| 中文 | 2-gram（单字短语保留自身） | 备份轮转 → 备份 / 份轮 / 轮转 |
| 英文 / 数字 | `\w+` 单词 | Index Build → index / build |

2-gram 的好处是不需要词典就能召回中文子串，代价是词项数变多；语料再大一些可换成结巴分词。

## 打分公式

```
score(doc) = Σ_term [ tf/(tf+1.2) × log(1 + N/df) ] × 1/√(1 + len/avg_len)
```

`tf/(tf+1.2)` 抑制长文里的词频堆积，`log(1+N/df)` 压制高频词，末尾长度归一避免长文档占优。
公式改在 `mini_search.py` 的 `Index.search()` 里，改完直接 `--selftest` 验证。

运行标注

可直接运行：保存 4 个文件到同一目录，执行 python mini_search.py --docs 你的目录 --query "关键词"。
需按需修改的配置项：无必改项。语料目录通过 --docs 传入（支持目录递归，索引 .txt/.md/.log/.csv/.json）；返回条数用 --top。不改代码即可直接使用。
大语料建议先建库再用：--index-save index.json，之后 --index-load index.json --query ... 秒级响应。
自测：python mini_search.py --selftest（或 python test_mini_search.py），6 个用例应全部通过。

复制结果为 # ===== 文件名 ===== 分隔的多文件文本，按文件名分别保存即可；也可逐块用右上角“复制代码”单独取用。

来源参考

实现与部署所依据的官方文档 / 仓库（均可直接访问）：

Python 官方文档 · re（分词正则） Python 官方文档 · math（log / sqrt 打分计算） Stanford IR Book · 倒排索引与索引构建（第 4 章，官方在线版） scikit-learn 官方文档 · TfidfVectorizer（TF-IDF 权重计算实现）