wehub-resource-sync
308 lines
11 KiB
Python
308 lines
11 KiB
Python
import argparse
|
|
import collections
|
|
import json
|
|
import os
|
|
import re
|
|
import string
|
|
import sys
|
|
from copy import deepcopy
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
|
|
class EvalOpts:
|
|
r"""
|
|
The options which the matrix evaluation process needs.
|
|
|
|
Arguments:
|
|
data_file (str): the SQuAD-style json file of the dataset in evaluation.
|
|
root_dir (str): the root directory of the raw WebSRC dataset, which contains the HTML files.
|
|
pred_file (str): the prediction file which contain the best predicted answer text of each question from the
|
|
model.
|
|
tag_pred_file (str): the prediction file which contain the best predicted answer tag id of each question from
|
|
the model.
|
|
result_file (str): the file to write down the matrix evaluation results of each question.
|
|
out_file (str): the file to write down the final matrix evaluation results of the whole dataset.
|
|
"""
|
|
def __init__(self, data_file, root_dir, pred_file, tag_pred_file, result_file='', out_file=""):
|
|
self.data_file = data_file
|
|
self.root_dir = root_dir
|
|
self.pred_file = pred_file
|
|
self.tag_pred_file = tag_pred_file
|
|
self.result_file = result_file
|
|
self.out_file = out_file
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('data_file', metavar='data.json', help='Input data JSON file.')
|
|
parser.add_argument('root_dir', metavar='./data', help='The root directory of the raw WebSRC dataset')
|
|
parser.add_argument('pred_file', metavar='pred.json', help='Model predictions.')
|
|
parser.add_argument('tag_pred_file', metavar='tag_pred.json', help='Model predictions.')
|
|
parser.add_argument('--result-file', '-r', metavar='qas_eval.json')
|
|
parser.add_argument('--out-file', '-o', metavar='eval.json',
|
|
help='Write accuracy metrics to file (default is stdout).')
|
|
if len(sys.argv) == 1:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
return parser.parse_args()
|
|
|
|
|
|
def make_pages_list(dataset):
|
|
r"""
|
|
Record all the pages which appears in the dataset and return the list.
|
|
"""
|
|
pages_list = []
|
|
last_page = None
|
|
for domain in dataset:
|
|
for w in domain['websites']:
|
|
for qa in w['qas']:
|
|
if last_page != qa['id'][:4]:
|
|
last_page = qa['id'][:4]
|
|
pages_list.append(last_page)
|
|
return pages_list
|
|
|
|
|
|
def make_qid_to_has_ans(dataset):
|
|
r"""
|
|
Pick all the questions which has answer in the dataset and return the list.
|
|
"""
|
|
qid_to_has_ans = {}
|
|
for domain in dataset:
|
|
for w in domain['websites']:
|
|
for qa in w['qas']:
|
|
qid_to_has_ans[qa['id']] = bool(qa['answers'])
|
|
return qid_to_has_ans
|
|
|
|
|
|
def normalize_answer(s):
|
|
"""Lower text and remove punctuation, articles and extra whitespace."""
|
|
|
|
def remove_articles(text):
|
|
regex = re.compile(r'\b(a|an|the)\b', re.UNICODE)
|
|
return re.sub(regex, ' ', text)
|
|
|
|
def white_space_fix(text):
|
|
return ' '.join(text.split())
|
|
|
|
def remove_punc(text):
|
|
exclude = set(string.punctuation)
|
|
return ''.join(ch for ch in text if ch not in exclude)
|
|
|
|
def lower(text):
|
|
return text.lower()
|
|
|
|
return white_space_fix(remove_articles(remove_punc(lower(s))))
|
|
|
|
|
|
def get_tokens(s):
|
|
r"""
|
|
Get the word list in the input.
|
|
"""
|
|
if not s:
|
|
return []
|
|
return normalize_answer(s).split()
|
|
|
|
|
|
def compute_exact(a_gold, a_pred):
|
|
r"""
|
|
Calculate the exact match.
|
|
"""
|
|
if normalize_answer(a_gold) == normalize_answer(a_pred):
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def compute_f1(a_gold, a_pred):
|
|
r"""
|
|
Calculate the f1 score.
|
|
"""
|
|
gold_toks = get_tokens(a_gold)
|
|
pred_toks = get_tokens(a_pred)
|
|
common = collections.Counter(gold_toks) & collections.Counter(pred_toks)
|
|
num_same = sum(common.values())
|
|
if len(gold_toks) == 0 or len(pred_toks) == 0:
|
|
# If either is no-answer, then F1 is 1 if they agree, 0 otherwise
|
|
return int(gold_toks == pred_toks)
|
|
if num_same == 0:
|
|
return 0
|
|
precision = 1.0 * num_same / len(pred_toks)
|
|
recall = 1.0 * num_same / len(gold_toks)
|
|
f1 = (2 * precision * recall) / (precision + recall)
|
|
return f1
|
|
|
|
|
|
def compute_pos(f, t_gold, addition, t_pred):
|
|
r"""
|
|
Calculate the POS score.
|
|
|
|
Arguments:
|
|
f (str): the html file on which the question is based.
|
|
t_gold (int): the gold answer tag id provided by the dataset (the value correspond to the key element_id).
|
|
addition (int): the addition information used for yes/no question provided by the dataset (the value
|
|
corresponding to the key answer_start).
|
|
t_pred (list[int]): the tag ids of the tags corresponding the each word in the predicted answer.
|
|
Returns:
|
|
float: the POS score.
|
|
"""
|
|
h = BeautifulSoup(open(f), "lxml")
|
|
p_gold, e_gold = set(), h.find(tid=t_gold)
|
|
if e_gold is None:
|
|
if len(t_pred) != 1:
|
|
return 0
|
|
else:
|
|
t = t_pred[0]
|
|
e_pred, e_prev = h.find(tid=t), h.find(tid=t-1)
|
|
if (e_pred is not None) or (addition == 1 and e_prev is not None) or\
|
|
(addition == 0 and e_prev is None):
|
|
return 0
|
|
else:
|
|
return 1
|
|
else:
|
|
p_gold.add(e_gold['tid'])
|
|
for e in e_gold.parents:
|
|
if int(e['tid']) < 2:
|
|
break
|
|
p_gold.add(e['tid'])
|
|
p = None
|
|
for t in t_pred:
|
|
p_pred, e_pred = set(), h.find(tid=t)
|
|
if e_pred is not None:
|
|
p_pred.add(e_pred['tid'])
|
|
if e_pred.name != 'html':
|
|
for e in e_pred.parents:
|
|
if int(e['tid']) < 2:
|
|
break
|
|
p_pred.add(e['tid'])
|
|
else:
|
|
p_pred.add(str(t))
|
|
if p is None:
|
|
p = p_pred
|
|
else:
|
|
p = p & p_pred # 预测值的公共祖先序列，except html&body
|
|
return len(p_gold & p) / len(p_gold | p)
|
|
|
|
|
|
def get_raw_scores(dataset, preds, tag_preds, root_dir):
|
|
r"""
|
|
Calculate all the three matrix (exact match, f1, POS) for each question.
|
|
|
|
Arguments:
|
|
dataset (dict): the dataset in use.
|
|
preds (dict): the answer text prediction for each question in the dataset.
|
|
tag_preds (dict): the answer tags prediction for each question in the dataset.
|
|
root_dir (str): the base directory for the html files.
|
|
|
|
Returns:
|
|
tuple(dict, dict, dict): exact match, f1, pos scores for each question.
|
|
"""
|
|
exact_scores = {}
|
|
f1_scores = {}
|
|
pos_scores = {}
|
|
for websites in dataset:
|
|
for w in websites['websites']:
|
|
f = os.path.join(root_dir, websites['domain'], w['page_id'][0:2], 'processed_data',
|
|
w['page_id'] + '.html')
|
|
for qa in w['qas']:
|
|
qid = qa['id']
|
|
gold_answers = [a['text'] for a in qa['answers']
|
|
if normalize_answer(a['text'])]
|
|
gold_tag_answers = [a['element_id'] for a in qa['answers']]
|
|
additional_tag_information = [a['answer_start'] for a in qa['answers']]
|
|
if not gold_answers:
|
|
# For unanswerable questions, only correct answer is empty string
|
|
gold_answers = ['']
|
|
if qid not in preds:
|
|
print('Missing prediction for %s' % qid)
|
|
continue
|
|
a_pred, t_pred = preds[qid], tag_preds[qid]
|
|
# Take max over all gold answers
|
|
exact_scores[qid] = max(compute_exact(a, a_pred) for a in gold_answers)
|
|
f1_scores[qid] = max(compute_f1(a, a_pred) for a in gold_answers)
|
|
pos_scores[qid] = max(compute_pos(f, t, a, t_pred)
|
|
for t, a in zip(gold_tag_answers, additional_tag_information))
|
|
return exact_scores, f1_scores, pos_scores
|
|
|
|
|
|
def make_eval_dict(exact_scores, f1_scores, pos_scores, qid_list=None):
|
|
r"""
|
|
Make the dictionary to show the evaluation results.
|
|
"""
|
|
if qid_list is None:
|
|
total = len(exact_scores)
|
|
return collections.OrderedDict([
|
|
('exact', 100.0 * sum(exact_scores.values()) / total),
|
|
('f1', 100.0 * sum(f1_scores.values()) / total),
|
|
('pos', 100.0 * sum(pos_scores.values()) / total),
|
|
('total', total),
|
|
])
|
|
else:
|
|
total = len(qid_list)
|
|
if total == 0:
|
|
return collections.OrderedDict([
|
|
('exact', 0),
|
|
('f1', 0),
|
|
('pos', 0),
|
|
('total', 0),
|
|
])
|
|
return collections.OrderedDict([
|
|
('exact', 100.0 * sum(exact_scores[k] for k in qid_list) / total),
|
|
('f1', 100.0 * sum(f1_scores[k] for k in qid_list) / total),
|
|
('pos', 100.0 * sum(pos_scores[k] for k in qid_list) / total),
|
|
('total', total),
|
|
])
|
|
|
|
|
|
def merge_eval(main_eval, new_eval, prefix):
|
|
for k in new_eval:
|
|
main_eval['%s_%s' % (prefix, k)] = new_eval[k]
|
|
|
|
|
|
def main(opts):
|
|
with open(opts.data_file) as f:
|
|
dataset_json = json.load(f)
|
|
dataset = dataset_json['data']
|
|
if isinstance(opts.pred_file, str):
|
|
with open(opts.pred_file) as f:
|
|
preds = json.load(f)
|
|
else:
|
|
preds = opts.pred_file
|
|
if isinstance(opts.tag_pred_file, str):
|
|
with open(opts.tag_pred_file) as f:
|
|
tag_preds = json.load(f)
|
|
else:
|
|
tag_preds = opts.tag_pred_file
|
|
qid_to_has_ans = make_qid_to_has_ans(dataset)
|
|
has_ans_qids = [k for k, v in qid_to_has_ans.items() if v]
|
|
no_ans_qids = [k for k, v in qid_to_has_ans.items() if not v]
|
|
exact, f1, pos = get_raw_scores(dataset, preds, tag_preds, opts.root_dir)
|
|
out_eval = make_eval_dict(exact, f1, pos)
|
|
if has_ans_qids:
|
|
has_ans_eval = make_eval_dict(exact, f1, pos, qid_list=has_ans_qids)
|
|
merge_eval(out_eval, has_ans_eval, 'HasAns')
|
|
if no_ans_qids:
|
|
no_ans_eval = make_eval_dict(exact, f1, pos, qid_list=no_ans_qids)
|
|
merge_eval(out_eval, no_ans_eval, 'NoAns')
|
|
print(json.dumps(out_eval, indent=2))
|
|
pages_list, write_eval = make_pages_list(dataset), deepcopy(out_eval)
|
|
for p in pages_list:
|
|
pages_ans_qids = [k for k, _ in qid_to_has_ans.items() if p in k]
|
|
page_eval = make_eval_dict(exact, f1, pos, qid_list=pages_ans_qids)
|
|
merge_eval(write_eval, page_eval, p)
|
|
if opts.result_file:
|
|
with open(opts.result_file, 'w') as f:
|
|
w = {}
|
|
for k, v in qid_to_has_ans.items():
|
|
w[k] = {'exact': exact[k], 'f1': f1[k], 'pos': pos[k]}
|
|
json.dump(w, f)
|
|
if opts.out_file:
|
|
with open(opts.out_file, 'w') as f:
|
|
json.dump(write_eval, f)
|
|
return out_eval
|
|
|
|
|
|
if __name__ == '__main__':
|
|
a="$4.99"
|
|
b="$4.99"
|
|
print(compute_exact(a,b))
|