text: #!/usr/bin/env python3
  # Copyright 2017-present, Facebook, Inc.
  # All rights reserved.
  #
  # This source code is licensed under the license found in the
  # LICENSE file in the root directory of this source tree.
  """Evaluate the accuracy of the WRMCQA retriever module."""
  
  try:
      import regex as re
  except ImportError:
      import re
  import logging
  import argparse
  try:
      import ujson as json
  except ImportError:
      import json
  import time
  import os
  
  from multiprocessing import Pool as ProcessPool
  from multiprocessing import cpu_count
  from multiprocessing.util import Finalize
  from functools import partial
  from wrmcqa import retriever, tokenizers
  from wrmcqa.retriever import utils
  
  # ------------------------------------------------------------------------------
  # Multiprocessing target functions.
  # ------------------------------------------------------------------------------
  
  PROCESS_TOK = None
  PROCESS_DB = None
  
  
  def init(tokenizer_class, tokenizer_opts, db_class, db_opts):
      global PROCESS_TOK, PROCESS_DB
      PROCESS_TOK = tokenizer_class(**tokenizer_opts)
      Finalize(PROCESS_TOK, PROCESS_TOK.shutdown, exitpriority=100)
      PROCESS_DB = db_class(**db_opts)
      Finalize(PROCESS_DB, PROCESS_DB.close, exitpriority=100)
  
  
  def regex_match(text, pattern):
      """Test if a regex pattern is contained within a text."""
      try:
          pattern = re.compile(
              pattern,
              flags=re.IGNORECASE + re.UNICODE + re.MULTILINE,
          )
      except BaseException:
          return False
      return pattern.search(text) is not None
  
  
  def has_answer(answer, doc_id, match):
      """Check if a document contains an answer string.
  
      If `match` is string, token matching is done between the text and answer.
      If `match` is regex, we search the whole text with the regex.
      """
      global PROCESS_DB, PROCESS_TOK
      text = PROCESS_DB.get_doc_text(doc_id)
      text = utils.normalize(text)
      if match == 'string':
          # Answer is a list of possible strings
          text = PROCESS_TOK.tokenize(text).words(uncased=True)
          for single_answer in answer:
              single_answer = utils.normalize(single_answer)
              single_answer = PROCESS_TOK.tokenize(single_answer)
              single_answer = single_answer.words(uncased=True)
              for i in range(0, len(text) - len(single_answer) + 1):
                  if single_answer == text[i: i + len(single_answer)]:
                      return True
      elif match == 'regex':
          # Answer is a regex
          single_answer = utils.normalize(answer[0])
          if regex_match(text, single_answer):
              return True
      return False
  
  
  def get_score(answer_doc, match):
      """Search through all the top docs to see if they have the answer."""
      answer, (doc_ids, doc_scores) = answer_doc
      for doc_id in doc_ids:
          if has_answer(answer, doc_id, match):
              return 1
      return 0
  
  
  # ------------------------------------------------------------------------------
  # Main
  # ------------------------------------------------------------------------------
  
  
  if __name__ == '__main__':
      logger = logging.getLogger()
      logger.setLevel(logging.INFO)
      fmt = logging.Formatter('%(asctime)s: [ %(message)s ]',
                              '%m/%d/%Y %I:%M:%S %p')
      console = logging.StreamHandler()
      console.setFormatter(fmt)
      logger.addHandler(console)
  
      parser = argparse.ArgumentParser()
      parser.add_argument('dataset', type=str, default=None)
      parser.add_argument('--model', type=str, default=None)
      parser.add_argument('--doc-db', type=str, default=None,
                          help='Path to Document DB')
      parser.add_argument('--tokenizer', type=str, default='regexp')
      parser.add_argument('--n-docs', type=int, default=5)
      parser.add_argument('--num-workers', type=int, default=int(cpu_count()/2))
      parser.add_argument('--match', type=str, default='string',
                          choices=['regex', 'string'])
      args = parser.parse_args()
  
      # start time
      start = time.time()
  
      # read all the data and store it
      logger.info('Reading data ...')
      questions = []
      answers = []
      for line in open(args.dataset):
          data = json.loads(line)
          question = data['question']
          answer = data['answer']
          questions.append(question)
          answers.append(answer)
  
      # get the closest docs for each question.
      logger.info('Initializing ranker...')
      ranker = retriever.get_class('tfidf')(tfidf_path=args.model)
  
      logger.info('Ranking...')
      closest_docs = ranker.batch_closest_docs(
          questions, k=args.n_docs, num_workers=args.num_workers
      )
      answers_docs = zip(answers, closest_docs)
  
      # define processes
      tok_class = tokenizers.get_class(args.tokenizer)
      tok_opts = {}
      db_class = retriever.DocDB
      db_opts = {'db_path': args.doc_db}
      processes = ProcessPool(
          processes=args.num_workers,
          initializer=init,
          initargs=(tok_class, tok_opts, db_class, db_opts)
      )
  
      # compute the scores for each pair, and print the statistics
      logger.info('Retrieving and computing scores...')
      get_score_partial = partial(get_score, match=args.match)
      scores = processes.map(get_score_partial, answers_docs)
  
      filename = os.path.basename(args.dataset)
      stats = (
          "\n" + "-" * 50 + "\n" +
          "{filename}\n" +
          "Examples:\t\t\t{total}\n" +
          "Matches in top {k}:\t\t{m}\n" +
          "Match % in top {k}:\t\t{p:2.2f}\n" +
          "Total time:\t\t\t{t:2.4f} (s)\n"
      ).format(
          filename=filename,
          total=len(scores),
          k=args.n_docs,
          m=sum(scores),
          p=(sum(scores) / len(scores) * 100),
          t=time.time() - start,
      )
  
      print(stats)
  