(tentative) text preprocess for JSUT dataset

lsut_text_preprocess.py

#!/usr/bin/env python3

from collections import Counter
from os.path import join
import re
import os

import pykakasi

# --- Routines

SOURCE_DIR = '../../scratch/jsut_ver1.1/'
TARGET_DIR = '../../scratch/jsut_ver1.1_tacotron2/'

os.system('mkdir -p "%s"' % TARGET_DIR)

subs = 'loanword128 repeat500 voiceactress100 basic5000 onomatopee300 travel1000 countersuffix26 precedent130 utparaphrase512'.split()

digit2kanji = {str(i):c for i, c in enumerate('零一二三四五六七八九')}

def handle_phone_number(text):
    def _repl(g):
        s = g.group(0)
        if '-' in s or len(s) > 5:  # it's really a phone number
            # return '++' + s + '++'
            s = [c for c in s if c !='-']
            s = ' '.join([digit2kanji[c] for c in s])
            s = ' ' + s + ' '
            return s
        else:
            return s  # no change
    text = re.sub(r'[0-9\-]+', _repl, text)
    return text

def handle_number(text):
    def _repl(g):
        s = g.group(0)
        
        if len(s) == 1:
            res = digit2kanji[s[0]]
        else:
            if len(s) <= 5:
                s = list(reversed(s))
                sep = '十百千萬'
                res = []
                for i in range(len(s)-1, -1, -1):
                    if ((s[i] == '1' or s[i] == '0') and (i > 0) and (i<=3)) or (s[i] == '0' and i == 0):
                        pass
                    else:
                        res = res + [digit2kanji[s[i]]]
                    if i > 0 and s[i] != '0':
                        res = res + [sep[i-1]]
                res = ''.join(res)
            else:
                res = s
        
        return res
    
    text = re.sub(r'[0-9]+', _repl, text)
    return text

'''
for i in list(range(10)) + [123,1234,3456,12345,10203]:
    print(i, handle_number(str(i)))
'''

def kanaize(text):
    kakasi = pykakasi.kakasi()
    kakasi.setMode('J', 'H')
    kakasi.setMode("s", True)  # add space, default: no separator
        
    kakasi.setMode("H","a") # Hiragana to ascii, default: no conversion
    kakasi.setMode("K","a") # Katakana to ascii, default: no conversion
    kakasi.setMode("J","a") # Japanese to ascii, default: no conversion
    
    kakasi.setMode("r","Kunrei") # Use Kunrei, default: use Hepburn Roman table

    conv = kakasi.getConverter()
    
    # Pre  
    ## remove. Note: 'ｘ' is not 'x'
    text = re.sub(r'[『』「」Ｉ]', '', text)
    ## full-width to half-width
    def _full_to_half(s):
        return ''.join([chr(ord(c) - ord('！') + ord('!')) for c in s.group(0)])
    text = re.sub(r'[！-～]', _full_to_half, text)
    text = handle_phone_number(text)
    text = handle_number(text)
    text = re.sub(r'\%', 'パーセント', text) 
    
    # Main convert
    text = conv.do(text)  # kanji -> romaji
    
    # Post
    ## to lowercase
    text = text.lower()
    ## puncuation
    text = re.sub(r'、', ',', text)
    text = re.sub(r'。', '.', text)
    text = re.sub(r'々', '', text)  # rare failed case. ingore.
    text = re.sub(r'[x!]', '', text)  # rare symbols. ignore.
    ## multiple space
    text = re.sub(r'  +', ' ', text)
    
    
    return text

def load_transcript(filepath):
    def proc_line(l):
        tokens = l.strip().split(':')
        tokens[0] += '.wav'
        tokens.append(kanaize(tokens[1]))
        return tokens  # 0: wav file name. 1: transcript. 2: transcript in kana
        
    with open(filepath, encoding='utf8') as fin:
        content = [proc_line(_) for _ in fin.readlines()]
    return content

# --- Make transcript

transcript = []
for sub in subs:
    SOURCE_SUB_DIR = join(SOURCE_DIR, sub)
    sub_transcript = load_transcript(join(SOURCE_SUB_DIR, 'transcript_utf8.txt'))
    transcript.extend(sub_transcript)


# --- Save transcript

def save_transcript(transcript, indices, filepath):
    with open(filepath, 'w', encoding='utf8') as fout:
        for entry in transcript:
            for index in indices:
                item = entry[index]
                print(item, file=fout)
            print('', file=fout)
            
save_transcript(transcript, [1], join(TARGET_DIR, 'transcript.txt'))
save_transcript(transcript, [2], join(TARGET_DIR, 'transcript_proc.txt'))
save_transcript(transcript, [0, 1, 2], join(TARGET_DIR, 'transcript_full.txt'))

def save_transcript_as_filelist(transcript, filepath):
    with open(filepath, 'w', encoding='utf8') as fout:
        for entry in transcript:
            print(entry[0] + '|' + entry[2], file=fout)

save_transcript_as_filelist(transcript, join(TARGET_DIR, 'filelist.txt'))

# --- Analyze transcript

def analyze_transcript(transcript):
    char_set = set([char for entry in transcript for char in entry[2]])
    char_set = list(char_set)
    char_set.sort()
    print('char set')
    print(char_set)
    
    counter = Counter([char for entry in transcript for char in entry[2]])
    print('counter')
    print(counter.most_common())
    
analyze_transcript(transcript)