keywords/src/wikidata_search.py

   1 # generate a list of wikidata items related to keywords
   2 from os import path
   3 from sys import stdout
   4 from wikidata_api_calls import search_wikidata, get_wikidata_api
   5 import csv
   6 from itertools import chain
   7
   8 class Wikidata_ResultSet:
   9     def __init__(self):
  10         self.results = []
  11
  12     def extend(self, term, results):
  13         self.results.append(
  14             (Wikidata_Result(term, result, i)
  15              for i, result in enumerate(results))
  16         )
  17
  18     def to_csv(self, outfile=None, mode='w'):
  19         if outfile is None:
  20             of = stdout
  21
  22         else:
  23             if path.exists(outfile) and mode != 'w':
  24                 of = open(outfile,'a',newline='')
  25             else:
  26                 of = open(outfile,'w',newline='')
  27         writer = csv.writer(of)
  28         writer.writerow(Wikidata_Result.__slots__)
  29         writer.writerows(map(Wikidata_Result.to_list, chain(* self.results)))
  30
  31
  32 class Wikidata_Result:
  33     # store unique items found in the search results, the position in the search result, and the date
  34     __slots__=['search_term','itemid','pageid','search_position','timestamp']
  35
  36     def __init__(self,
  37                  term,
  38                  search_result,
  39                  position):
  40
  41         self.search_term = term.strip()
  42         self.itemid = search_result['title']
  43         self.pageid = int(search_result['pageid'])
  44         self.search_position = int(position)
  45         self.timestamp = search_result['timestamp']
  46
  47     def to_list(self):
  48         return [self.search_term,
  49                 self.itemid,
  50                 self.pageid,
  51                 self.search_position,
  52                 self.timestamp]
  53
  54 def run_wikidata_searches(terms):
  55     api = get_wikidata_api()
  56     resultset = Wikidata_ResultSet()
  57     for term in terms:
  58         search_results = search_wikidata(api, term)
  59         resultset.extend(term, search_results)
  60     return resultset
  61
  62 def read_google_trends_files(terms_files):
  63     def _read_file(infile):
  64         return csv.DictReader(open(infile,'r',newline=''))
  65
  66     for row in chain(* [_read_file(terms_file) for terms_file in terms_files]):
  67         yield row['query']
  68
  69
  70 def trawl_google_trends(terms_files, outfile = None, mode='w'):
  71     terms = list(read_google_trends_files(terms_files))
  72     resultset = run_wikidata_searches(terms)
  73     resultset.to_csv(outfile, mode)
  74
  75 def trawl_base_terms(infiles, outfile = None, mode='w'):
  76     terms = list(chain(* (open(infile,'r') for infile in infiles)))
  77     resultset = run_wikidata_searches(terms)
  78     resultset.to_csv(outfile, mode)
  79
  80     ## search each of the base terms in wikidata
  81
  82     # store unique items found in the search results, the position in the search result, and the date
  83
  84 if __name__ == "__main__":
  85     import argparse
  86     parser = argparse.ArgumentParser("Search wikidata for items related to a set of terms.")
  87     parser.add_argument('inputs', type=str, nargs='+', help='one or more files to read')
  88     parser.add_argument('--use-gtrends', action='store_true', help = 'toggle whether the input is the output from google trends')
  89     parser.add_argument('--output', type=str, help='an output file. defaults to stdout')
  90     parser.add_argument('--overwrite', action='store_true', help = 'overwrite existing output files instead of appending')
  91     args = parser.parse_args()
  92     if args.use_gtrends:
  93         trawl_google_trends(args.inputs, args.output)
  94     else:
  95         trawl_base_terms(args.inputs, args.output)