CML-TTS-Toolkit/execute_text_converter_beta.py at main · freds0/CML-TTS-Toolkit · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
import argparse
from spacy.lang.pt import Portuguese
from spacy.lang.pl import Polish
from tqdm import tqdm
from os.path import join
from utils.download_dataset import  download_language_dataset, download_books_dataset, extract_transcript_files, extract_book_files
from text_tools.text_normalization import customized_text_cleaning, portuguese_text_normalize, polish_text_normalize
from text_tools.custom_tokenizer import infix_re
from cleantext import clean
import collections
from text_tools.search_substring_with_threads import execute_threads_search_substring_by_char, execute_threads_search_substring_by_word

abbrev2language = {
    'pt': 'portuguese',
    'pl': 'polish',
    'it': 'italian',
    'sp': 'spanish',
    'fr': 'french',
    'du': 'dutch',
    'ge': 'german',
    'en': 'english'
}

def text_cleaning(text):
    text = clean(text,
                      fix_unicode=True,  # fix various unicode errors
                      to_ascii=False,  # transliterate to closest ASCII representation
                      lower=False,  # lowercase text_tools
                      no_line_breaks=True,  # fully strip line breaks as opposed to only normalizing them
                      no_urls=False,  # replace all URLs with a special token
                      no_emails=False,  # replace all email addresses with a special token
                      no_phone_numbers=False,  # replace all phone numbers with a special token
                      no_numbers=False,  # replace all numbers with a special token
                      no_digits=False,  # replace all digits with a special token
                      no_currency_symbols=False,  # replace all currency symbols with a special token
                      no_punct=False,  # remove punctuations
                      replace_with_punct="",  # instead of removing punctuations you may replace them
                      replace_with_url="<URL>",
                      replace_with_email="<EMAIL>",
                      replace_with_phone_number="<PHONE>",
                      replace_with_number="<NUMBER>",
                      replace_with_digit="0",
                      replace_with_currency_symbol="<CUR>",
                      lang="en"  # set to 'de' for German special handling
                      )
    text = customized_text_cleaning(text)
    return text

def get_text_normalization(language_abbrev = 'pt'):
    if language_abbrev == 'pl':
        norm = polish_text_normalize
    else:
        norm = portuguese_text_normalize
    return norm

def get_tokenizer(language_abbrev = 'pt'):
    if language_abbrev == 'pl':
        nlp = Polish()
    else:
        nlp = Portuguese()

    nlp.tokenizer.infix_finditer = infix_re.finditer
    nlp.max_length = 9990000  # or any large value, as long as you don't run out of RAM

    return nlp

def get_transcripts(transcripts_text):
    transcripts_dict = {}
    for line in transcripts_text:
        filename, text = line.split('\t')
        transcripts_dict[filename] = text.strip()
    # Sorting dict by key (filename)
    ordered_transcripts_dict = collections.OrderedDict(sorted(transcripts_dict.items()))
    return ordered_transcripts_dict


def execute(language_abbrev='pt', sequenced_text=False, similarity_metric='hamming', search_type='word', number_threads = 2):
    '''
    Execute convertion pipeline.
    '''

    print('Downloading {} dataset tar.gz file...'.format(language_abbrev))
    transcripts_tar_filename =  download_language_dataset(lang=language_abbrev)
    if not transcripts_tar_filename:
        return False

    print('Extracting files {}...'.format(transcripts_tar_filename))
    transcript_files_list = extract_transcript_files(transcripts_tar_filename)

    print('Downloading {} books tar.gz file...'.format(language_abbrev))
    books_tar_filename = download_books_dataset(lang=language_abbrev)

    print('Extracting files {}...'.format(books_tar_filename))
    books_folder = extract_book_files(books_tar_filename)

    language = abbrev2language[language_abbrev]

    separator = '|'

    total_similarity = 0.0
    book_id = ''

    # Iterates over [dev, test, train] files
    for transcript_file in transcript_files_list:

        output_filename = transcript_file.split('/')[1]
        output_filename = 'output_' + language + '_' + output_filename + '.csv'
        output_f = open(output_filename, "w")

        with open(transcript_file) as f:
            transcripts_text = f.readlines()

        start_position = 0

        # Create ordered dict from trascripts list
        transcripts_dict = get_transcripts(transcripts_text)
        # Iterates over each transcription
        for filename, text in tqdm(transcripts_dict.items()):
            print('Processing {}'.format(filename))

            new_book_id = filename.split('_')[1]
            # If it is a new book, updates book_text content
            if new_book_id != book_id:
                book_id = new_book_id

                with open(join(books_folder, language, book_id + '.txt')) as f:
                    book_text = f.read()

                # Cleaning complete text_tools
                book_text = text_cleaning(book_text)

            if search_type == 'char':
                text_result, similarity, start_position = execute_threads_search_substring_by_char(text, book_text,
                                                                                                   start_position=0,
                                                                                                   similarity_metric='hamming',
                                                                                                   total_threads=int(
                                                                                                       number_threads))
            else:
                text_result, similarity, start_position = execute_threads_search_substring_by_word(text, book_text,
                                                                                                   start_position=0,
                                                                                                   similarity_metric='hamming',
                                                                                                   total_threads=int(
                                                                                                       number_threads))
            # Debug
            print(text.strip())
            print(text_result.strip())
            print(similarity)

            total_similarity += similarity
            line = separator.join([filename.strip(), text.strip(), text_result.strip(), str(similarity) + '\n'])
            output_f.write(line)

        print('Mean Similarity: {}'.format(total_similarity / len(transcripts_text)))
        output_f.close()

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('-b', '--base_dir', default='./')
    parser.add_argument('-m', '--metric', default='hamming', help='Options: hamming (low accuracy, low computational cost), levenshtein (high accuracy, high computational cost) or ratcliff (average accuracy, average computational cost)')
    parser.add_argument('-l', '--language', default='pt', help='Options: pt (portuguese), pl (polish), it (italian), sp (spanish), fr (french), du (dutch), ge (german), en (english)')
    parser.add_argument('-n', '--number_threads', default=4)
    parser.add_argument('-t', '--search_type', default='word', help='Options: word or char')
    parser.add_argument('-s', '--sequenced_text', action='store_true', default=False)
    args = parser.parse_args()
    execute(args.language, args.sequenced_text, args.metric, args.search_type, args.number_threads)

if __name__ == "__main__":
    main()