LoginSignup
3
4

More than 3 years have passed since last update.

【Python】新型コロナウイルスの都道県別情報を自動で厚生労働省のPDFから読み取ってスプレッドシートやエクセルに書き込む

Posted at

はじめに

厚生労働省のホームページ上には毎日更新される新型コロナウイルスの情報が記載されています。
新型コロナウイルス感染症に関する報道発表資料(発生状況、国内の患者発生、海外の状況、その他)
そこから都道府県別の詳細なデータを(PDFの書式が同一な5/10以降のデータのみではありますが)収集しGoogleSpreadSheetやExcelに記録するプログラムを作成したので紹介したいと思います。

表計算ソフトに記録するので、後から簡単に感染者数や重症化率等の推移をグラフ化などして俯瞰することが出来ます。

準備

Excelで使用する場合はそのままで使えますが、PythonのプログラムからGoogleSpreadSheetにアクセスするために少々準備が必要です。
下記サイト等を参照して、下準備をし、プログラム内のコメント部分(105行目付近)にjsonファイルの名前やスプレッドシートキーを入力してください。
【もう迷わない】Pythonでスプレッドシートに読み書きする初期設定まとめ
PythonでGoogleスプレッドシートを編集

プログラム

import requests
import re
import regex
from bs4 import BeautifulSoup
from tika import parser
import gspread
from oauth2client.service_account import ServiceAccountCredentials
import openpyxl
import pickle

isGss = False if bool(int(input("1. GoogleSpreadSheet  2. Excel    ")) -
                      1) else True

req = requests.get(
    "https://www.mhlw.go.jp/stf/seisakunitsuite/bunya/0000121431_00086.html")
req.encoding = 'utf-8'
soup = BeautifulSoup(req.text, 'html.parser')

urls = {}
try:
    with open("dates.pickle", "rb") as f:
        dates = pickle.load(f)
except:
    dates = []
column_cnt = len(dates) + 2

for i in soup.select("div[class='m-grid__col1']")[1].find_all("a"):
    url = i["href"]
    if "新型コロナウイルス感染症の現在の状況と厚生労働省の対応について" not in i.text: continue
    b = re.findall("令和\d{1,2}年\d{1,2}月\d{1,2}日", i.text)
    if b: date = b[0]
    else:
        temp_r = requests.get(url)
        temp_r.encoding = 'utf-8'
        date = re.findall("令和\d{1,2}年\d{1,2}月\d{1,2}日", temp_r.text)[0]
    date = date.translate(
        str.maketrans({
            "0": "0",
            "1": "1",
            "2": "2",
            "3": "3",
            "4": "4",
            "5": "5",
            "6": "6",
            "7": "7",
            "8": "8",
            "9": "9"
        }))
    date = "".join([(i.zfill(2) if i.isdigit() else i) for i in re.findall(
        "(令和)(\d{1,2})(年)(\d{1,2})(月)(\d{1,2})(日)", date)[0]])
    if re.findall("令和02年05月0[1-9]日|令和02年0[1-4]月\d{1,2}日", date): continue
    elif date not in dates:
        urls[date] = i["href"]
        dates.append(date)

with open("dates.pickle", "wb") as f:
    pickle.dump(dates, f)

pdfs = {}
for date, url in urls.items():
    temp_r = requests.get(url)
    temp_r.encoding = 'utf-8'
    soup = BeautifulSoup(temp_r.text, 'html.parser')
    for j in soup.find_all('a', href=True):
        if '都道府県の検査陽性者の状況' in j.text:
            pdfs[date] = "https://www.mhlw.go.jp" + j['href']


def makeDict(text):
    l = {}
    for i in regex.findall(
            "(\p{Han}(?:\s+\p{Han}|計){1,2}|(その他))([※\d\s]+?[\nG])",
            re.sub("※\d{1,2}", "", text).translate(
                str.maketrans({
                    "\u3000": " ",
                    ",": "",
                    "-": "0"
                }))):
        a = list(map(int, i[1].replace("G", "").split()))
        b = "".join(i[0].split())
        l[b] = {}
        l[b]["陽性者数"] = a[0]
        l[b]["PCR検査実施人数"] = a[1]
        l[b]["入院治療等を要する者"] = a[2]
        l[b]["うち重症"] = a[3]
        l[b]["退院又は療養解除となった者の数"] = a[4]
        l[b]["死亡(累積)"] = a[5]
        l[b]["確認中"] = a[6]
    return l


cnt = 0

if len(pdfs) == 0: pass
elif isGss:
    scope = [
        'https://spreadsheets.google.com/feeds',
        'https://www.googleapis.com/auth/drive'
    ]

    credentials = ServiceAccountCredentials.from_json_keyfile_name('jsonファイルの名前.json', scope) #jsonファイルの名前
    gc = gspread.authorize(credentials)

    SPREADSHEET_KEY = 'スプレッドシートキー' #スプレッドシートキー

    workbook = gc.open_by_key(SPREADSHEET_KEY)
    try:
        worksheets = [
            workbook.worksheet('陽性者数'),
            workbook.worksheet('PCR検査実施人数'),
            workbook.worksheet('入院治療等を要する者'),
            workbook.worksheet('うち重症'),
            workbook.worksheet('退院又は療養解除となった者の数'),
            workbook.worksheet('死亡(累積)'),
            workbook.worksheet('確認中')
        ]
        isFirst = False
    except:
        workbook.add_worksheet(title='陽性者数', rows=100, cols=500)
        workbook.add_worksheet(title='PCR検査実施人数', rows=100, cols=500)
        workbook.add_worksheet(title='入院治療等を要する者', rows=100, cols=500)
        workbook.add_worksheet(title='うち重症', rows=100, cols=500)
        workbook.add_worksheet(title='退院又は療養解除となった者の数', rows=100, cols=500)
        workbook.add_worksheet(title='死亡(累積)', rows=100, cols=500)
        workbook.add_worksheet(title='確認中', rows=100, cols=500)
        workbook.del_worksheet(workbook.sheet1)
        worksheets = [
            workbook.worksheet('陽性者数'),
            workbook.worksheet('PCR検査実施人数'),
            workbook.worksheet('入院治療等を要する者'),
            workbook.worksheet('うち重症'),
            workbook.worksheet('退院又は療養解除となった者の数'),
            workbook.worksheet('死亡(累積)'),
            workbook.worksheet('確認中')
        ]
        isFirst = True

    cas = [
        worksheet.range(1, column_cnt, 50, column_cnt + len(pdfs) - 1)
        for worksheet in worksheets
    ]
    for date, url in sorted(pdfs.items()):
        print(date)
        file_data = parser.from_buffer(requests.get(url))
        text = file_data["content"]

        l = makeDict(text)
        if isFirst:
            for worksheet in worksheets:
                cells = worksheet.range(2, 1, 50, 1)
                for i, key in enumerate(l.keys()):
                    cells[i].value = key
                worksheet.update_cells(cells)
            isFirst = False
        for i, worksheet in enumerate(worksheets):
            cells = cas[i][cnt::len(pdfs)]
            cells[0].value = date
            for j, a in enumerate(l.keys()):
                key = list(l[a].keys())[i]
                cells[j + 1].value = l[a][key]
        cnt += 1

    for i, worksheet in enumerate(worksheets):
        worksheet.update_cells(cas[i])

else:
    filename = "covid19.xlsx"
    try:
        wb = openpyxl.load_workbook(filename)
        wss = [
            wb.get_sheet_by_name('陽性者数'),
            wb.get_sheet_by_name('PCR検査実施人数'),
            wb.get_sheet_by_name('入院治療等を要する者'),
            wb.get_sheet_by_name('うち重症'),
            wb.get_sheet_by_name('退院又は療養解除となった者の数'),
            wb.get_sheet_by_name('死亡(累積)'),
            wb.get_sheet_by_name('確認中')
        ]
        isFirst = False
    except:
        wb = openpyxl.Workbook()
        wss = [
            wb.create_sheet('陽性者数'),
            wb.create_sheet('PCR検査実施人数'),
            wb.create_sheet('入院治療等を要する者'),
            wb.create_sheet('うち重症'),
            wb.create_sheet('退院又は療養解除となった者の数'),
            wb.create_sheet('死亡(累積)'),
            wb.create_sheet('確認中')
        ]
        wb.remove(wb.get_sheet_by_name('Sheet'))
        isFirst = True

    for date, url in sorted(pdfs.items()):
        print(date)
        file_data = parser.from_buffer(requests.get(url))
        text = file_data["content"]

        l = makeDict(text)
        if isFirst:
            for ws in wss:
                for i, key in enumerate(l.keys()):
                    ws.cell(i + 2, 1, key)
            isFirst = False
        for i, ws in enumerate(wss):
            ws.cell(1, column_cnt + cnt, date)
            for j, a in enumerate(l.keys()):
                key = list(l[a].keys())[i]
                ws.cell(j + 2, column_cnt + cnt, l[a][key])
        cnt += 1

    wb.save(filename)

使い方

保存してPythonから実行するだけです。
記録済みの日付の保存用にdates.pickleが生成されます、実行ファイルと同じディレクトリから動かさないようにしてください。

結果

image.png

感染者数の推移なども簡単に可視化出来ます。
image.png

まとめ

全国の感染者数の推移を見てみると第二波も落ち着いてきてるのかな…?
感染推移を詳細に観測したい方、統計的な処理をしてみたい方、お試しください。
タスクスケジューラやGCFなどを使って定期実行させれば自動的に更新できそうです。

3
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
3
4