Pythonを使って日本語の要約データを取得する

　
自然言語処理のタスクは、Transformer が現れて以来一段と盛り上がっています。これまで精度がいまいちだったタスクで、人間以上の精度になってきています。それは、文章の要約タスクでも同様です。文章要約は、与えれた文章の中で重要なワードや文を抜き取り、場合によっては言い換え、意味がわかる自然な文章を作るタスクです。ただ文章を短くすればよいというわけではなく、意味がわかるようにまとめなければいけないというところが、他の自然言語処理のタスクと比べて難しい所以です。

精度の高い要約モデルを作るためには、莫大なデータで事前学習済みのモデルを fine-tuning します。しかし、日本語の文章と要約が対になっている要約データは、簡単に用意できません。そこで、今回は日本語の要約データを取得するスクリプトに関して記載しました。

今回のコードはこちらに置いています。

準備

今回取得する要約データは、livedoor ニュースの3行要約データです。下記の GitHub では、 livedoor ニュースのニュース ID データがあります。まずはこのデータを取得します。

github.com

必要なモジュールの Import とディレクトリの作成

スクレイピングをして要約データを取得するため、「requests」と「bs4」をインポートしています。また、取得したデータを保存するディレクトリ「data」を作っています。

from pathlib import Path
import re
import time
import requests
import pandas as pd
import urllib
from bs4 import BeautifulSoup
from tqdm import tqdm

data_dir_path = Path('data')
if not data_dir_path.exists():
    data_dir_path.mkdir(parents=True)

ニュースIDデータの取得

このディレクトリにあるニュース ID データの「train.csv」と「test.csv」をダウンロードします。

def download_data(url, data_dir_path):
    
    file_path = data_dir_path.joinpath(Path(url).name)

    data = requests.get(url).content
    with open(file_path, 'wb') as file:
        file.write(data)

url = 'https://raw.githubusercontent.com/KodairaTomonori/ThreeLineSummaryDataset/master/data/train.csv'
download_data(url=url, data_dir_path=data_dir_path)

url = 'https://raw.githubusercontent.com/KodairaTomonori/ThreeLineSummaryDataset/master/data/test.csv'
download_data(url=url, data_dir_path=data_dir_path)

ニュースIDデータの加工

取得したニュース ID データ２つをロードし、連結させます。その後、ニュース ID の不要な文字を消し、すでにニュース本文を取得しているニュース ID があれば、「anti_join関数」を使いそれを省くようにしています。

def anti_join(data1, data2, by):

    joined_data = data1.copy()
    target_data = data2.copy()
    target_data['flag_tmp'] = 1

    if type(by) is str:
        by = [by]

    joined_data = pd.merge(
        joined_data, target_data[by + ['flag_tmp']].drop_duplicates(),
        on=by, how='left'
    ).query('flag_tmp.isnull()', engine='python').drop(columns='flag_tmp').copy()

    return joined_data

columns = ['year', 'month', 'category', 'article_id', 'type_label']

articles = pd.DataFrame()
for data_name in ['train.csv', 'test.csv']:
    data = pd.read_csv(data_dir_path.joinpath(data_name))
    tmp = data.columns.tolist()
    if data_name == 'train.csv':
        data.columns = columns[:-1]
        data = pd.concat([
            data, pd.DataFrame([tmp], columns=columns[:-1])
        ], axis=0)
        data['type_label'] = None
    else:
        data.columns = columns
        data = pd.concat([
            data, pd.DataFrame([tmp], columns=columns)
        ], axis=0)
        
    articles = pd.concat([articles, data], axis=0)

articles = articles.assign(
    year=lambda x: x.year.astype(int),
    article_id=lambda x: x.article_id.map(lambda y: re.sub(r'[a-z\.]', '', str(y))).astype(int)
)

if body_data_file_path.exists():
    articles = anti_join(
        articles,
        pd.read_csv(body_data_file_path).assign(article_id=lambda x: x.article_id.astype(int)),
        by='article_id'
    )

ニュース本文と要約を取得

上記の処理でニュース ID データの取得と加工が完了しました。このデータのニュース ID のニュース本文と要約をスクレイピングします。この際、サーバー処理に負荷をかけないように、スクレイピングするたびに「waiting_time」で指定した時間の処理待ちをしています。また、取得件数を指定しているのが「n_writing_data」です。

waiting_time = 3              # スクレイピングの間隔
n_writing_data = 10000        # 取得する件数 
article_url = 'http://news.livedoor.com/article/detail/{}/'
body_data_file_path = data_dir_path.joinpath('body_data.csv')
summary_data_file_path = data_dir_path.joinpath('summary_data.csv')

target_articles = articles.sort_values(
    'year', ascending=False
).head(min(len(articles), n_writing_data))

ニュース本文と要約をスクレイピング

加工したニュース ID データから１件ずつスクレイピングし、ニュース本文と要約を獲得します。取得できなかった ID も残し、再度スクレイピングする際に、取得できなかった ID はスクレイピングしないようにします。途中で処理落ちする事を考慮し、50件ずつ保存する処理も入れています。

def read_url_to_soup(url):
    
    try:
        response = urllib.request.urlopen(url)
        html = response.read().decode(response.headers.get_content_charset(), errors='ignore')
        soup = BeautifulSoup(html, 'html.parser')
    except Exception:
        soup = None
    
    return soup

def write_data(data, file_path):
    if file_path.exists():
        data = pd.concat([data, pd.read_csv(file_path)]).assign(
            article_id=lambda x: x.article_id.astype(int)
        ).drop_duplicates()
    data.to_csv(file_path, index=False)

body_data = []
summary_data = []
i = 1
for article_id in tqdm(target_articles['article_id']):

    url = article_url.format(article_id)

    soup = read_url_to_soup(url)
    if soup is None or soup.find(class_='articleBody') is None or soup.find(class_='summaryList') is None:
        body_data.append((article_id, None, None))
        summary_data.append((article_id, None))
    else:

        title = soup.find(id='article-body').find('h1').text.strip()

        body = soup.find(class_='articleBody').find('span', {'itemprop': 'articleBody'}).text
        body = re.sub('\n+', '\n', body)
        body_data.append((article_id, title, body))

        summary_list = soup.find(class_='summaryList').find_all('li')
        summary_list = list(map(lambda x: x.text.strip(), summary_list))

        summary_data.extend([(article_id, summary) for summary in summary_list])
    
    if i % 50 == 0:        
        body_data = pd.DataFrame(body_data, columns=['article_id', 'title', 'text'])
        summary_data = pd.DataFrame(summary_data, columns=['article_id', 'text'])
        write_data(data=body_data, file_path=body_data_file_path)
        write_data(data=summary_data, file_path=summary_data_file_path)
        body_data = []
        summary_data = []        

    i += 1
    time.sleep(waiting_time)

if len(body_data) > 0:
    body_data = pd.DataFrame(body_data, columns=['article_id', 'title', 'text'])
    summary_data = pd.DataFrame(summary_data, columns=['article_id', 'text'])
    write_data(data=body_data, file_path=body_data_file_path)
    write_data(data=summary_data, file_path=summary_data_file_path)

取得したデータの確認

取得したデータをロードして確認します。

data = pd.read_csv(body_data_file_path)
data.head()

data = pd.read_csv(summary_data_file_path)
data.head()

意図したデータを取得できていることがわかります。