ナード戦隊データマン

機械学習と自然言語処理についてのブログ

webvtt形式のパラレルコーパスのalignment

webvtt1は字幕データの一般的なファイル形式です。

概要

youtube-dl2を使って字幕データのパラレルコーパスを生成した場合、そのままの状態では1対1の文のペアにはなっていません。そこで、なんらかの方法でalignmentします。

youtube-dlでは、以下のコマンドで字幕を収集できます。

youtube-dl -ciw --all-subs --skip-download YoutubeチャンネルURL

コード

align.py

# coding: utf-8
import webvtt
import re
import MeCab
 
re_spc = re.compile(r"[ ]+")
tagger = MeCab.Tagger("-Owakati")
 
 
def tokenizer(text, tagger=tagger):
    return tagger.parse(text).split()
 
 
def time_second(time_str):
    time_data = list(map(float, time_str.split(":")))
    total = 0
    for i, t in enumerate(time_data):
        total += t * (60**(2 - i))
    return total
 
 
def length_score(text1, text2, tokenizer1, tokenizer2):
    s1 = len(tokenizer1(text1))
    s2 = len(tokenizer2(text2))
    diff_word = abs(s1 / (s1 + s2 + 1) - s2 / (s1 + s2 + 1))
    return 1 / (diff_word + 1)
 
 
def cross_inner(last_j, i, d1, data2, data2_arrived, threshold, tokenizer1, tokenizer2):
    d1_s = time_second(d1.start)
    d1_e = time_second(d1.end)
    try:
        data2[last_j]
    except IndexError:
        return None, None, False
    data = data2[last_j:]
    for j, d2 in enumerate(data):
        if last_j + j in data2_arrived:
            continue
        d2_s = time_second(d2.start)
        d2_e = time_second(d2.end)
        if (d1_s <= d2_s and d2_s <= d1_e) \
           or (d2_e <= d1_e and d1_s <= d2_e):
            data2_arrived[last_j+j] = True
            text1 = re.sub(re_spc, ' ', d1.text.replace("\n", " "))
            text2 = re.sub(re_spc, ' ', d2.text.replace("\n", ""))
            last_j = last_j + j + 1
            if length_score(text1, text2, tokenizer1,
                            tokenizer2) > threshold:
                return [text1, text2], last_j, True
            else:
                return None, last_j, False
        elif d2_s > d1_e:
            last_j = last_j + j + 1
            return None, last_j, False
    try:
        data2[last_j+1]
        return None, last_j+1, False
    except IndexError:
        return None, None, False
 
    
def cross(data1,
          data2,
          threshold=0.65,
          tokenizer1=tokenizer,
          tokenizer2=tokenizer):
    out = []
    data2_arrived = {}
    last_j = 0
    for i, d1 in enumerate(data1):
        row, last_j, flag = cross_inner(
            last_j, i, d1, data2, data2_arrived, threshold,
            tokenizer1, tokenizer2)
        if last_j is None:
            break
        if flag:
            out.append(row)
    return out

align_example.py

# coding: utf-8
 
import os
import align
import webvtt
from tqdm import tqdm
 

def run(files_en, files_ja):
    out = []
    for file_en, file_ja in tqdm(zip(files_en, files_ja)):
        try:
            data1 = [x for x in webvtt.read(file_en)]
            data2 = [x for x in webvtt.read(file_ja)]
            out += align.cross(data1, data2)
        except FileNotFoundError:
            print(file_en)
    with open("en-ja_all.txt", "w") as f:
        f.write('\n'.join(['\t'.join(x) for x in out]))
    return out
 
if __name__=="__main__":
    files = list([x for x in os.listdir("data_ted") if x.endswith("vtt")])
    titles_ja = {
        '.'.join(x.split(".")[:-2])
        for x in files if x.split(".")[-2] == "ja"
    }
    titles = sorted(list(titles_ja))
    files_en = [os.path.join("data_ted", x + ".en.vtt") for x in titles]
    files_ja = [os.path.join("data_ted", x + ".ja.vtt") for x in titles]
 
    run(files_en, files_ja)

アルゴリズムの説明

言語Aのある字幕Xと言語Bのある字幕Yがあったとき、この2つの字幕の時間が少しでも重なった場合、ペアの候補にします。ペアの候補は、length-based手法によってさらに絞り込まれます。

抽出結果の一部

Taking it a step further,    もう少し踏み込むと
MIT's media lab is working on robots    MITのメディアラボはもっと
that can interact more like humans. 人間のようにやり取りできるロボットを作っていますが
But what if those robots were able to interact  これらが特定の人物の
based on the unique characteristics of a specific person    独特の性格に基づいて対話することができたら?
based on the hundreds of thousands of pieces of content その人物が生前作った
that person produces in their lifetime? 何十万ものコンテンツをもとすることができたら?
Finally, think back to this famous scene    最後に 米国の2008年選挙の夜の
from election night 2008    この有名なシーンを
back in the United States,  思い出してください
where CNN beamed a live hologram    アンダーソン・クーパーとのインタビューのため
of hip hop artist will.i.am into their studio   CNNがヒップホップアーティストの
for an interview with Anderson Cooper.  will.i.amをホログラムでライブ中継しました
What if we were able to use that same type of technology    同じタイプの技術を使って
to beam a representation of our loved ones into our living rooms -- 愛する人たちの映像を自分の居間に映し それが
interacting in a very lifelike way  生前に作ったコンテンツをもとに
based on all the content they created while they were alive?    まるで生きているかのように話せたらどうでしょう?
I think that's going to become completely possible  私たちが使っているデータの量と
as the amount of data we're producing   それを理解する技術が両方
and technology's ability to understand it   飛躍的に拡大するにつれて
both expand exponentially.  これは全く可能となると思います
Now in closing, I think what we all need to be thinking about   最後に これが現実になって欲しいのか
is if we want that to become our reality -- もしそうなら
and if so,  これが人生やその後すべてに対し
what it means for a definition of life and everything that comes after it.  どんな意味があるのか考える必要があると思います
Thank you very much.    ありがとうございました

考察

ルールベースの場合、alignmentするために訓練のための追加のパラレルコーパスが必要になることがありません。ただし、精度があまり良くないかもしれません。TEDの場合は、全体的にほとんど1対1に近いかたちになっているので問題が起きにくいですが、もっと雑な翻訳者の作った字幕の場合、前述の手法では対処が難しいです。

例えば、字幕の表示時間が言語ごとに異なっていたりした場合は、一致する時間を基準に探すアルゴリズムが無駄になる可能性があります。

TEDの翻訳者は、各々の言語において自然な翻訳がなされるようにしているようなので、SVO形式の文法と、SOV形式の文法のペアでは、表示が前後する可能性があります。この解決策の一つは、TEDの英文は、英語ではコロンとピリオドがちゃんと使われていて、文の最初は大文字になっているので、この事実を用いて「文」の単位に修正することです。この場合、文全体では意味的に通るようになるので、語順の前後の問題が回避しやすくなります。

追記

alignmentを文ごとに生成するように修正しました。

[align.py]

# coding: utf-8
import webvtt
import re
import MeCab
 
re_spc = re.compile(r"[ ]+")
 
 
def time_second(time_str):
    time_data = list(map(float, time_str.split(":")))
    total = 0
    for i, t in enumerate(time_data):
        total += t * (60**(2 - i))
    return total
 
 
def cross_inner(last_j, i, d1, data2, data2_arrived):
    d1_s = time_second(d1.start)
    d1_e = time_second(d1.end)
    try:
        data2[last_j]
    except IndexError:
        return None, None, False
    data = data2[last_j:]
    for j, d2 in enumerate(data):
        if last_j + j in data2_arrived:
            continue
        d2_s = time_second(d2.start)
        d2_e = time_second(d2.end)
        if (d1_s <= d2_s and d2_s <= d1_e) \
           or (d2_e <= d1_e and d1_s <= d2_e):
            data2_arrived[last_j+j] = True
            text1 = re.sub(re_spc, ' ', d1.text.replace("\n", " "))
            text2 = re.sub(re_spc, ' ', d2.text.replace("\n", ""))
            last_j = last_j + j + 1
            return [text1, text2], last_j, True
        elif d2_s > d1_e:
            last_j = last_j + j
            return None, last_j, False
    try:
        data2[last_j+1]
        return None, last_j+1, False
    except IndexError:
        return None, None, False
 
    
def cross(data1, data2):
    out = []
    data2_arrived = {}
    last_j = 0
    for i, d1 in enumerate(data1):
        row, last_j, flag = cross_inner(
            last_j, i, d1, data2, data2_arrived)
        if last_j is None:
            break
        if flag:
            out.append(row)
    return out

[align2]

from tqdm import tqdm
 
ALPHA = "abcdefghijklmnopqrstuvwxyz"
 
ALPHA_DICT = {c: True for c in list(ALPHA) + list(ALPHA.upper())}
 
 
def merge_sents(data_file):
    out = []
    with open(data_file) as f:
        current_en = []
        current_ja = []
        for line in tqdm(f):
            line = line.strip().split("\t")
            c = None
            i = 0
            while (c not in ALPHA_DICT and i<len(line[0])):
                c = line[0][i]
                i += 1
            if i >= len(line[0]):
                continue
            if c.isupper():
                entxt = ' '.join(current_en)
                jatxt = ''.join(current_ja)
                out.append(entxt + "\t" + jatxt)
                current_en = []
                current_ja = []
            current_en.append(line[0])
            current_ja.append(line[1])
        if current_en and current_ja:
            entxt = ' '.join(current_en)
            jatxt = ''.join(current_ja)
            out.append(entxt + "\t" + jatxt)
    return out
 
 
if __name__ == "__main__":
    out = merge_sents("./data/TED/en-ja_all.txt")
    with open("./en-ja_fixed.txt", "w") as f:
        f.write('\n'.join(out))

参考