一步步带你搭建第一个 Easysearch 搜索服务

引言 #

你是否想过在自己的服务器上搭建一个功能强大的中文搜索引擎？比如搜索自己收藏的维基百科、文档库、或者知识库？

本文将手把手教你用 Python 和 Easysearch 搭建一个支持中文分词的私有维基百科搜索引擎。Easysearch 内置了中文分词的支持，这意味着我们不需要折腾复杂的中文分词配置，直接就能用！

在文章的最后，你将收获一下如下的私有搜索引擎：

$ python3.10 search.py 大学
============================================================
Easysearch 维基百科搜索
============================================================
...

找到 1 条结果

[1] Template:Potd/2006-1-2zh
    URL: https://zh.wikipedia.org/wiki/Template%3APotd/2006-1-2zh
    高亮:
      ...伦敦帝国学院成立于1907年，是一个专精于科学技术的<em>大学</em>，它还是联邦<em>大学</em>伦敦<em>大学</em>的一个加盟学院。虽然它的正式全名是的“帝国科学，技术和医学学院”，自2002年来，它通常使用的英文名称是伦敦帝国学院。中文中，它通常被称为伦敦帝国理工学院或简称为帝国理工。...

准备工作 #

在开始之前，请确认：

Easysearch 已在运行：假设你的 Easysearch 服务在 127.0.0.1:9200 运行，用户名为 admin，密码为 your_password（请根据实际情况修改）。
Python 3.10 环境：

$ python3.10 --version
Python 3.10.19

安装必要的 Python 库：

python3.10 -m pip install "elasticsearch<7.14" wikiextractor

注意：

- 必须使用 Python 3.10，因为 wikiextractor 库在后续 Python 版本中不能用。
- `elasticsearch<7.14`：因为 Easysearch 基于 Elasticsearch 7.10.2，使用更高版本的 Python SDK 可能会出现兼容性问题。

第一步：获取维基百科数据 #

中文维基百科定期发布完整的数据库转储文件。
下载链接：https://dumps.wikimedia.org/zhwiki/latest/zhwiki-latest-pages-articles.xml.bz2

# 下载数据（约 3GB+，下载时间较长）
wget https://dumps.wikimedia.org/zhwiki/latest/zhwiki-latest-pages-articles.xml.bz2

第二步：创建项目 #

创建一个 wiki_zh 的文件夹，将中文维基百科的数据放到文件夹下

$ mkdir wiki_zh
$ cd wiki_zh
$ ls -l
.rw-r--r--@     1 3.2G user staff 28 Jan 14:35  zhwiki-latest-pages-articles.xml.bz2

第三步：从 XML 中提取纯文本 JSON #

维基百科的原始数据是 XML 格式，包含大量的模板和标记。我们使用 wikiextractor 库提取干净的文本。

快速测试：提取 20 篇文章 #

使用我们提供的 Python 脚本，它利用 wikiextractor 的 Extractor 类进行文本清理：

python3.10 extract_sample.py

这个脚本会提取前 20 篇有效文章，输出到 wiki_data/AA/wiki_00.json。

脚本代码 (extract_sample.py)：

#!/usr/bin/env python3.10
"""
使用 wikiextractor 库提取少量维基百科文章样本。
利用 wikiextractor 的 Extractor 类进行文本清理。
"""

import bz2
import json
import re
import sys
from pathlib import Path
from io import StringIO

from wikiextractor.extract import Extractor

# ===============================
# 配置参数
# ===============================
OUTPUT_DIR = "./wiki_data"
MAX_DOCUMENTS = 20  # 提取 20 篇文章用于快速测试
INPUT_FILE = "./zhwiki-latest-pages-articles.xml.bz2"
URL_BASE = "https://zh.wikipedia.org/wiki/"

# XML 标签正则表达式（来自 WikiExtractor）
tagRE = re.compile(r'(.*?)<(/?\w+)[^>]*?(?:/>|>(.*?)</\2|>(.*?))', re.DOTALL)


def extract_wikipedia_sample(input_file, output_dir, max_docs=20):
    """使用 wikiextractor 的 Extractor 类提取前 N 个维基百科页面"""

    print(f"正在从 {input_file} 提取 {max_docs} 篇文档...")
    print("(使用 wikiextractor 的 Extractor 类)\n")

    # 创建输出目录
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True)

    # 创建子目录（wikiextractor 会创建嵌套目录，如 AA/, AB/ 等）
    subdir = output_path / "AA"
    subdir.mkdir(exist_ok=True)
    output_file = subdir / "wiki_00.json"

    # 配置 Extractor 以 JSON 格式输出
    Extractor.to_json = True
    Extractor.keepLinks = False
    Extractor.keepSections = True
    Extractor.HtmlFormatting = False

    count = 0
    documents = []
    pages_scanned = 0
    pages_skipped = 0  # 重定向 + 内容过短

    # 解析 bz2 文件
    with bz2.BZ2File(input_file, "rb") as f:
        page = []
        page_id = ''
        revid = ''
        in_text = False
        redirect = False
        title = ''

        for line_num, line in enumerate(f):
            if count >= max_docs:
                print(f"\n✓ 已达到目标 {max_docs} 篇文档")
                break

            line_str = line.decode("utf-8", errors="ignore")

            # 快速路径：没有标签
            if '<' not in line_str:
                if in_text:
                    page.append(line_str)
                continue

            # 解析 XML 标签
            m = tagRE.search(line_str)
            if not m:
                if in_text:
                    page.append(line_str)
                continue

            tag = m.group(2)

            if tag == 'page':
                page = []
                redirect = False
                title = ''
                page_id = ''
                revid = ''

            elif tag == 'id' and not page_id:
                page_id = m.group(3).strip()

            elif tag == 'id' and page_id:
                revid = m.group(3).strip()

            elif tag == 'title':
                title = m.group(3).strip()

            elif tag == 'redirect':
                redirect = True

            elif tag == 'text':
                in_text = True
                text_content = m.group(3) or m.group(4) or ""
                page.append(text_content)
                if m.lastindex == 4:  # 自闭合标签
                    in_text = False

            elif tag == '/text':
                in_text = False
                text_content = m.group(3) or ""
                page.append(text_content)

            elif tag == '/page':
                pages_scanned += 1

                # 处理完成的页面（跳过重定向）
                if not redirect and title and page:
                    try:
                        # 使用 wikiextractor 的 Extractor 类
                        extractor = Extractor(page_id, revid, URL_BASE, title, page)
                        output_buffer = StringIO()

                        # 使用 wikiextractor 的逻辑提取
                        extractor.extract(output_buffer, html_safe=False)

                        # 获取 JSON 输出
                        json_str = output_buffer.getvalue().strip()

                        if json_str:
                            data = json.loads(json_str)
                            # 只包含内容充实的文章
                            text_len = len(data.get('text', ''))
                            if text_len > 100:
                                documents.append(data)
                                count += 1
                                print(f"  [{count}/{max_docs}] ✓ {title} ({text_len} 字符)")
                                if count >= max_docs:
                                    print(f"\n✓ 已达到目标 {max_docs} 篇文档")
                                    break
                            else:
                                pages_skipped += 1

                    except Exception as e:
                        # 跳过有问题的页面
                        pages_skipped += 1
                        print(f"  [错误] 处理失败: {title} - {e}")
                else:
                    pages_skipped += 1

                # 重置以处理下一页
                page = []
                page_id = ''
                revid = ''
                title = ''
                redirect = False

    # 统计摘要
    print(f"\n统计信息:")
    print(f"  扫描页面总数: {pages_scanned}")
    print(f"  跳过 (重定向/过短/错误): {pages_skipped}")
    print(f"  成功提取: {count}")

    # 写入 JSONL 文档
    print(f"\n正在将 {len(documents)} 篇文档写入 {output_file}...")

    with open(output_file, "w", encoding="utf-8") as f:
        for doc in documents:
            json.dump(doc, f, ensure_ascii=False)
            f.write("\n")

    print(f"✓ 完成!")
    print(f"  输出: {output_file}")

    return len(documents)


if __name__ == "__main__":
    print("="*60)
    print("维基百科样本提取器 (使用 wikiextractor 库)")
    print("="*60)
    print()

    count = extract_wikipedia_sample(INPUT_FILE, OUTPUT_DIR, MAX_DOCUMENTS)

    print()
    print("="*60)
    print(f"✓ 成功提取 {count} 篇文档")
    print("="*60)

输出结果：

============================================================
维基百科样本提取器 (使用 wikiextractor 库)
============================================================

正在从 ./zhwiki-latest-pages-articles.xml.bz2 提取 20 篇文档...
(使用 wikiextractor 的 Extractor 类)

...

✓ 已达到目标 20 篇文档

统计信息:
  扫描页面总数: 149146
  跳过 (重定向/过短/错误): 149126
  成功提取: 20

正在将 20 篇文档写入 wiki_data/AA/wiki_00.json...
✓ 完成!
  输出: wiki_data/AA/wiki_00.json

============================================================
✓ 成功提取 20 篇文档
============================================================

最终会在 wiki_data/AA/wiki_00.json 生成 JSONL 文件，每行是一篇维基百科文章：

{"id": "8572", "revid": "5313", "url": "https://zh.wikipedia.org/wiki/?curid=8572", "title": "MediaWiki:Gnunote", "text": "所有文本在..."}
...

第四步：索引数据到 Easysearch #

创建索引并将数据导入 Easysearch。

脚本代码 (index.py)：

#!/usr/bin/env python3.10
"""
将维基百科数据索引到 Easysearch。
创建带有中文分词器的索引并批量导入文档。
"""

import json
import urllib.parse
from pathlib import Path

from elasticsearch import Elasticsearch, helpers

# ===============================
# 配置参数
# ===============================
ES_HOST = "https://127.0.0.1:9200"
ES_USER = "admin"
ES_PASSWORD = "your_password"
INDEX_NAME = "wiki_zh"
WIKI_DATA_DIR = "./wiki_data"
MAX_DOCUMENTS = 20  # 索引的文档数量

# ===============================
# 连接 Easysearch
# ===============================
def create_es_client():
    """创建并返回 Elasticsearch 客户端"""
    client = Elasticsearch(
        [ES_HOST],
        http_auth=(ES_USER, ES_PASSWORD),  # elasticsearch<7.14 使用 http_auth
        verify_certs=False,
        ssl_show_warn=False,
    )
    if not client.ping():
        raise ConnectionError(f"无法连接到 Easysearch: {ES_HOST}")
    print(f"已连接到 Easysearch: {ES_HOST}")
    return client

# ===============================
# 创建索引
# ===============================
def create_index(client: Elasticsearch):
    """创建索引并配置中文分词器"""
    if client.indices.exists(index=INDEX_NAME):
        client.indices.delete(index=INDEX_NAME)
        print(f"已删除现有索引: {INDEX_NAME}")

    # 使用 ik_max_word 分词器配置索引映射
    mapping = {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ik_max_word",
                "fields": {
                    "keyword": {
                        "type": "keyword",
                        "ignore_above": 256
                    }
                }
            },
            "content": {
                "type": "text",
                "analyzer": "ik_max_word"
            },
            "url": {
                "type": "keyword"
            }
        }
    }

    client.indices.create(index=INDEX_NAME, body={"mappings": mapping})
    print(f"已创建索引: {INDEX_NAME}")
    print(f"  - title: text (ik_max_word)")
    print(f"  - content: text (ik_max_word)")
    print(f"  - url: keyword")

# ===============================
# 生成文档
# ===============================
def generate_documents():
    """读取 wiki_data 目录并生成待索引的文档"""
    count = 0
    wiki_path = Path(WIKI_DATA_DIR)

    if not wiki_path.exists():
        raise FileNotFoundError(f"找不到 wiki_data 目录: {WIKI_DATA_DIR}")

    # 递归查找所有 JSONL 文件
    for json_file in wiki_path.rglob("*.json"):
        with open(json_file, "r", encoding="utf-8") as f:
            for line in f:
                if count >= MAX_DOCUMENTS:
                    return

                line = line.strip()
                if not line:
                    continue

                try:
                    data = json.loads(line)
                    title = data.get("title", "").strip()
                    # wikiextractor 输出使用 'text' 键
                    content = data.get("text", "").strip()

                    if not title or not content:
                        continue

                    # 构建维基百科 URL
                    url_title = urllib.parse.quote(title.replace(" ", "_"))
                    url = f"https://zh.wikipedia.org/wiki/{url_title}"

                    count += 1
                    yield {
                        "_index": INDEX_NAME,
                        "_source": {
                            "title": title,
                            "content": content,
                            "url": url
                        }
                    }

                except json.JSONDecodeError:
                    continue

# ===============================
# 索引文档
# ===============================
def index_documents(client: Elasticsearch):
    """使用 streaming_bulk 批量索引文档"""
    print(f"\n正在索引文档 (最多 {MAX_DOCUMENTS} 篇)...")

    success_count = 0
    error_count = 0

    for ok, info in helpers.streaming_bulk(
        client,
        generate_documents(),
        chunk_size=50,
        raise_on_error=False,
        raise_on_exception=False
    ):
        if ok:
            success_count += 1
        else:
            error_count += 1
            print(f"  错误: {info}")

    # 刷新索引使文档可被搜索
    client.indices.refresh(index=INDEX_NAME)

    print(f"\n索引完成!")
    print(f"  成功: {success_count}")
    print(f"  错误: {error_count}")

# ===============================
# 主函数
# ===============================
def main():
    print("="*60)
    print("Easysearch 维基百科索引器")
    print("="*60)

    client = create_es_client()
    create_index(client)
    index_documents(client)

    print("="*60)
    print("完成!")
    print("="*60)

if __name__ == "__main__":
    main()

运行索引脚本：

python3.10 index.py

输出：

============================================================
Easysearch 维基百科索引器
============================================================
已连接到 Easysearch: https://127.0.0.1:9200
已创建索引: wiki_zh
  - title: text (ik_max_word)
  - content: text (ik_max_word)
  - url: keyword

正在索引文档 (最多 20 篇)...

索引完成!
  成功: 20
  错误: 0
============================================================
完成!
============================================================

第五步：搜索验证 #

创建独立的搜索脚本，支持命令行关键词和交互式搜索。

脚本代码 (search.py)：

#!/usr/bin/env python3.10
"""
在 Easysearch 中搜索维基百科索引并支持高亮显示。
"""

from elasticsearch import Elasticsearch

# ===============================
# 配置参数
# ===============================
ES_HOST = "https://127.0.0.1:9200"
ES_USER = "admin"
ES_PASSWORD = "your_password"
INDEX_NAME = "wiki_zh"

# ===============================
# 连接 Easysearch
# ===============================
def create_es_client():
    """创建并返回 Elasticsearch 客户端"""
    client = Elasticsearch(
        [ES_HOST],
        http_auth=(ES_USER, ES_PASSWORD),
        verify_certs=False,
        ssl_show_warn=False,
    )
    if not client.ping():
        raise ConnectionError(f"无法连接到 Easysearch: {ES_HOST}")
    return client

# ===============================
# 搜索并高亮
# ===============================
def search(client: Elasticsearch, keyword: str, size: int = 5):
    """搜索关键词并返回带有高亮的结果"""
    query = {
        "query": {
            "match": {
                "content": keyword
            }
        },
        "highlight": {
            "fields": {
                "content": {
                    "fragment_size": 150,
                    "number_of_fragments": 2
                }
            }
        },
        "size": size
    }

    response = client.search(index=INDEX_NAME, body=query)
    return response

# ===============================
# 显示结果
# ===============================
def display_results(response, keyword: str):
    """格式化打印搜索结果"""
    total = response['hits']['total']['value']
    hits = response['hits']['hits']

    print(f"关键词: '{keyword}'")
    print(f"找到 {total} 条结果\n")

    for i, hit in enumerate(hits, 1):
        source = hit["_source"]
        highlights = hit.get("highlight", {}).get("content", [])

        print(f"[{i}] {source['title']}")
        print(f"    URL: {source['url']}")
        print(f"    高亮:")

        for hl in highlights:
            print(f"      ...{hl}...")
        print()

# ===============================
# 主函数
# ===============================
def main():
    import sys

    print("="*60)
    print("Easysearch 维基百科搜索")
    print("="*60)

    client = create_es_client()
    print(f"已连接到: {ES_HOST}")
    print(f"索引: {INDEX_NAME}\n")

    # 使用命令行关键词或默认关键词
    keyword = sys.argv[1] if len(sys.argv) > 1 else "维基"
    print(f"搜索关键词: {keyword}\n")

    response = search(client, keyword)
    display_results(response, keyword)

if __name__ == "__main__":
    main()

运行搜索脚本：

# 使用默认关键词 "维基"
python3.10 search.py

# 或指定关键词
python3.10 search.py 历史

输出示例：

============================================================
Easysearch 维基百科搜索
============================================================
已连接到: https://127.0.0.1:9200
索引: wiki_zh

搜索关键词: 维基

关键词: 维基

找到 1 条结果

[1] Template:Rfn
    URL: https://zh.wikipedia.org/wiki/Template%3ARfn
    高亮:
      ...这里一般没有人对你的文章进行校阅哦，只要不违反<em>维</em><em>基</em>原则的文字都没被保存下来。每个人都是平等的，大家都在参与贡献而已...

注意搜索结果中的 <em> 标签——这就是关键词高亮！在实际的 Web 应用中，你可以用 CSS 把这些标签渲染成黄色背景。

总结 #

恭喜！你已经成功搭建了一个支持中文分词的搜索服务。回顾一下我们做了什么：

获取中文维基百科数据
使用 wikiextractor 库提取并清理文本
用 Python 连接 Easysearch
配置 ik_max_word 中文分词器
使用 streaming_bulk 高效索引数据
执行带高亮的搜索查询

完整文件清单 #

文件	用途
`zhwiki-latest-pages-articles.xml.bz2`	中文维基百科数据
`extract_sample.py`	提取 20 篇样本文档（使用 wikiextractor 库）
`index.py`	创建索引并导入数据
`search.py`	搜索并显示高亮结果
`wiki_data/AA/wiki_00.json`	提取后的维基百科数据

接下来 #

你可以：

尝试搜索其他关键词，比如"人工智能"、“量子力学"等
修改 extract_sampley.py和 index.py ，提取并索引更多数据
开发一个 Web 前端，打造真正的搜索应用

祝你玩得开心！

标签

Easysearch 搜索服务私有部署搜索引擎维基百科