features/graph_utils.py

import networkx as nx
import numpy as np
from cdlib import algorithms


# these functions are heavily influenced by the HF squad_metrics.py script
def normalize_text(s):
    """Removing articles and punctuation, and standardizing whitespace are all typical text processing steps."""
    import string, re

    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 compute_exact_match(prediction, truth):
    return int(normalize_text(prediction) == normalize_text(truth))


def compute_f1(prediction, truth):
    pred_tokens = normalize_text(prediction).split()
    truth_tokens = normalize_text(truth).split()

    # if either the prediction or the truth is no-answer then f1 = 1 if they agree, 0 otherwise
    if len(pred_tokens) == 0 or len(truth_tokens) == 0:
        return int(pred_tokens == truth_tokens)

    common_tokens = set(pred_tokens) & set(truth_tokens)

    # if there are no common tokens then f1 = 0
    if len(common_tokens) == 0:
        return 0

    prec = len(common_tokens) / len(pred_tokens)
    rec = len(common_tokens) / len(truth_tokens)

    return 2 * (prec * rec) / (prec + rec)


def is_date_or_num(answer):
    answer = answer.lower().split()
    for w in answer:
        w = w.strip()
        if w.isnumeric() or w in ["ngày", "tháng", "năm"]:
            return True
    return False


def find_best_cluster(answers, best_answer, thr=0.79):
    if len(answers) == 0:  # or best_answer not in answers:
        return best_answer
    elif len(answers) == 1:
        return answers[0]
    dists = np.zeros((len(answers), len(answers)))
    for i in range(len(answers) - 1):
        for j in range(i + 1, len(answers)):
            a1 = answers[i].lower().strip()
            a2 = answers[j].lower().strip()
            if is_date_or_num(a1) or is_date_or_num(a2):
                # print(a1, a2)
                if a1 == a2 or ("tháng" in a1 and a1 in a2) or ("tháng" in a2 and a2 in a1):
                    dists[i, j] = 1
                    dists[j, i] = 1
                # continue
            elif a1 == a2 or (a1 in a2) or (a2 in a1) or compute_f1(a1.lower(), a2.lower()) >= thr:
                dists[i, j] = 1
                dists[j, i] = 1
    # print(dists)
    try:
        thr = 1
        dups = np.where(dists >= thr)
        dup_strs = []
        edges = []
        for i, j in zip(dups[0], dups[1]):
            if i != j:
                edges.append((i, j))
        G = nx.Graph()
        for i, answer in enumerate(answers):
            G.add_node(i, content=answer)
        G.add_edges_from(edges)
        partition = algorithms.louvain(G)
        max_len_comm = np.max([len(x) for x in partition.communities])
        best_comms = []
        for comm in partition.communities:
            # print([answers[i] for i in comm])
            if len(comm) == max_len_comm:
                best_comms.append([answers[i] for i in comm])
        # if len(best_comms) > 1:
        #     return best_answer
        for comm in best_comms:
            if best_answer in comm:
                return best_answer
        mid = len(best_comms[0]) // 2
        # print(mid, sorted(best_comms[0], key = len))
        return sorted(best_comms[0], key=len)[mid]
    except Exception as e:
        print(e, "Disconnected graph")
        return best_answer