wd_importPEP/wikidata/wd_PEP.py

import os

import yaml
import json


class wikidata_PEP(object):
    def __init__(self, config_file):
        with open(config_file, "r") as stream:
            try:
                self.config = yaml.safe_load(stream)
            except yaml.YAMLError as exc:
                print(exc)

    def import_members_of_parliament_dict(self, list_of_countries):
        self.full_dictionary_member_lists = {}

        for country in list_of_countries:
            print("started to parse data of members of " + country + " ..")

            f = open("crawlers/output/" + country + "MemberList.txt")
            text = f.read()

            self.full_dictionary_member_lists[country] = eval(text)

        # print(self.full_dictionary_member_lists)

    def check_for_entity_ids(self, list_of_countries):
        from wikibaseintegrator import WikibaseIntegrator
        from wikibaseintegrator import wbi_helpers

        full_dictionary_member_lists = self.full_dictionary_member_lists

        for country in list_of_countries:
            for member_id in full_dictionary_member_lists[country].keys():
                name = full_dictionary_member_lists[country][member_id]["name"]

                results = wbi_helpers.search_entities(search_string=name)

                for entity_id in results:
                    wbi = WikibaseIntegrator()
                    wikidata_item = wbi.item.get(entity_id=entity_id)

                    for claim_key in wikidata_item.get_json()["claims"].keys():
                        if claim_key == "P31":
                            if (
                                wikidata_item.get_json()["claims"][claim_key][0][
                                    "mainsnak"
                                ]["datavalue"]["value"]["id"]
                                == "Q5"
                            ):
                                print(entity_id)
                                print("---------")
                                print(name)
                                print("is a human")

    def create_member_on_wikidata(self):
        from wikibaseintegrator import wbi_login, WikibaseIntegrator
        from wikibaseintegrator.datatypes import ExternalID, Item
        from wikibaseintegrator.wbi_config import config as wbi_config

        wbi_config["USER_AGENT"] = "PEPimport/1.0 (https://www.wikidata.org/wiki/User:)"

        # login object
        login_instance = wbi_login.OAuth2(consumer_token="", consumer_secret="")

        wbi = WikibaseIntegrator(login=login_instance)

        # data type object, e.g. for a NCBI gene entrez ID
        is_human = Item(value="Q5", prop_nr="P31")
        occupation_politician = ExternalID(value="Q82955", prop_nr="P106")
        occupation_deputy = ExternalID(value="Q1055894", prop_nr="P106")
        # referenceURL = URL(value='http://legislacion.asamblea.gob.ni/Tablas%20Generales.nsf/InfoDiputado.xsp?documentId=3D4CFDC4B3006D70062587C5007C29E1&action=openDocument&SessionID=1868803A06AB73D50B7F89BD0AB', prop_nr='P106')

        # print(is_human)
        # print(referenceURL)

        # data goes into a list, because many data objects can be provided to
        data1 = [is_human]
        data2 = [occupation_deputy]
        data3 = [occupation_politician]

        # Create a new item
        item = wbi.item.new()

        # Set an english label
        item.labels.set(language="en", value="Carlos Humberto Ruíz")

        # Carlos Humberto Ruiz has the Qid Q116918332

        # Set a French description
        item.descriptions.set(
            language="en", value="Nicaraguan National Assembly Deputy"
        )

        item.claims.add(data1)
        # item.claims.add(data2)
        # item.claims.add(data3)
        print(item.write())

    def editMemberOnWikidata(self, Qid):
        from wikibaseintegrator import wbi_login, WikibaseIntegrator
        from wikibaseintegrator.datatypes import ExternalID, Item, Time, String
        from wikibaseintegrator.wbi_config import config as wbi_config
        from wikibaseintegrator.wbi_enums import ActionIfExists
        from wikibaseintegrator.wbi_enums import WikibaseDatePrecision

        wbi_config[
            "USER_AGENT"
        ] = "PEPimport/1.0 (https://www.wikidata.org/wiki/User:Alpcentaur)"

        # login object
        login_instance = wbi_login.OAuth2(consumer_token="", consumer_secret="")

        wbi = WikibaseIntegrator(login=login_instance)

        # data type object, e.g. for a NCBI gene entrez ID
        # is_human = Item(value='Q5', prop_nr='P31')
        # occupation_politician = Item(value='Q82955', prop_nr='P106')
        # occupation_deputy = Item(value='Q1055894', prop_nr='P106')
        # referenceURL = ExternalID(value='http://legislacion.asamblea.gob.ni/Tablas%20Generales.nsf/InfoDiputado.xsp?documentId=3D4CFDC4B3006D70062587C5007C29E1&action=openDocument&SessionID=1868803A06AB73D50B7F89BD0AB', prop_nr='P854')

        # print(is_human)
        # print(referenceURL)

        references = [
            [
                ExternalID(
                    value="http://legislacion.asamblea.gob.ni/Tablas%20Generales.nsf/InfoDiputado.xsp?documentId=3D4CFDC4B3006D70062587C5007C29E1&action=openDocument&SessionID=1868803A06AB73D50B7F89BD0AB",
                    prop_nr="P854",
                ),
                Time(
                    time="+2023-02-27T00:00:00Z",
                    prop_nr="P813",
                    precision=WikibaseDatePrecision.DAY,
                ),
            ]
        ]

        occupation_deputy = Item(value="Q1055894", prop_nr="P106", references=references)

        ## data goes into a list, because many data objects can be provided to
        # data1 = [is_human]
        data2 = [occupation_deputy]
        # data3 = [occupation_politician]
        # data4 = [referenceURL]
        ## get item for Qid
        item = wbi.item.get(entity_id=Qid)

        # print(item.claims)

        # Set an english label
        # item.labels.set(language='en', value='Carlos Humberto Ruíz', action_if_exists=ActionIfExists.KEEP)

        # Set a French description
        # item.descriptions.set(language='en', value='Nicaraguan National Assembly Deputy', action_if_exists=ActionIfExists.KEEP)

        # item.claims.add(data4)
        item.claims.add(data2)
        # item.claims.add(data3)
        print(item.write())