transliterations/src/wikidata_transliterations.py

   1 from wikidata_api_calls import run_sparql_query
   2 from itertools import chain, islice
   3 import csv
   4 from json import JSONDecodeError
   5
   6 class LabelData:
   7     __slots__ = ['entityid','label','langcode','is_alt']
   8
   9     def __init__(self, wd_res, entityid, is_alt):
  10         obj = wd_res.get('label',None)
  11         self.label = obj.get('value',None)
  12         self.langcode = obj.get('xml:lang',None)
  13         self.entityid = entityid
  14         self.is_alt = is_alt
  15
  16     def to_list(self):
  17         return [self.entityid,
  18                 self.label,
  19                 self.langcode,
  20                 self.is_alt]
  21
  22
  23 def GetAllLabels(in_csv, outfile, topN):
  24
  25     def load_entity_ids(in_csv, topN=5):
  26         with open(in_csv,'r',newline='') as infile:
  27             reader = csv.DictReader(infile)
  28             for row in reader:
  29                 if int(row['search_position']) < topN:
  30                     yield row["entityid"]
  31
  32     ids = set(load_entity_ids(in_csv, topN))
  33
  34     labeldata = chain(* map(GetEntityLabels, ids))
  35
  36     with open(outfile, 'w', newline='') as of:
  37         writer = csv.writer(of)
  38         writer.writerow(LabelData.__slots__)
  39         writer.writerows(map(LabelData.to_list,labeldata))
  40
  41
  42 def GetEntityLabels(entityid):
  43
  44     def run_query_and_parse(query, entityid, is_alt):
  45         results = run_sparql_query(query % entityid)
  46         try:
  47             jobj = results.json()
  48             res = jobj.get('results',None)
  49             if res is not None:
  50                 res = res.get('bindings',None)
  51             if res is None:
  52                 raise requests.APIError(f"got invalid response from wikidata for {query % entityid}")
  53             for info in res:
  54                 yield LabelData(info, entityid, is_alt)
  55
  56         except JSONDecodeError as e:
  57             print(e)
  58             print(query % entityid)
  59
  60
  61     label_base_query = """
  62     SELECT DISTINCT ?label WHERE {
  63     wd:%s rdfs:label ?label;
  64     }"""
  65
  66     altLabel_base_query = """
  67     SELECT DISTINCT ?label WHERE {
  68     wd:%s skos:altLabel ?label;
  69     }"""
  70
  71     label_results = run_query_and_parse(label_base_query, entityid, is_alt=False)
  72
  73     altLabel_results = run_query_and_parse(altLabel_base_query, entityid, is_alt=True)
  74
  75     return chain(label_results, altLabel_results)
  76
  77
  78 if __name__ == "__main__":
  79     GetAllLabels("../data/output/wikidata_search_results.csv","../data/output/wikidata_entity_labels.csv", topN=20)