utils.py

import difflib
import regtag
import random


def merge_span(words, tags):
    spans, span_tags = [], []
    current_tag = 'O'
    span = []
    for w, t in zip(words, tags):
        w = w.strip(":-")
        if len(w) == 0:
            continue
        t_info = t.split('-')
        if t_info[-1] != current_tag or t_info[0] == 'B':
            if len(span) > 0:
                spans.append(' '.join(span))
                span_tags.append(current_tag)
            span = [w]
            current_tag = t_info[-1]
        else:
            span.append(w)
    if len(span) > 0:
        spans.append(' '.join(span))
        span_tags.append(current_tag)
    return spans, span_tags


def make_spoken(text, do_split=True):
    src, tgt = [], []
    if do_split:
        chunk_size = random.choice(list(range(0, 10)) + list(range(10, 35)) * 4)
        if chunk_size > 0:
            text = random.choice(split_chunk_input(text, chunk_size))
        else:
            text = ''
    words, word_tags = merge_span(*regtag.tagging(text))
    for span, t in zip(words, word_tags):
        if t == 'O':
            for w in span.split():
                w = w.strip('/.,?!').lower()
                if len(w) > 0:
                    src.append(w)
                    tgt.append(w)
                if random.random() < 0.01:
                    random_value = regtag.augment.get_random_span()
                    tgt.append(random_value[0])
                    src.append(random_value[1].lower())
        else:
            random_value = regtag.augment.get_random_span(t, span.lower())
            tgt.append(random_value[0])
            src.append(random_value[1].lower())

    if len(src) == 0:
        tgt, src = regtag.get_random_span()
        src = [src]
        tgt = [tgt]

    return src, tgt


def split_chunk_input(raw_text, chunk_size):
    input_words = raw_text.strip().split()
    clean_data = [input_words[i:i + chunk_size] for i in range(0, len(input_words), chunk_size)]
    if len(clean_data) > 1:
        clean_data = [" ".join(clean_data[i] + clean_data[i + 1]) for i in range(len(clean_data) - 1)]
    else:
        clean_data = [" ".join(clean_data[0])]
    return clean_data


def split_chunk_input(raw_text, chunk_size=40, overlap=10):
    input_words = raw_text.strip().split()
    part_per_chunk = chunk_size // overlap
    clean_data = [input_words[i:i + overlap] for i in range(0, len(input_words), overlap)]
    if len(clean_data) > 1:
        merge_data = []
        for i in range(0, len(clean_data) - 1, part_per_chunk - 1):
            merge_data.append(' '.join([y for x in clean_data[i:i + part_per_chunk] for y in x]))
    else:
        merge_data = [" ".join(clean_data[0])]
    return merge_data


def merge_two_chunk(chunk_1, chunk_2, overlap, debug=False):
    def extract_phrase_word(phrase):
        if phrase.startswith('<mask>'):
            return phrase[7:].split('](')[1][:-1].split()
        else:
            return [phrase]

    def has_tag(phrase):
        if phrase.startswith('<') and phrase.endswith(')'):
            return True
        return False

    def extract_compete_region(list_phrases, is_head):
        if is_head:
            list_phrases = list_phrases[::-1]
        compete = []
        remain = []
        handle_count = 0
        for phrase in list_phrases:
            phrase_word = extract_phrase_word(phrase)
            if len(phrase_word) + handle_count <= overlap:
                compete.append(phrase)
                handle_count += len(phrase_word)
            else:
                if handle_count < overlap:
                    remain_compete_count = overlap - handle_count
                    remain.append(phrase)
                    if not is_head:
                        compete.extend(["<delete>({})".format(item) for item in phrase_word[:remain_compete_count]])
                    else:
                        compete.extend(
                            ["<delete>({})".format(item) for item in phrase_word[::-1][:remain_compete_count]])
                    handle_count = overlap
                else:
                    remain.append(phrase)
        if is_head:
            compete = compete[::-1]
            remain = remain[::-1]
        return remain, compete

    def is_equal(phrase_1, phrase_2):
        if phrase_1 == phrase_2:
            return True
        if extract_phrase_word(phrase_1) == extract_phrase_word(phrase_2):
            if phrase_1.startswith('<mask>') and phrase_2.startswith('<mask>'):
                return True
        return False

    def merge_compete(list_1, list_2):
        idx_list_1, idx_list_2, combine_phrases = [], [], []
        mark_term_complete = []
        list_raw = [extract_phrase_word(item) for item in list_1]
        list_raw = [y for x in list_raw for y in x]
        for idx, phrase in enumerate(list_1):
            idx_list_1.extend([idx] * len(extract_phrase_word(phrase)))
        for idx, phrase in enumerate(list_2):
            idx_list_2.extend([idx] * len(extract_phrase_word(phrase)))
        # print(idx_list_1, idx_list_2)
        for idx, (idx_1, idx_2) in enumerate(zip(idx_list_1, idx_list_2)):
            if list_1[idx_1].startswith('<delete>') or list_2[idx_2].startswith('<delete>'):
                continue
            elif is_equal(list_1[idx_1], list_2[idx_2]):
                # print(list_1[idx_1])
                if '1_{}'.format(idx_1) not in mark_term_complete and '2_{}'.format(idx_2) not in mark_term_complete:
                    if idx <= overlap//2:
                        combine_phrases.append(list_1[idx_1])
                        mark_term_complete.append('1_{}'.format(idx_1))
                    else:
                        combine_phrases.append(list_2[idx_2])
                        mark_term_complete.append('2_{}'.format(idx_2))
            else:
                combine_phrases.append(list_raw[idx])
                mark_term_complete.extend(['1_{}'.format(idx_1), '2_{}'.format(idx_2)])
        # print(mark_term_complete)
        return combine_phrases

    remain_1, compete_1 = extract_compete_region(chunk_1, is_head=True)
    remain_2, compete_2 = extract_compete_region(chunk_2[1:-1], is_head=False)
    compromise = merge_compete(compete_1, compete_2)

    if debug:
        print(remain_1, '\n', compete_1)
        print('-----------------------')
        print(compete_2, '\n', remain_2)
        print('-----------------------')
        print(compromise, '\n\n')

    return remain_1 + compromise + remain_2


def merge_chunk_pre_norm(list_chunks, overlap, debug=False):
    if len(list_chunks) == 0:
        return []
    if len(list_chunks) == 1:
        return list_chunks[0][1:-1]
    current_chunk = list_chunks[0][1:-1]
    for tmp_chunk in list_chunks[1:]:
        current_chunk = merge_two_chunk(current_chunk, tmp_chunk, overlap, debug=debug)
    return current_chunk


def equalize(s1, s2):
    l1 = s1.split()
    l2 = s2.split()
    res1 = []
    res2 = []
    combine = []
    prev = difflib.Match(0, 0, 0)
    for match in difflib.SequenceMatcher(a=l1, b=l2).get_matching_blocks():
        if prev.a + prev.size != match.a:
            for i in range(prev.a + prev.size, match.a):
                res2 += ['_' * len(l1[i])]
            res1 += l1[prev.a + prev.size:match.a]

            for i in l1[prev.a + prev.size:match.a]:
                if len(combine) < len(l1) // 2:
                    print(l1[prev.a + prev.size:match.a])
                    combine.append(i)
        if prev.b + prev.size != match.b:
            for i in range(prev.b + prev.size, match.b):
                res1 += ['_' * len(l2[i])]
            res2 += l2[prev.b + prev.size:match.b]

            for i in l2[prev.b + prev.size:match.b]:
                if len(combine) >= len(l2) // 2:
                    print(l2[prev.b + prev.size:match.b])
                    combine.append(i)
        res1 += l1[match.a:match.a + match.size]
        res2 += l2[match.b:match.b + match.size]
        combine += l2[match.b:match.b + match.size]
        prev = match
    return ' '.join(res1), ' '.join(res2), combine


def count_overlap(words_1, words_2):
    # print(words_1, words_2)
    assert len(words_1) == len(words_2)
    len_overlap = 0
    for match in difflib.SequenceMatcher(a=words_1, b=words_2).get_matching_blocks():
        len_overlap += match.size

    # for w1, w2 in zip(words_1, words_2):
    #     if w1 == w2:
    #         len_overlap += 1
    return len_overlap


def find_overlap_chunk(txt_1, txt_2):
    # print(txt_1)
    # print(txt_2)
    window_view = 1
    idx_1 = len(txt_1) - window_view
    idx_2 = window_view
    over_lap = 0
    current_best_idx_1 = len(txt_1)
    current_best_idx_2 = 0

    while window_view <= len(txt_1) and window_view <= len(txt_2):
        current_overlap = count_overlap(txt_1[idx_1:], txt_2[:idx_2])
        print(current_overlap)
        if over_lap < current_overlap:
            over_lap = current_overlap
            current_best_idx_1 = idx_1
            current_best_idx_2 = idx_2
        window_view += 1
        idx_1 = len(txt_1) - window_view
        idx_2 = window_view
        # else:
        #     break
    print('----->', txt_1[current_best_idx_1:], txt_2[:current_best_idx_2])
    return txt_1[current_best_idx_1:], txt_2[:current_best_idx_2]


def concat_chunks(list_chunks):
    concat_string = list_chunks[0].split()
    for i in range(1, len(list_chunks)):
        remain_string = list_chunks[i].split()
        s1, s2 = find_overlap_chunk(concat_string, remain_string)
        s1 = ' '.join(s1)
        s2 = ' '.join(s2)
        _, _, overlap_merged = equalize(s1, s2)
        merge_len = len(s1.split())

        concat_string = concat_string[:len(concat_string) - merge_len] + overlap_merged + remain_string[merge_len:]

    concat_string = ' '.join(concat_string)
    return concat_string