日本語トークナイザーの作り方（トークナイザー後編）

# ドライブのマウント
from google.colab import drive
drive.mount('/content/drive')
# ベースとなるパスの指定
import os
BASE_PATH = "/content/drive/MyDrive/GENIAC/WORK/tokenizer"
%cd {BASE_PATH}

!pip install wikiextractor

import logging
import os
import bz2
import json
import shutil
import requests

# データセットダウンロード関数
def download_dataset(date: str, output_base: str = "output", lang: str = "ja") -> None:
    filename = f"{lang}wiki-{date}-pages-articles-multistream.xml.bz2"

    dump_path = os.path.join(output_base, f"tmp/wikipedia/{date}/{lang}")
    os.makedirs(dump_path, exist_ok=True)

    url = f"https://dumps.wikimedia.org/{lang}wiki/{date}/{filename}"
    if not os.path.exists(os.path.join(dump_path, filename)):
        logging.info(f"Downloading {url}")
        with requests.get(url, stream=True) as r:
            r.raise_for_status()
            logging.info(f"Saving to {os.path.join(dump_path, filename)}")
            with open(os.path.join(dump_path, filename), 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
    else:
        logging.info(f"File {os.path.join(dump_path, filename)} already exists")
        logging.info(f"Skipping download\n")

    # ダンプデータをパースする
    output_path = os.path.join(output_base, f"datasets/wikipedia/{date}/{lang}")
    if os.path.exists(output_path):
        shutil.rmtree(output_path)
    os.makedirs(output_path)

    logging.info(f"Parse and process {os.path.join(dump_path, filename)}")

# ダンプデータのダウンロード
input_date = "20240301" #最新のときは、"latest"
output_base = "dump"
lang = "ja"
status = download_dataset(input_date, output_base, lang)
# ＸＭＬファイルをテキスト形式に変換
import wikiextractor
file_name = lang + "wiki-" + input_date + "-pages-articles-multistream.xml.bz2"
dump_dir =  os.path.join(output_base, f"tmp/wikipedia/{input_date}/{lang}")
dump_path = os.path.join(dump_dir, file_name)
output_path = os.path.join(output_base, f"datasets")
!python -m wikiextractor.WikiExtractor --processes 8 -o {output_path} --no-templates {dump_path}

# ドライブのマウント
from google.colab import drive
drive.mount('/content/drive')
# ベースとなるパスの指定
import os
BASE_PATH = "/content/drive/MyDrive/GENIAC/WORK/tokenizer"
%cd {BASE_PATH}
# fugashiのインストール
!pip install fugashi[unidic]
!python -m unidic download
# 未知語の間引き処理
import os
import logging
import sys
logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    force=True)
def thin_out_unknown(input_path: str, output_base: str, testmode: bool) -> None:
  from fugashi import Tagger
  tagger = Tagger('-Owakati')
  input = 0
  output = 0
  flag = 0
  logging.info(f"testmode: {testmode}")
  if testmode:
    output_path = os.path.join(output_base, f"ja_wiki_thinout_test.txt")
  else:
    #output_path = os.path.join(output_base, f"ja_wiki_thinout.txt")
    output_path = os.path.join(output_base, f"ja_wiki_filter_thinout.txt")
  logging.info(f"output_path: {output_path}")
  with open(output_path, 'w', encoding='utf-8') as o:
    with open(input_path, 'r', encoding='utf-8') as f:
      for line in f:
        input += 1
        for word in tagger(line):
          if word.feature.lForm == None:
            flag += 1
        if flag == 0:
          o.write(line)
          output += 1
        flag = 0
        if input % 100000 == 0:
          logging.info(f"input: {input}, output: {output}, rate: {output / input}")
          if testmode:
            break
  logging.info(f"input: {input}, output: {output}, rate: {output / input}")
DATA_PATH = BASE_PATH + "/datasets"
text_path = DATA_PATH + "hogehoge.txt" # 使用するデータセット名に変更してください
thinout_path = os.path.join(DATA_PATH, f"thinout")
os.makedirs(thinout_path, exist_ok=True)
status = thin_out_unknown(text_path, thinout_path, testmode=False)

# ドライブのマウント
from google.colab import drive
drive.mount('/content/drive')
# ベースとなるパスの指定
import os
BASE_PATH = "/content/drive/MyDrive/GENIAC/WORK/tokenizer"
%cd {BASE_PATH}
# Mecabのインストール
!sudo apt install -y mecab libmecab-dev mecab-ipadic-utf8
# 分かち書き処理
!mecab -F"%M||||" -E"\n" input.txt -o output.txt