国会会議録検索システムAPIを使って国会発言をJSONで取得しCSV出力する

はじめに

政治の世界ってわかりにくいですよね。日々揚げ足取りみたいな議論が行われている様子や失言した議員さんなどがニュースに取り上げられがちですが、実際には様々な議論が交わされています（きっと）。
そんな議論の内容、議事録データを国立国会図書館がAPIで提供してくれていて、政治のオープン化やデータ解析の題材としてとても有用だと感じました。

本記事では、そのデータをJSON形式で取得してCSV出力するPythonプログラムを作成したのでそちらを紹介します。

国会議事録APIの利用例と本記事の要点について

国会会議録検索システムAPIの公式説明を抜粋すると下記のように紹介されています。

「国会会議録検索システム」は、第1回国会（昭和22（1947）年5月開会）以降の国会会議録を検索・閲覧することができるデータベースです。国立国会図書館が、衆議院・参議院と共同で提供しています。

実はこの記事執筆のちょうど1年ほど前に、別の方がQiitaで同様の記事を書いてくださっています。この中で、国会会議録検索システムについての説明や、取得したデータの活用例を紹介されていますので是非そちらもご参照ください。

一方、その記事中で当該APIの課題として「APIのレスポンスがJSON形式はなくXML形式のみ」といった点が挙げられているのですが、おそらく2020年にAPIの仕様が改善されたのかJSONでのデータ取得ができるようになっていたので、二番煎じながらJSONデータで取得するケースを紹介させていただく次第です。

国会議事録APIからのデータ取得プログラム

私の実行環境 >> OS: macOS Big Sur / Python 3.6.6

KokkaiSpeech.py

import requests
import time
import json
import sys
import csv
import re

# ベースとなるURL
base_url = "https://kokkai.ndl.go.jp/api/speech"

# URLパラメータ用の辞書を空の状態で用意し、後から順次格納する。01はヒット総数の確認用、02はデータ取得用。
params_01 = {} 
params_02 = {}

# パラメータを対話的に入力する
text_input = input('検索する文字列を入力（Enterキーでスキップ） >> ')
if text_input != "":
    params_01['any'] = str(text_input)
    params_02['any'] = str(text_input)
else:
    pass

speaker_input = input('検索する発言者名を入力（Enterキーでスキップ） >> ')
if speaker_input != "":
    params_01['speaker'] = str(speaker_input)
    params_02['speaker'] = str(speaker_input)
else:
    pass

from_input = input('開始日を入力 (e.g. 2020-09-01) >> ')
if re.match(r'[0-9]{4}-[0-1][0-9]-[0-3][0-9]', from_input): #正規表現によるパターンマッチングにて入力値が有効か判定
    params_01['from'] = str(from_input)
    params_02['from'] = str(from_input)
else:
    params_01['from'] = "2020-09-01"
    params_02['from'] = "2020-09-01"
    print("'From' date is set to 2020-09-01 due to invalid input")

until_input = input('終了日を入力 (e.g. 2020-11-30) >> ')
if re.match(r'[0-9]{4}-[0-1][0-9]-[0-3][0-9]', until_input): #正規表現によるパターンマッチングにて入力値が有効か判定
    params_01['until'] = str(until_input)
    params_02['until'] = str(until_input)
else:
    params_01['until'] = "2020-09-30"
    params_02['until'] = "2020-09-30"
    print("'Until' date is set to 2020-09-30 due to invalid input")

params_01['maximumRecords'] = 1
params_01['recordPacking'] = "json"

response_01 = requests.get(base_url, params_01) #URLのパラメータをエンコードしてAPIへリクエスト
jsonData_01 = response_01.json() #APIからのレスポンスをJSON形式で取得

# レスポンスに含まれているヒット件数を確認（レスポンスのJSONにレコード数の項目がない場合はクエリに問題ありと判断しエラー終了）
try:
    total_num = jsonData_01["numberOfRecords"]
except:
    print("クエリエラーにより取得できませんでした。")
    sys.exit()

# 件数を表示し、データ取得を続行するか確認をとる
next_input = input("検索結果は " + str(total_num) + "件です。\nキャンセルする場合は 1 を、データを取得するにはEnterキーまたはその他を押してください。 >> ")
if next_input == "1":
    print('プログラムをキャンセルしました')
    sys.exit()
else:
    pass

max_return = 100 #発言内容は一回のリクエストにつき100件まで取得可能なため、その上限値を取得件数として設定
pages = (int(total_num) // int(max_return)) + 1 #ヒットした全件を取得するために何回リクエストを繰り返すか算定

# 全件取得用のパラメータを設定
params_02['maximumRecords'] = max_return
params_02['recordPacking'] = "json"

Records = [] #取得データを格納するための空リストを用意

# 全件取得するためのループ処理
i = 0
while i < pages:
    i_startRecord = 1 + (i * int(max_return))
    params_02['startRecord'] = i_startRecord
    response_02 = requests.get(base_url, params_02)
    jsonData_02 = response_02.json()
    #JSONデータ内の各発言データから必要項目を指定してリストに格納する
    for list in jsonData_02['speechRecord']:
        list_id = list['speechID']
        list_kind = list['imageKind']
        list_house = list['nameOfHouse']
        list_topic = list['nameOfMeeting']
        list_issue = list['issue']
        list_date = list['date']
        list_order = list['speechOrder']
        list_speaker = list['speaker']
        list_group = list['speakerGroup']
        list_position = list['speakerPosition']
        list_role = list['speakerRole']
        list_speech = list['speech'].replace('\r\n', ' ').replace('\n', ' ') #発言内容の文中には改行コードが含まれるため、これを半角スペースに置換
        list_url01 = list['speechURL']
        list_url02 = list['meetingURL']
        Records.append([list_id, list_kind, list_house, list_topic, list_issue, list_date, list_order, list_speaker, list_group, list_position, list_role, list_speech, list_url01, list_url02])

    sys.stdout.write("\r%d/%d is done." % (i+1, pages)) #進捗状況を表示する
    i += 1
    time.sleep(1) #リクエスト１回ごとに若干時間をあけてAPI側への負荷を軽減する

# CSVへの書き出し
with open("kokkai_speech_" + str(total_num) + ".csv", 'w', newline='') as f:
    csvwriter = csv.writer(f, delimiter=',', quotechar='"', quoting=csv.QUOTE_NONNUMERIC) #CSVの書き出し方式を適宜指定
    csvwriter.writerow(['発言ID', '種別', '院名', '会議名', '号数', '日付', '発言番号', '発言者名', '発言者所属会派', '発言者肩書き', '発言者役割', '発言内容', '発言URL', '会議録URL'])
    for record in Records:
        csvwriter.writerow(record)

解説

上記のコード内コメントでほとんど解説を入れていますが、ポイントとしては下記となります。基本的には、API側に過剰な負荷をかけないような作法を心掛けて構成しています。

• コンソール上で対話的に検索クエリを入力していく形にしている。
• 検索クエリは発言内容の文字列や発言者名、期間を指定可能。必要なものを入力し不要なものはスキップできる。なお、期間のクエリをスキップした場合やそこに日付として不適な文字列が入った場合は、2020年9月の期間が設定されるようにプログラム上で条件分岐している。
• APIの仕様上、１回のリクエストで発言データは100件までしか取得できないので、まずは1件のみのリクエストを投げてそのレスポンスから検索クエリでのヒット総件数を確認し、その後に必要回数リクエストを繰り返してヒットした全件を取得する。
• 検索クエリでのヒット総件数を表示した時点でデータ取得をキャンセルできるようにした（検索結果が想定より多すぎる、少なすぎるといった際に無駄なデータ取得をしないように配慮）。
• プログラムの過程でAPIに繰り返しリクエストする際、API側への負荷軽減のため若干のインターバルを入れる。

コンソール上での動作例

例として、「宇宙」を含む今年の国会発言を取得してみましょう。
下記の要領でPythonコードを実行すると、コンソール上で検索文字列や検索開始日、終了日などの入力が順次求められます。必要なものは入力し、不要であればスキップします。
検索文字列は半角スペース区切りでOR検索が出来たりしますが、細かい仕様は当該APIの公式サイトをご覧ください。
その後は検索結果の件数が表示されるので、Enterキーなどを押して処理を進めます。
今回の例では268件が該当したので、3回に分けてプログラムが自動でリクエストを投げます。この進捗状況もコンソール上に表示されます。

shell

python KokkaiSpeech.py
検索する文字列を入力（Enterキーでスキップ） >> 宇宙
検索する発言者名を入力（Enterキーでスキップ） >> 
開始日を入力 (e.g. 2020-09-01) >> 2020-01-01
終了日を入力 (e.g. 2020-11-30) >> 2020-12-15
検索結果は 268件です。
キャンセルする場合は 1 を、データを取得するにはEnterキーまたはその他を押してください。 >>  
3/3 is done.

処理が終了したら、プログラムがあるフォルダに「kokkai_speech_268.csv」という名称のCSVファイルが出力されます。

おわりに

私も技術的にまだまだ勉強中の身ですが、APIを介したJSON形式でのデータ取得は一般的によく実施されるケースだと思いますので、そうした実務においても参考になれば幸いです。

また、本記事で紹介した私のプログラムは「発言内容」単位のものですが、当該APIでは「会議」単位での取得も可能になっていますので、そのあたりは適宜補正してお使いください。その際は、APIサーバ側に極端な負荷をかけないようご留意ください。

こうした議事録を文字起こしして配信いただいている国会図書館や関係者の方々に感謝しつつ、政治のオープン化、そしてより良い社会作りに役立てていきたいですね。