【React】React/Python/ElasticSearchで簡易的なドキュメント検索Webアプリを作成

はじめに

CSVやPDF, Excel などの様々なデータソースを ElasticSearch に溜めて、指定の文言が含まれているファイルを検索できる簡易的なドキュメント検索Webアプリケーションを作成しました。

今回のソースコードは下記にプッシュしています。

github.com

 

機能、動作イメージ

ドキュメント検索機能

• 任意の文字列を入力して検索を押下
• 検索文字列が含まれるドキュメントがある場合、そのファイル名とファイル内のテキスト（200文字まで）を検索結果として表示する

 

 

ドキュメントアップロード機能

• 選択ボタンをクリックするとファイル選択ダイアログが立ち上がり、任意のドキュメントを選択
• 送信ボタンをクリックすると選択したドキュメントが ElasticSearch に登録される

※今回は、CSV, PDF, Excel ファイルいずれかのみ受け付ける形にしています

 

 

使用技術

• フロントエンド
  • React v18.2.0
  • TypeScript v4.9.5

• バックエンド
  • Python v3.8.10
  • FastAPI v0.110.0
  • ElasticSearch v7.5.1
  • Docker v20.10.17

 

実装詳細ーバックエンド

以下で ElasticSearch とのつなぎこみ箇所を中心にソースコードを添付しています。

ElasticSearch

ElasticSearch のDockerイメージから環作成しています。

docker-compose.yml

version: "3"

services:

  sysctl:

    image: alpine

    container_name: sysctl

    command: ["sysctl", "-w", "vm.max_map_count=262144"]

    privileged: true

    networks:

      - esnet

  es01:

    build:

      context: .

      dockerfile: Dockerfile

    container_name: es01

    environment:

      - node.name=es01

      - cluster.initial_master_nodes=es01

      - cluster.name=docker-cluster

      - bootstrap.memory_lock=true

      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"

    ulimits:

      memlock:

        soft: -1

        hard: -1

    volumes:

      - esdata01:/usr/share/elasticsearch/data

    ports:

      - 9200:9200

    networks:

      - esnet

    depends_on:

      - sysctl

volumes:

  esdata01:

    driver: local

networks:

  esnet:

 

• esdata01 というボリュームを作成してマウントしています
• ElasticSearch を起動後、メモリ割り当てが足りない旨のエラーが出てそのまま停止してしまう事象が発生したため、 vm.max_map_count=262144 でメモリ割り当てを増やして停止しないようにしています

Dockerfile

FROM docker.elastic.co/elasticsearch/elasticsearch:7.5.1

RUN elasticsearch-plugin install analysis-kuromoji

 

• ElasticSearch コンテナはこちらで指定しています
• 今回日本語も扱うため、kuromojiプラグイン を追加しています

 

バックエンド

リクエストハンドリング

@app.get("/search", response_model=SearchDocumentResponse)

def search(text: Optional[str] = Query(None, description="検索クエリー")):

    return search_document(text)

@app.post("/upload")

async def upload(file: UploadFile = File(...)):

    result = await upload_document(file)

    return JSONResponse(content={"result": result})

 

• FastAPI を使い、「ドキュメント検索」と「ドキュメントアップロード」機能用のエンドポイントを定義しています
• 検索エンドポイントは検索文字列を受け取る必要があるため、text というクエリーパラメーターで受け取るようにしています
• ドキュメントアップロードで送信されたファイルデータは、UploadFile というクラスで受け取ることができます

 

ElasticSearch のドキュメント検索

import re

import requests

from models.search_document import SearchDocumentResponse, SearchDocumentResult

def search_document(text: str):

    try:

        # Elasticsearchへのクエリ実行

        if is_alphanumeric(text):

            # 半角英数字の場合

            query = {"query": {"regexp": {"doc.text": f".*{text}.*"}}}

        else:

            # 日本語の場合

            query = {"query": {"match_phrase": {"doc.text": text}}}

        response = requests.post("http://localhost:9200/book/_search", json=query)

        # ステータスコードが200以外の場合はエラーを出力して終了

        if response.status_code != 200:

            print(

                f"Error: Failed to retrieve data from Elasticsearch. Status code: {response.status_code}"

            )

            return SearchDocumentResponse(count=0, result=)

        # レスポンスからhitsを取得

        data = response.json()["hits"]["hits"]

        # 検索結果を格納するリスト

        search_results =

        # レスポンスのhitsに対してループ処理

        for hit in data:

            file_name = hit["_source"]["doc"]["name"]

            # textがリストか文字列かで処理を分岐

            text_content = hit["_source"]["doc"]["text"]

            if isinstance(text_content, list):

                # textがリストの場合、各要素を連結して1つの文字列にする

                text_content = "\n".join(text_content)

            search_results.append(

                SearchDocumentResult(file_name=file_name, text=text_content)

            )

        # 検索結果をSearchDocumentResponseに格納して返す

        return SearchDocumentResponse(count=len(search_results), result=search_results)

    except Exception as e:

        print(

            f"Error: An error occurred while processing the response from Elasticsearch: {str(e)}"

        )

        return SearchDocumentResponse(count=0, result=)

def is_alphanumeric(text):

    pattern = re.compile(r"^[a-zA-Z0-9]*$")

    return bool(pattern.match(text))

 

• 検索文字列を text 変数として受け取り、ElasticSearch への検索クエリーを構築します
• 日本語と英数字両方に対応している検索クエリーが見当たらかなったため、日本語と英数字で投げるクエリーを動的に変えています（いずれも部分一致検索で指定の文字列が含まれるか検索する）
• 複数件ヒットする可能性があるため、配列に検索結果を詰めて返却しています

 

ElasticSearch にドキュメント登録

async def upload_document(file: UploadFile = File(...)):

    content_type = file.content_type

    if content_type == FILE_TYPE_PDF:

        await upload_pdf(file)

    elif content_type == FILE_TYPE_CSV:

        await upload_csv(file)

    elif content_type == FILE_TYPE_EXCEL:

        await upload_excel(file)

    else:

        print("not found")

        pass

• 今回はCSV, Excel, PDF ドキュメントを対象に、ElasticSearch にドキュメントを登録します

ElasticSearch にドキュメント登録（CSV）

from csv import DictReader

import json

import os

from fastapi import File, UploadFile

import requests

from constant.constant import (

    ELASTIC_SEARCH_REQUEST_HEADERS,

    ELASTIC_SEARCH_URL,

)

async def upload_csv(file: UploadFile = File(...)):

    try:

        file_name = file.filename

        file_path = os.path.join(os.getcwd(), file_name)

        # アップロードされたファイルを保存

        with open(file_path, "wb") as f:

            f.write(await file.read())

        file_content =

        with open(file_path, "rt", encoding="utf-8") as file:

            reader = DictReader(file)

            for row in reader:

                row_json = json.dumps(row, ensure_ascii=False)

                file_content.append(row_json)

        # Elasticsearchに送信するデータを構築

        data = {"doc": {"name": file_name, "text": file_content}}

        print("Uploaded file data:", data)

        response = requests.post(

            ELASTIC_SEARCH_URL,

            json=data,

            headers=ELASTIC_SEARCH_REQUEST_HEADERS,

        )

        if response.status_code == 201:

            print("Document indexed successfully.")

        else:

            print(f"Failed to index document. Status code: {response.text}")

            return False

        # 一時ファイルを削除

        os.unlink(file_path)

        return True

    except Exception as e:

        print(

            f"Error: An error occurred while processing uploading csv to Elasticsearch: {str(e)}"

        )

        return False

 

• 引数の UploadFile からファイル名とファイル内容を読み込みます
• ファイル内容の読み込み部分は、いったん Temp ファイルに退避しておいて、Temp ファイルからファイル内容を文字列に読み込んでいます（直接 UploadFile から読み込んで ElasticSearch に送信したところエラーになったため。。）
• name と text というフィールドを持ったドキュメントを登録しています。（※事前に book というインデックスを作成しこれを使用しています）

 

 

ElasticSearch にドキュメント登録（Excel）

import json

import os

from fastapi import File, UploadFile

import pandas as pd

import requests

from constant.constant import ELASTIC_SEARCH_REQUEST_HEADERS, ELASTIC_SEARCH_URL

async def upload_excel(file: UploadFile = File(...)):

    try:

        file_name = file.filename

        file_path = os.path.join(os.getcwd(), file_name)

        # アップロードされたファイルを保存

        with open(file_path, "wb") as f:

            f.write(await file.read())

        # ブック全体の内容を入れるための空のリストを作成

        book_content = []

        # Excelファイルから各シートのデータを読み込んでリストに追加する

        with pd.ExcelFile(file_path) as xls:

            for sheet_name in xls.sheet_names:

                df = pd.read_excel(xls, sheet_name)

                records = df.to_dict(orient="records")

                book_content.extend(records)

        # Elasticsearchに送信するデータを構築

        data = {

            "doc": {

                "name": file_name,

                "text": json.dumps(book_content, ensure_ascii=False),

            }

        }

        print("Uploaded file data:", data)

        response = requests.post(

            ELASTIC_SEARCH_URL,

            json=data,

            headers=ELASTIC_SEARCH_REQUEST_HEADERS,

        )

        if response.status_code == 201:

            print("Document indexed successfully.")

        else:

            print(f"Failed to index document. Status code: {response.text}")

            return False

        # 一時ファイルを削除

        os.unlink(file_path)

        return True

    except Exception as e:

        print(

            f"Error: An error occurred while processing uploading excel to Elasticsearch: {str(e)}"

        )

        return False

 

• CSVの処理同様に、ファイル名とファイル内容を読み込んでいます
• pandas を使いシートごとに読み込んだ内容を配列に読み込み、最終的に１つの文字列に変換して ElasticSearch に登録します

 

ElasticSearch にドキュメント登録（PDF）

import os

from tempfile import NamedTemporaryFile

from fastapi import File, UploadFile

from pdfminer.high_level import extract_text

import requests

from constant.constant import ELASTIC_SEARCH_REQUEST_HEADERS, ELASTIC_SEARCH_URL

async def upload_pdf(file: UploadFile = File(...)):

    try:

        file_name = file.filename

        file_path = os.path.join(os.getcwd(), file_name)

        # アップロードされたファイルを保存

        with open(file_path, "wb") as f:

            f.write(await file.read())

        # PDFからテキストを抽出

        file_text = extract_text(file_path)

        # Elasticsearchに送信するデータを構築

        data = {"doc": {"name": file_name, "text": file_text}}

        print("Uploaded file data:", data)

        response = requests.post(

            ELASTIC_SEARCH_URL,

            json=data,

            headers=ELASTIC_SEARCH_REQUEST_HEADERS,

        )

        if response.status_code == 201:

            print("Document indexed successfully.")

        else:

            print(f"Failed to index document. Status code: {response.text}")

            return False

        # 一時ファイルを削除

        os.unlink(file_path)

        return True

    except Exception as e:

        print(

            f"Error: An error occurred while processing uploading pdf to Elasticsearch: {str(e)}"

        )

        return False

 

• PDF の読み込みは pdfminer がシンプルに書けて使いやすかったので採用しました
• こちらもCSV, Excel の処理と同じ形式で、ファイル名とファイル内容を読み込んで ElasticSearch に登録します

 

以上のような実装で ElasticSearch へドキュメントの登録と検索を実現することができました。

※ elasticsearch · PyPI があり、これを最初試していたもののDocker起動している ElasticSearch にうまく接続ができず、今回は HTTP リクエストで登録する方式を取りました

 

おわりに

今回はCSV, Excel, PDF のみでしたが、他にもあらゆる形式のドキュメントを ElascticSearch に溜めこむことができると思いました。

これを応用して例えば、自組織のファイルを ElascticSearch にためていって、任意の文章で検索して目的のファイルを探す、みたいな使い方もできそうですね。