API, URIとPythonを使って、Wikidataからデータを取得する (5) 実装の要点とコード

import http.client
import urllib.parse
import json

def get_data(url):
    """URLで指定された場所からデータを取得する。

    URLで指定された先に、HTTPのGETリクエストを送信し、レスポンスをUTF-8文字列として返す。

    Args:
        url (str): URLを表す文字列。クエリが後ろにくっついている。

    Returns:
        str: レスポンス文字列。

    """
    # URLを構成要素に解析
    parsed_url = urllib.parse.urlparse(url)
    host = parsed_url.hostname
    path = parsed_url.path + "?" + parsed_url.query
    scheme = parsed_url.scheme
    
    if not path:
        path = '/'
    
    # HTTPSが使えれば使う
    if scheme == 'https':
        connection = http.client.HTTPSConnection(host)
    else:
        connection = http.client.HTTPConnection(host)
    
    # GETリクエスト送信
    connection.request('GET', path)
    
    # レスポンス取得
    response = connection.getresponse()
    data = response.read()
    
    # 接続を閉じる
    connection.close()
    
    return data.decode('utf-8')

def entity_name_to_id(query):
    """エンティティ名をエンティティIDに変換する。

    エンティティ名をAPIで検索する。見つかればそのエンティティIDを文字列として返す。

    Args:
        query (str): 検索文字列。

    Returns:
        str: エンティティID。1つも検索にヒットしなかった場合は、"NULL"が返る。

    """
    encoded_query = urllib.parse.quote(query, encoding='UTF-8')
    url = "https://www.wikidata.org/w/api.php?action=wbsearchentities&language=ja&format=json&search=" + encoded_query

    json_str = get_data(url)

    try:
        json_dict = json.loads(json_str) # json文字列をpythonの辞書に
        entity_list = json_dict['search']

        if type(entity_list) == None: # 検索にヒットせず
            return "NULL"
        else:
            # 大抵は最も若い番号のものが探してるやつだが、不安なら完全一致するかしっかり調べたほうが良い
            return entity_list[0].get('id') 
        
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")
        exit()

def property_name_to_id(query):
    """プロパティ名をプロパティIDに変換する。

    プロパティ名をAPIで検索する。見つかればそのプロパティIDを文字列として返す。

    Args:
        query (str): 検索文字列。

    Returns:
        str: プロパティID。1つも検索にヒットしなかった場合は、"NULL"が返る。

    """
    encoded_query = urllib.parse.quote(query, encoding='UTF-8')
    url = "https://www.wikidata.org/w/api.php?action=wbsearchentities&language=ja&format=json&type=property" + "&search=" + encoded_query
    json_str = get_data(url)
    try:
        json_dict = json.loads(json_str) # 辞書に変換
        prop_list = json_dict['search']
        if type(prop_list) == None:
            return "NULL"
        else:
            return prop_list[0].get('id') # ここも不安なら、完全一致するか調べたほうが良い
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")
        exit()

def id_to_name(id):
    """エンティティ、プロパティIDをエンティティ、プロパティ名に変換する。

    URIで該当するIDのデータを取得し、ラベルを取得する。

    Args:
        id (str): 検索ID。

    Returns:
        str: ラベル。

    """
    id_head = id[0]
    if (id_head == 'P' or id_head == 'Q'):
        url = "https://www.wikidata.org/wiki/Special:EntityData/" + id +  ".json"
        entity_dict = json.loads(get_data(url))
        value_dict = entity_dict['entities'][id]['labels']['ja']
        if (value_dict == None):
            print("NULL")
            exit()
        else:
            return value_dict['value']
    else:
        print("IDが不正です。")
        exit()

def entity_prop_to_vals(entity_id, prop_id):
    """「エンティティ」の「プロパティ」に該当する「値」を返す。

    エンティティのデータをURIで取得し、JSONの中から該当するデータをプロパティIDを使って見つける。

    Args:
        entity_id (str): エンティティID。
        prop_id (str): プロパティID。

    Returns:
        str: 値のリスト。

    """
    vals = []

    url = "https://www.wikidata.org/wiki/Special:EntityData/" + entity_id +  ".json"
    entity_dict = json.loads(get_data(url))

    claim_dict = entity_dict['entities'][entity_id]['claims']
    if (claim_dict != None):
        if (prop_id in claim_dict):
            prop_list = claim_dict[prop_id]
            for prop_dict in prop_list:
                val = prop_dict['mainsnak']['datavalue']['value']
                if (type(val) == str):
                    vals.append(val)
                elif (type(val) == dict):
                    id = val['id']
                    jaLabel = id_to_name(id)
                    vals.append(jaLabel)
                else:
                    vals.append(val.to_string())
        else:
            return []
    else:
        exit()
    return vals


def main():
    """今までの関数のテスト。日本の首都を求めさせる。
    """
    entity_id = entity_name_to_id("日本")
    print(entity_id)
    prop_id = property_name_to_id("首都")
    print(prop_id)
    vals = entity_prop_to_vals(entity_id, prop_id)
    print(vals)


if __name__ == "__main__":
    main()