Watson Discovery 汎用ツール(コレクションの文書一括削除など)

はじめに

(2019-07-15) いつの間にかライブラリ名から変わってしまっていて、以下の解説が役に立たなくなりました。新しい記事Jupyter NotebookからWatson Discoveryを操作する(2019-07-15最新版)を書き直したので、今後はこちらを参照さえて下さい。
Watson Discoveryを使った簡単な検証をしようとしているのですが、UIツールの機能が意外と足りなくて簡単なことができず(例えばコレクションの文書を全部消すとか、あるディレクトリ配下のファイルをまとめてアップロードするとか。後者はクローラ使えばできるのですが、ちょっとした目的でいちいちクローラの構成を組むのも大変なので)イライラしたので、ある程度汎用的なツールを作ってみました。

機能とソースを以下に添付しておきます。
また、Githubにもアップしておいたので、ここからダウンロードすることも可能です。

• 2018-02-19 dicoveryインスタンスを作るまでの共通処理をcommon.pyに外出ししました。
• 2018-06-29 全文書数をAPIで取得するように改修(前はMax100件決め打ちだった)

前提

Python3を前提としています。

動作確認はMacOS 10.13.3, Python 3.6.0で行っています。

必要ライブラリ

watson_developer_cloudとdotenvです。

どちらもpip installコマンドで導入します。

Githubからダウンロードした場合は、次のコマンドでまとめて導入することも可能です。

$ pip install -r requirements.txt 

パラメータの設定

認証情報や、各種IDは.envから読み込む前提です。

以下のコマンドで、.envファイルを作った後、環境に応じて各IDを設定して下さい。

$ cp .env.example .env

(参考)設定ファイルの内容

DISCOVERY_URL=https://gateway.watsonplatform.net/discovery/api"
DISCOVERY_USERNAME='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'
DISCOVERY_PASSWORD='xxxxxxxxxxxx'
ENVIRONMENT_ID='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'
CONFIGURATION_ID='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'
COLLECTION_ID='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'

各シェルの説明

common.py

パラメータを読み込んでdicoveryインスタンスを作るまでの共通処理を行います。

common.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# common.py
#
# 共通処理 
#

import sys
import os
import json
from watson_developer_cloud import DiscoveryV1
from dotenv import load_dotenv
from os.path import join, dirname

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)
username = os.environ.get("DISCOVERY_USERNAME")
password = os.environ.get("DISCOVERY_PASSWORD")
environment_id = os.environ.get("ENVIRONMENT_ID")
collection_id = os.environ.get("COLLECTION_ID")

if ( username is None ):
    print("username is null")
    exit(1)

if ( password is None ):
    print("password is null")
    exit(1)

discovery = DiscoveryV1(
  username=username,
  password=password,
  version="2017-11-07"
)

delete-all.py

(引数なし)
対象コレクションの文書をすべて削除します。

ソースはこちら

delete-all.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# delete-all.py
#
# 対象コレクションのドキュメントを全部削除します。
#

import common
discovery = common.discovery
environment_id = common.environment_id
collection_id = common.collection_id

# 文書件数取得
collection = discovery.get_collection(environment_id, collection_id)
doc_count = collection['document_counts']['available']

# id一覧取得
results = discovery.query(environment_id, collection_id, return_fields = 'id', count = doc_count)["results"]
ids = [item["id"] for item in results]

for id in ids:
    print('deleting doc: id =' + id)
    discovery.delete_document(environment_id, collection_id, id)

import-docs.py

(引数1つ)
引数をディレクトリ名と解釈し、該当ディレクトリ配下のPDFファイルをすべて検索して、Discveryにアップロードします。

ソースは下記になります。

import-docs.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# import-docs.py
# 
# パラメータで指定したディレクトリ配下のpdfファイルをすべてDiscoveryに取り込みます 
#

import os
import sys
from os.path import join, dirname

import common
discovery = common.discovery
environment_id = common.environment_id
collection_id = common.collection_id

argvs = sys.argv
path = argvs[1]

def find_all_files(directory):
    for root, dirs, files in os.walk(directory):
        yield root
        for file in files:
            yield os.path.join(root, file)

files = find_all_files(path)
files_ext = [os.path.splitext(file) for file in files]
pdf_files = [file+ext for file, ext in files_ext if ext in ['.pdf']]

for pdf_file in pdf_files:
    print("adding ducument " + pdf_file)
    file = open(pdf_file, "rb")
    response = discovery.add_document(environment_id, collection_id, 
                file=file,
                file_content_type='application/pdf', 
                filename=pdf_file)
    print(response)

extract-text.py

(引数1つ)
対象コレクションの文書をすべて検索し、textフィールドの内容をファイルに書き込みます。

ファイル名は、<Discovery上の文書ID>.txtとなります。

引数をディレクトリ名と解釈し、ファイルの保存先は引数で指定したディレクトリとなります。

ソースは以下です。

extract-text.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Discoveryの全文書のtextを指定ディレクトリにexportする
#

import sys
import codecs

import common
discovery = common.discovery
environment_id = common.environment_id
collection_id = common.collection_id

argvs = sys.argv
path = argvs[1]

# 文書件数取得
collection = discovery.get_collection(environment_id, collection_id)
doc_count = collection['document_counts']['available']

results = discovery.query(
        environment_id, 
        collection_id, 
        count=doc_count,
        return_fields='id,text')["results"]

for result in results:
    id = result['id']
    text = result['text']
    outfn = path + '/' + id + '.txt'
    print(outfn)
    outfile = codecs.open(outfn, 'w', 'utf-8')
    outfile.write(text)
    outfile.close()

終わりに

初めてDiscoveryをPythonから呼び出してみました。
Pythonのライブラリは、全部同期処理でコールバックのことを考えないでいいのがJavaScriptと比べて便利なのですが、APIリファレンスがとにかく不親切でした。
仕方がないので、パラメータの設定はライブラリのソースを読んで実装しました。
例えば return_fields='id,text' みたいなパラメータの設定は、APIリファレンスだけ見ても絶対わからないと思います。
やりたいことがぴったりあっていなくても、このサンプルを見ると、Python APIを使った実装の仕方がわかるかと思います。

(参考)　ライブラリソースのうち、今回参考にした箇所を以下に添付しておきます。

    def query(self,
              environment_id,
              collection_id,
              filter=None,
              query=None,
              natural_language_query=None,
              passages=None,
              aggregation=None,
              count=None,
              return_fields=None,
              offset=None,
              sort=None,
              highlight=None,
              passages_fields=None,
              passages_count=None,
              passages_characters=None,
              deduplicate=None,
              deduplicate_field=None):
        """
        Query documents.
        See the [Discovery service
        documentation](https://console.bluemix.net/docs/services/discovery/using.html) for
        more details.
        :param str environment_id: The ID of the environment.
        :param str collection_id: The ID of the collection.
        :param str filter: A cacheable query that limits the documents returned to exclude any documents that don't mention the query content. Filter searches are better for metadata type searches and when you are trying to get a sense of concepts in the data set.
        :param str query: A query search returns all documents in your data set with full enrichments and full text, but with the most relevant documents listed first. Use a query search when you want to find the most relevant search results. You cannot use `natural_language_query` and `query` at the same time.
        :param str natural_language_query: A natural language query that returns relevant documents by utilizing training data and natural language understanding. You cannot use `natural_language_query` and `query` at the same time.
        :param bool passages: A passages query that returns the most relevant passages from the results.
        :param str aggregation: An aggregation search uses combinations of filters and query search to return an exact answer. Aggregations are useful for building applications, because you can use them to build lists, tables, and time series. For a full list of possible aggregrations, see the Query reference.
        :param int count: Number of documents to return.
        :param list[str] return_fields: A comma separated list of the portion of the document hierarchy to return_fields.
        :param int offset: The number of query results to skip at the beginning. For example, if the total number of results that are returned is 10, and the offset is 8, it returns the last two results.
        :param list[str] sort: A comma separated list of fields in the document to sort on. You can optionally specify a sort direction by prefixing the field with `-` for descending or `+` for ascending. Ascending is the default sort direction if no prefix is specified.
        :param bool highlight: When true a highlight field is returned for each result which contains the fields that match the query with `<em></em>` tags around the matching query terms. Defaults to false.
        :param list[str] passages_fields: A comma-separated list of fields that passages are drawn from. If this parameter not specified, then all top-level fields are included.
        :param int passages_count: The maximum number of passages to return. The search returns fewer passages if the requested total is not found. The default is `10`. The maximum is `100`.
        :param int passages_characters: The approximate number of characters that any one passage will have. The default is `400`. The minimum is `50`. The maximum is `2000`.
        :param bool deduplicate: When `true` and used with a Watson Discovery News collection, duplicate results (based on the contents of the `title` field) are removed. Duplicate comparison is limited to the current query only, `offset` is not considered. Defaults to `false`. This parameter is currently Beta functionality.
        :param str deduplicate_field: When specified, duplicate results based on the field specified are removed from the returned results. Duplicate comparison is limited to the current query only, `offset` is not considered. This parameter is currently Beta functionality.
        :return: A `dict` containing the `QueryResponse` response.
        :rtype: dict
        """
        if environment_id is None:
            raise ValueError('environment_id must be provided')
        if collection_id is None:
            raise ValueError('collection_id must be provided')
        params = {
            'version': self.version,
            'filter': filter,
            'query': query,
            'natural_language_query': natural_language_query,
            'passages': passages,
            'aggregation': aggregation,
            'count': count,
            'return': self._convert_list(return_fields),
            'offset': offset,
            'sort': self._convert_list(sort),
            'highlight': highlight,
            'passages.fields': self._convert_list(passages_fields),
            'passages.count': passages_count,
            'passages.characters': passages_characters,
            'deduplicate': deduplicate,
            'deduplicate.field': deduplicate_field
        }
        url = '/v1/environments/{0}/collections/{1}/query'.format(
            *self._encode_path_vars(environment_id, collection_id))
        response = self.request(
            method='GET', url=url, params=params, accept_json=True)
        return response