自然言語処理に関する機械学習の手法を試す際には大量のテキストデータが必要になることがあります。そこで、手軽に使えるテキストデータとしてWikipediaの記事データを使うことにしたのですが、全データを対象にしてアルゴリズムを学習させるのは時間がかかり過ぎますし、特定のカテゴリ（分野）に特化して学習したモデルを作りたいと思うことも少なくありません。
MySQLを使って抽出する方法もあるようですが環境の構築が少々面倒だと思ったので、全データから特定カテゴリの記事のみを抽出するプログラムを書くことにしました。
全記事データを取得するまで
Wikipediaのコンテンツはクローラなどのプログラムを用いてスクレイピングすることは禁止されています。しかし、Wikipediaでは全記事のダンプデータが提供されていますので、記事の収集にはこれを利用します。
この項目の手順についてはこちらのページを参考にさせていただきました。ほぼそのままなので簡単に書きます。
１．日本語版Wikipediaの最新情報ダウンロードページに行き、
jawiki-latest-pages-articles.xml.bz2をダウンロードする。その後、適当なフォルダにコピーしておきます。
２．Wikipedia EXtractorのGithubへ行き、「Code」→「Download ZIP」をクリック
３．ダウンロード後、フォルダの中にあるWikiExtractor.pyを先ほどダウンロードしたjawiki-latest-pages-articles.xml.bz2ファイルと同じフォルダにコピーする
４．コマンドプロンプトでそのフォルダに移動後、以下のコマンドを実行します。
Anacondaをインストールしてパスを通していないという方は、Anaconda Promptを使いましょう。
python WikiExtractor.py -b 500K -o wikipedia jawiki-latest-pages-articles.xml.bz2
これでWikipediaの記事データがwikipediaというフォルダに最大500KBずつのファイルで出力されます。この処理には25分くらいかかりました。
私の環境ではwikipediaフォルダ直下のフォルダ数はAA～CJの62、最後のCJフォルダを除いて各フォルダにwiki_00～wiki99まで100のファイルが入っているという構成になりました。
＜以下のフォルダ構成図と出力ファイルサンプルは参考サイトより引用＞
└── wikipedia
├─── AA
│ ├── wiki_00
│ ├── wiki_02
│ │ ・
│ │ ・
│ └── wiki_99
│
├─── AB
│ ・
│ ・
└─── CJ
<出力ファイルサンプル (wiki_01の一部抜粋)>
<doc id="56" url="https://ja.wikipedia.org/wiki?curid=56" title="地理"> 地理 地理（ちり、英: Geography） 「地理」という表現は古くからあり、有名なところでは漢書の『地理志』がある。 地理学とは、地球の表面と住民の状態、その相互関係を研究する学問である。 「地理」は、日本の学校で設置されている、「人間の生活に影響を与える地域的、社会的な構造」を学ぶための科目である。自然環境や産業環境などを含む環境を学習対象としている。小学校および中学校においては、歴史や公民と並び、社会科の一分野である。高等学校においては、最近は「地理歴史科」という教科の中の一科目となっており、「地理A」「地理B」に細分されている。 </doc>
特定カテゴリのページIDを得る
ここからが本題です。出力ファイルサンプルにあるように、各記事には
そこで、PetScanという検索ツールを使ってこのIDを取得することにします。
このツールの概要はこちらです。
PetScan (以前の名称CatScan) は、指定された条件に基づいてウィキペディアの記事カテゴリ（とそのサブカテゴリ）を検索し、条件に合った記事やスタブ、画像、カテゴリを見つけ出すことができる外部ツールです。また、ふたつのカテゴリに属しているすべての記事を見つけ出すためにも使用できます。
このページ左上の『PetScanを開く』でツールが開きます。
使い方については、マニュアルページに詳しく書いてありますが、簡単に使う場合は以下の設定だけでもidを取得することができます。
ツールを開いて、
- 「カテゴリ」タブの＜カテゴリ深度＞と＜カテゴリ＞を入力する
- 「出力」タブの＜出力形式＞をCSVにして実行をクリックする
これでcsvファイルのダウンロードが開始されます。
カテゴリ深度はツリーの深さを表し、数字が大きい程取得するページが多くなっていくようです。また、カテゴリ欄は以下のように複数書く*1ことで組み合わせて検索（AND検索）もできます。またその際は、'|'(パイプ文字)と数字を組み合わせてカテゴリ深度も個別に指定できます。
たとえば、この例だと医薬品はカテゴリ深度5、薬学はカテゴリ深度3といった具合です。
最初は出力形式を「HTML」にして、実行後に表示されるページ名がだいたい自分が欲しいものになっているかを何度か試して確認してから、csv出力された方がいいかもしれません。実際、自分が入力したカテゴリとはあまり関係なさそうなページ名が入っていることがありましたが、組み合わせや深度、その他の設定でうまく絞ることができそうです。
ファイルのダウンロードが終わったら、上で作成したwikipediaフォルダがある所と同じ場所（階層）に移しておきます。
特定カテゴリのページIDに該当する記事のみを抜き出す
これで準備が整いました。あとは上で作成した全データの中からcsvファイルのページIDと一致する記事だけを抽出し、別の場所にファイル出力すれば、目的は果たせそうです。
しかし、最初に抽出したデータは500KB近いファイルが6200以上もある*2ので、単純にcsvファイルから得た1つのページIDの記事を探すのに、毎回全ファイルを走査していたのでは無駄が多く、とんでもなく時間がかかりそうです。
ある程度プログラムを書ける人であれば、そんな実装にはしないと思いますが、私も少しでも効率よく記事を収集するプログラムを考えることにしました。
プログラムの概要
幸い、全データの方はAAからCJまで、idが昇順になっているようなのでこれを利用します。先に探索する場所（フォルダ）を絞り込んでからidが一致する記事を探す、というものです。
プログラムの流れは次の通りです。
- csvファイルからページIDを抜き出す
- そのページIDがありそうなフォルダをAA～CJ（末尾フォルダ）の中から選ぶ
- フォルダ中のファイルを1つずつ走査していき、ID、タイトル、記事を抽出する
- IDが一致するものがあれば記事をファイル出力する
- １に戻る
以下、要点を順番にみていきます。
csvファイルの読み込み
上の手順1の部分です。
私の場合はカテゴリは「医薬品」、カテゴリ深度は「2」を選んだところ、件数は2961件でした。（以下は各自ダウンロードしたファイルに合わせて読み替えて下さい。）
得られたcsvファイルの形式は、以下のとおりです。
"number","title","pageid","namespace","length","touched"
このうち、プログラムで使うことになる項目は、"pageid"（ページID）、"namespace"（名前空間）の2つです。（"title"は変数page_nameとして見つからなかった場合に保存するだけ）
namespaceが「Category」または「Template」の場合は、そのページIDは無視して処理を次に飛ばすことにしました。
ページIDから探索するフォルダを決定する
wikipediaフォルダ内のAA～CJ（末尾フォルダ）に含まれているwikiファイル（00～99）のidは昇順になっています。各フォルダに含まれている最大のidがわかれば、次の図のように探索対象のフォルダの当たりをつけることができます。私の環境では各フォルダに含まれている最大のidは次の図のようになっていました。
この図の場合、たとえばcsvのページIDが15000なら探索フォルダはAB、IDが53000ならAEということになります。
この探索フォルダの決定を行うため、キーがフォルダ名、値がそのフォルダ中にある最大のidの辞書を先に作成しておきます。
# key: フォルダ名, value: フォルダ中の最大idの辞書を作成する # 各ディレクトリの最後のファイルの最後の記事からid(=そのディレクトリの最大のid)を抜き出して辞書の値とする def mapping_dir(dir_path): dir_list = os.listdir(dir_path) for _dir in dir_list: last_f_path = sorted(glob.glob(os.path.join(dir_path, _dir) + '\\*'))[-1] max_id, _, _ = format_text(sorted(get_articlelist(last_f_path))[-1]) dir_info[_dir] = max_id return dir_info
この関数にはwikipediaフォルダがあるパスを与えます。
各フォルダの最後のファイルの最後の記事からid(=最大のid)を抜き出し、それを辞書の値としています。関数format_textは<doc>～</doc>で囲まれたテキストからid、タイトル、記事本文を抽出する関数、get_articlelistはファイルを読み込み<doc>で囲まれた記事のリストを返す関数です。
ページの最後にプログラム全文を載せていますので、そちらもご参照ください。
なお、辞書は並び順が保証されるOrderedDictを使用します。
この関数によって、次のような辞書が作成されます。
OrderedDict([('AA', 7722), ('AB', 17456), ('AC', 28826), ・ ・ ('CJ', 4160198)])
次は、この辞書とcsvファイルから読み込んだページIDを使って探索フォルダを返す関数です。
# csvファイルから読み込んだpage_idと作成した辞書(dir_info)から # page_idがあると思われるフォルダ名を探索フォルダとして返す def get_search_dir(page_id, dir_info): min_id = 0 for dir_name, max_id in dir_info.items(): search_dir = dir_name if min_id <= page_id <= max_id: return search_dir else: min_id = max_id return next(iter(dir_info)) # 見つからなかった場合は最初のkey(AA)を返す
min_id <= page_id <= max_idのところで受け取ったページIDがどの範囲に入るのかを確認しています。
フォルダ内のファイルを走査する
探索フォルダの当たりがついたところで、今度はその中のファイルを全て返す処理です。
ここで、もし探索フォルダの中のファイル（wiki_00からwiki_99まで）を探しても見つからなかった場合、2つの選択肢があります。
- 念のためすべてのフォルダのファイルも探す
- 諦めて次のページIDに処理を移す
一応2つとも実装（1がget_all_files、2がget_files）しましたが、私が試したところでは、与えられた探索フォルダになかった場合は、すべてのフォルダのファイルを走査しても見つかりませんでした。すべて確認したわけではありませんが、csvのページIDに対して転送ページが存在しているようなもの（IDが変わっている?）は見つからないようです。
プログラムの実行結果
csvファイルの件数が2961件で処理時間は約10分、見つからなかったページIDは45件でした。一度、簡単にデータを前から順に走査するプログラムを書いて試してみましたが、csvファイルが200件くらいでも30分くらいかかったので、この件数だと総当たりでは数時間、あるいはもっとかかったと思います。
テキストデータは少し変えたりして何度も実験してみたいので、少しでも速いものが作れてよかったと思います。正直、タグを取り除く正規表現の部分などは自分でも拙いコードだと思いますが、取りあえず目的は果たすことができました。
今回、PetScanでカテゴリを「医薬品」、深度を「2」とし、ダウンロードしたcsvファイルに対してこのプログラムを動かすと出力先フォルダにはファイルが2700近く作成されました*3。これらのファイル一つ一つに、次のような抽出されたテキスト（記事）が入っています。
最後に、プログラムで使用するフォルダやファイルの位置関係を示しておきます。
コード全文です。
import os import re import csv import glob import time from collections import OrderedDict # ウィキペディアからダウンロードした全データファイルのidとPetScanにて # 収集した特定カテゴリのページIDを突き合わせて、特定カテゴリだけの # テキストデータを作成する. # key: フォルダ名, value: そのフォルダ中の最大idの辞書を作成して返す # 各ディレクトリの最後のファイルの最後の記事からid(=そのディレクトリの最大のid)を抜き出して辞書の値とする def mapping_dir(dir_path): dir_list = os.listdir(dir_path) for _dir in dir_list: last_f_path = sorted(glob.glob(os.path.join(dir_path, _dir) + '\\*'))[-1] max_id, _, _ = format_text(sorted(get_articlelist(last_f_path))[-1]) dir_info[_dir] = max_id return dir_info # csvファイルから読み込んだpage_idとmapping_dirで作成した辞書(dir_info)から # page_idがあると思われるフォルダ名を探索フォルダとして返す def get_search_dir(page_id, dir_info): min_id = 0 for dir_name, max_id in dir_info.items(): search_dir = dir_name if min_id <= page_id <= max_id: return search_dir else: min_id = max_id return next(iter(dir_info)) # 見つからなかった場合は最初のkey(AA)を返す # search_dirで見つからなかった場合は一応最初(AA)から最後のフォルダまで探索する. # 全探索しても見つからない場合がほとんど(?). 使っていませんが, 下のget_filesと置換可. def get_all_files(base_path, search_dir): dir_iter = iter(dir_info.keys()) # もしsearch_dirになかったら, 順番に取り出してみていく dir_list = list(dir_info.keys()) # keyのリスト. 一巡したかの終了判定で使う cnt = 1 while True: search_dir_path = os.path.join(base_path, search_dir) print(search_dir_path) for _, _, files in os.walk(search_dir_path): for file in files: yield os.path.join(search_dir_path, file) search_dir = next(dir_iter) cnt += 1 if search_dir == dir_list[-1] and cnt == len(dir_list): # AAから最後まで一巡したが見つからなかった場合 return # search_dir内で見つからなかった場合は諦める処理だと速い. こちらを使っています def get_files(base_path, search_dir): while True: search_dir_path = os.path.join(base_path, search_dir) print(search_dir_path) for _, _, files in os.walk(search_dir_path): for file in files: yield os.path.join(search_dir_path, file) else: return # 1つの<doc>タグ入りの記事からid, タイトル, 本文を抽出して返す def format_text(data): head = head_reg.search(data) # 先頭のヘッダ部分(<doc id=...>)を抜き出し id_tag = id_reg.search(head.group()) doc_id = re.search(r'\d+', id_tag.group()) # ヘッダ部分からidを抽出 title_tag = title_reg.search(head.group()) doc_title = re.search(r'".+"', title_tag.group()) # ヘッダ部分からtitleを抽出 doc_id = int(doc_id.group()) title = doc_title.group().replace('"', '') # ダブルクォーテーションを削除 # ヘッダ部分(<doc id=...>)と末尾のタグを削除, 先頭のタイトル分のみスライスで削除 # (title titleのように連続しているため)して本文のみに text = data.replace(head.group(), '').rstrip('</doc>')[len(title):] return doc_id, title, text # ファイルを読み込み記事のリストを返す def get_articlelist(f_path): f = open(f_path, 'r', encoding='utf-8') file_data = f.read() file_data = file_data.replace('\n', '') return doc_reg.findall(file_data) def output_file(f_name, text): # ファイル名に使えない文字がある場合は削除 f_name = re.sub(r'¥|/|:|\*|\?|"|<|>|\||', '', f_name) path = OUTPUT_PATH + f_name f = open(path, 'w', encoding='utf-8') f.write(text) f.close() if __name__ == '__main__': # 用いる正規表現. コンパイルしておく id_reg = re.compile(r'id="\d+"') title_reg = re.compile(r'title=".+"') doc_reg = re.compile(r'<doc id=.+?</doc>') head_reg = re.compile(r'<doc id.+?">') # フォルダやファイルの場所. 各自の環境に合わせて変更する BASE_PATH = 'D:\\data\\wiki_data\\' CSV_FILE = BASE_PATH + 'download_csv_医薬品' # PetScanでダウンロードしたファイル. 適当に名前を付ける WIKI_DIR = BASE_PATH + 'wikipedia' OUTPUT_PATH = 'D:\\data\\医薬品\\' # 先にフォルダ名とその中で最大のidの辞書を作成しておく dir_info = OrderedDict() dir_info = mapping_dir(WIKI_DIR) counter = 1 # ループカウンタ. 表示用 # csvファイルの読み込み. 先頭行は飛ばしておく f = open(CSV_FILE, 'r', encoding='utf-8') reader = csv.reader(f) next(csv.reader(f)) not_found = {} # 見つからなかったページIDとページ名 # ループの処理時間を計測 start_time = time.time() for line in reader: flag = False page_name = line[1] page_id = int(line[2]) namespace = line[3] if namespace == "Category" or namespace == "Template": continue search_dir = get_search_dir(page_id, dir_info) for f_path in get_files(WIKI_DIR, search_dir): for d in get_articlelist(f_path): doc_id, title, text = format_text(d) if page_id == doc_id: print(counter, title) output_file(title, text) counter += 1 flag = True break if flag: break else: not_found[page_id] = page_name print("ページID: {}は見つかりませんでした.".format(page_id)) end_time = time.time() processing_time = end_time - start_time print('processing_time(sec): ', processing_time) # 見つからなかったページIDとタイトルを表示 for i, (_id, _title) in enumerate(not_found.items()): print(i, _id, _title)
