모두의 말뭉치: 구어 말뭉치 loader

Korpora/korpus_modu_spoken.py

@@ -0,0 +1,84 @@
+import json
+import os
+import re
+from dataclasses import dataclass
+from glob import glob
+from tqdm import tqdm
+from typing import List
+from Korpora.korpora import Korpus, KorpusData
+
+
+description = """    모두의 말뭉치는 문화체육관광부 산하 국립국어원에서 제공하는 말뭉치로
+    총 13 개의 말뭉치로 이뤄져 있습니다.
+    해당 말뭉치를 이용하기 위해서는 국립국어원 홈페이지에 가셔서 "회원가입 > 말뭉치 신청 > 승인"의
+    과정을 거치셔야 합니다.
+    https://corpus.korean.go.kr/#none
+    모두의 말뭉치는 승인 후 다운로드 가능 기간 및 횟수 (3회) 에 제한이 있습니다.
+    로그인 기능 및 Korpora 패키지에서의 다운로드 기능을 제공하려 하였지만,
+    국립국어원에서 위의 이유로 이에 대한 기능은 제공이 불가함을 확인하였습니다.
+    Korpora==0.2.0 에서는 "개별 말뭉치 신청 > 승인"이 완료되었다고 가정,
+    로컬에 다운로드 된 말뭉치를 손쉽게 로딩하는 기능만 제공할 예정입니다
+    (Korpora 개발진 lovit@github, ratsgo@github)"""
+
+license = """    모두의 말뭉치의 모든 저작권은 `문화체육관광부 국립국어원
+    (National Institute of Korean Language)` 에 귀속됩니다.
+    정확한 라이센스는 확인 중 입니다."""
+
+
+class ModuSpokenKorpus(Korpus):
+    def __init__(self, root_dir_or_paths, force_download=False):
+        super().__init__(description, license)
+        paths = find_corpus_paths(root_dir_or_paths)
+        self.train = KorpusData('모두의_구어_말뭉치.train', load_modu_spoken(paths))
+
+
+def find_corpus_paths(root_dir_or_paths):
+    prefix_pattern = re.compile('S[ABDE]RW')
+    def match(path):
+        prefix = path.split(os.path.sep)[-1][:4]
+        return prefix_pattern.match(prefix)
+
+    # directory + wildcard
+    if isinstance(root_dir_or_paths, str):
+        paths = sorted(glob(f'{root_dir_or_paths}/*.json') + glob(root_dir_or_paths))
+    else:
+        paths = root_dir_or_paths
+
+    paths = [path for path in paths if match(path)]
+    if not paths:
+        raise ValueError('Not found corpus files. Check `root_dir_or_paths`')
+    return paths
+
+
+def load_modu_spoken(paths):
+    texts = []
+    for i_path, path in enumerate(tqdm(paths, desc='Loading Spoken', total=len(paths))):
+        with open(path, encoding='utf-8') as f:
+            data = json.load(f)
+        documents = data['document']
+        texts += [paragraph for document in documents for paragraph in document_to_texts(document)]
+    return texts
+
+
+def document_to_texts(document):
+    utterance = document['utterance']
+    texts = []
+    buffer = []
+    speaker_id = -1
+    for u in utterance:
+        form = u["original_form"]
+        speaker_id_u = u["speaker_id"]
+        if (speaker_id != speaker_id_u) and buffer:
+            texts.append(' '.join(buffer))
+            buffer = []
+        buffer.append(form)
+        speaker_id = speaker_id_u
+    texts.append(' '.join(buffer))
+    return texts
+
+
+def fetch_modu():
+    raise NotImplementedError(
+        "국립국어원에서 API 기능을 제공해 줄 수 없음을 확인하였습니다."
+        "\n이에 따라 모두의 말뭉치는 fetch 기능을 제공하지 않습니다"
+    )

README.md

@@ -557,3 +557,22 @@ corpus.train[0].speaker_id[:10]
 | original_form | 대화 원본 텍스트 |
 | speaker_id | 발화자 (숫자가 아님) |
 | time | `yyyymmdd hh:mm` 형식 |
+
+
+### 모두의 말뭉치: 구어 말뭉치 (loader)
+- author: 국립국어원
+- repository: https://corpus.korean.go.kr/
+- example
+  - 구어 말뭉치 내 한 화자의 연속된 말을 하나의 paragraph 로 정리, 이를 train 의 데이터로 로드
+```python
+from Korpora.korpus_modu_spoken import ModuSpokenKorpus
+
+paths_or_dir = 'path/to/NIKL_SPOKEN(v1.0)/'
+paths_or_dir = 'path/to/NIKL_SPOKEN(v1.0)/SARW1800000001*'
+corpus = ModuSpokenKorpus(paths_or_dir)
+
+corpus.train[0]
+# '요즘처럼 추운 날씨에는 따뜻한 라테 한잔 찾는 분들 많으실 텐데요. 라테 위에 그려진 다양한 라테 아트를 구경하는 것도 ...
+type(corpus.train[0])
+# str
+```