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, 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 = wd_res.get('entity',None).get('value',None)
  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 def GetAllLabels(in_csvs, outfile, topNs):
  23
  24     def load_entity_ids(in_csv, topN=5):
  25         with open(in_csv,'r',newline='') as infile:
  26             reader = list(csv.DictReader(infile))
  27             for row in reader:
  28                 if int(row['search_position']) < topN:
  29                     yield row["entityid"]
  30
  31     ids = set(chain(* map(lambda in_csv, topN: load_entity_ids(in_csv, topN), in_csvs, topNs)))
  32
  33     labeldata = GetEntityLabels(ids)
  34
  35     with open(outfile, 'w', newline='') as of:
  36         writer = csv.writer(of)
  37         writer.writerow(LabelData.__slots__)
  38         writer.writerows(map(LabelData.to_list,labeldata))
  39
  40
  41 def GetEntityLabels(entityids):
  42
  43     def run_query_and_parse(query, is_alt):
  44         results = run_sparql_query(query)
  45         try:
  46             jobj = results.json()
  47
  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
  54             for info in res:
  55                 yield LabelData(info, is_alt)
  56
  57         except JSONDecodeError as e:
  58             print(e)
  59             print(query)
  60
  61     def prep_query(query, prop, entityids):
  62         values = ' '.join(('wd:{0}'.format(id) for id in entityids))
  63         return query.format(prop, values)
  64
  65     base_query = """
  66     SELECT DISTINCT ?entity ?label WHERE {{
  67     ?entity {0} ?label;
  68     VALUES ?entity  {{ {1} }}
  69     }}"""
  70
  71     # we can't get all the entities at once. how about 100 at a time?
  72     chunksize = 100
  73     entityids = (id for id in entityids)
  74     chunk = list(islice(entityids, chunksize))
  75     calls = []
  76     while len(chunk) > 0:
  77         label_query = prep_query(base_query, "rdfs:label", chunk)
  78         altLabel_query = prep_query(base_query, "skos:altLabel", chunk)
  79         label_results = run_query_and_parse(label_query,  is_alt=False)
  80         altLabel_results = run_query_and_parse(altLabel_query, is_alt=True)
  81         calls.extend([label_results, altLabel_results])
  82         chunk = list(islice(entityids, chunksize))
  83
  84     return chain(*calls)
  85
  86
  87 if __name__ == "__main__":
  88     import argparse
  89     parser = argparse.ArgumentParser("Use wikidata to find transliterations of terms")
  90     parser.add_argument('inputs', type=str, nargs='+', help='one or more files to read. the inputs are generated by wikidata_search.py')
  91     parser.add_argument('--topN', type=int, nargs='+', help='limit number of wikidata search results to use, can pass one arg for each source.')
  92     parser.add_argument('--output', type=str, help='an output file. defaults to stdout',default=20)
  93
  94     args = parser.parse_args()
  95
  96     GetAllLabels(args.inputs, args.output, topNs=args.topN)