下記の住居地は県外で処理
| 住居地 | カウント |
|---|---|
| 岐阜県 | 23 |
| 東京都 | 18 |
| 尾張地方 | 11 |
| 三重県 | 11 |
| 三河地方 | 5 |
| 神奈川県 | 3 |
| 静岡県 | 3 |
| 兵庫県 | 2 |
| 千葉県 | 2 |
| 中国武漢市 | 2 |
| 福岡県 | 2 |
| 京都府 | 2 |
| 大阪府 | 2 |
| 福井県 | 1 |
| 滋賀県 | 1 |
| 土岐市 | 1 |
| 可児市 | 1 |
| 一宮保健所管内 | 1 |
| 不定 | 1 |
| 愛媛県 | 1 |
| 茨城県 | 1 |
| 石川県 | 1 |
| 沖縄県 | 1 |
| 岐阜市 | 1 |
開始日と終了日が固定だとreindexで補完すればいいので簡単
以下のコードをPandasで移植
スクレイピング
!pip install pdfplumber !pip install simplejson
import datetime import pathlib import re import sys from urllib.parse import urljoin import pandas as pd import pdfplumber import requests from bs4 import BeautifulSoup def fetch_file(url, dir="."): r = requests.get(url) r.raise_for_status() p = pathlib.Path(dir, pathlib.PurePath(url).name) p.parent.mkdir(parents=True, exist_ok=True) with p.open(mode="wb") as fw: fw.write(r.content) return p def days2date(s): y = 2021 if s.name > 16576 else 2020 days = re.findall("[0-9]{1,2}", s["発表日"]) if len(days) == 2: m, d = map(int, days) return pd.Timestamp(year=y, month=m, day=d) else: return pd.NaT url = "https://www.pref.aichi.jp/site/covid19-aichi/" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; rv:11.0) like Gecko" } JST = datetime.timezone(datetime.timedelta(hours=+9), "JST") dt_now = datetime.datetime.now(JST) str_now = dt_now.strftime("%Y/%m/%d %H:%M") r = requests.get(url, headers=headers) r.raise_for_status() soup = BeautifulSoup(r.content, "html.parser") dfs = [] for tag in soup.find("span", text="▶ 愛知県内の発生事例").parent.find_all( "a", href=re.compile(".pdf$") )[::-1]: link = urljoin(url, tag.get("href")) path_pdf = fetch_file(link) with pdfplumber.open(path_pdf) as pdf: for page in pdf.pages: table = page.extract_table() df_tmp = pd.DataFrame(table[1:], columns=table[0]) dfs.append(df_tmp) df = pd.concat(dfs).set_index("No") # 発表日が欠損を削除 df.dropna(subset=["発表日"], inplace=True) df.index = df.index.astype(int) df.sort_index(inplace=True) df["発表日"] = df.apply(days2date, axis=1) df["date"] = df["発表日"].dt.strftime("%Y-%m-%d") df["w"] = (df["発表日"].dt.dayofweek + 1) % 7 df["w"] = df["w"].astype(str) df["short_date"] = df["発表日"].dt.strftime("%m\\/%d") df["発表日"] = df["発表日"].dt.strftime("%Y/%m/%d %H:%M") cjk = str.maketrans("⻲⻑黑戶⻯⻄⻘⻤", "亀長黒戸竜西青鬼") df["住居地"] = df["住居地"].str.normalize("NFKC") df["住居地"] = df["住居地"].apply(lambda s: s.translate(cjk)) df["年代・性別"] = df["年代・性別"].str.normalize("NFKC") p = pathlib.Path("/data/patients.csv") p.parent.mkdir(parents=True, exist_ok=True) df.to_csv(p, encoding="utf_8_sig")
集計
""" aichi_citys = [ "名古屋市", "一宮市", "瀬戸市", "春日井市", "犬山市", "江南市", "小牧市", "稲沢市", "尾張旭市", "岩倉市", "豊明市", "日進市", "清須市", "北名古屋市", "長久手市", "東郷町", "豊山町", "大口町", "扶桑町", "津島市", "愛西市", "弥富市", "あま市", "大治町", "蟹江町", "飛島村", "半田市", "常滑市", "東海市", "大府市", "知多市", "阿久比町", "東浦町", "南知多町", "美浜町", "武豊町", "岡崎市", "碧南市", "刈谷市", "豊田市", "安城市", "西尾市", "知立市", "高浜市", "みよし市", "幸田町", "豊橋市", "豊川市", "蒲郡市", "新城市", "田原市", "設楽町", "東栄町", "豊根村", ] """ # city_data.json with open("city_data.json", "r") as fr: json_load = json.load(fr) city_dic = {k: v["市町村コード"] for k, v in json_load["data"].items()} aichi_citys = [k for k in city_dic if k != "愛知県"] ages_list = [ "10歳未満", "10代", "20代", "30代", "40代", "50代", "60代", "70代", "80代", "90代", "100代", ] df = pd.read_csv( "/data/patients.csv", keep_default_na=False, ) patients_list = df.values.tolist() df["発表日"] = pd.to_datetime(df["発表日"]) df["住居地"] = df["住居地"].where(df["住居地"].isin(aichi_citys), "県外") # 年代と性別を分割 df_ages = df["年代・性別"].str.extract("(.+)(男性|女性|その他)").rename(columns={0: "年代", 1: "性別"}) df = df.join(df_ages) df["年代"] = df["年代"].str.strip() df["年代"] = df["年代"].where(df["年代"].isin(ages_list), "その他") df["年代"].unique() args = sys.argv # 日付のリストを生成 strdt = datetime.datetime.strptime("2020-01-26", "%Y-%m-%d") # 開始日 enddt = datetime.datetime.strptime(args[1], "%Y-%m-%d") # 終了日 dt_range = pd.date_range(strdt, enddt) df_date_num = df["発表日"].value_counts().reindex(index=dt_range, fill_value=0) df_date_num df_date_place = pd.crosstab(df["発表日"], df["住居地"]).reindex( index=dt_range, columns=aichi_citys + ["県外"], fill_value=0 ) df_date_place df_date_age_sex = pd.crosstab(index=df["発表日"], columns=[df["性別"], df["年代"]]).reindex( index=dt_range, fill_value=0 ) # 県外の市町村コードを99999追加 city_dic.update({"県外": 99999}) # 市町村名を市町村コードに置換 df_date_place.rename(columns=city_dic, inplace=True) df_date_age_sex patients_summary_list = [] for dt_date in dt_range: patients_summary_list.append( { "日付": dt_date.strftime("%Y-%m-%d"), "小計": int(df_date_num.loc[dt_date]), "住居地": df_date_place.loc[dt_date].to_dict(), "年代": df_date_age_sex.loc[dt_date].unstack(fill_value=0).to_dict(), } ) main_summary_history_df = pd.read_csv( "/data/main_summary_history.csv", keep_default_na=False, ) inspection_persons_summary_df = pd.read_csv( "/data/inspection_persons_summary.csv", keep_default_na=False, ) inspections_summary_df = pd.read_csv( "/data/inspections_summary.csv", parse_dates=[0], keep_default_na=False, ) inspections_summary_df.rename(columns={"検査日": "日付", "PCR検査件数（件）": "小計"}, inplace=True) inspections_summary_df.drop(columns=["陽性者数（人）", "備考"], inplace=True) inspections_summary_df["日付"] = inspections_summary_df["日付"].dt.strftime("%Y-%m-%d") inspections_summary_list = inspections_summary_df.to_dict(orient="records") inspections_summary_list data = { "lastUpdate": str_now, "patients": {"date": str_now, "data": patients_list}, "patients_summary": {"date": str_now, "data": patients_summary_list}, "inspections_summary": {"date": str_now, "data": inspections_summary_list}, "main_summary_history": { "date": str_now, "data": main_summary_history_df.to_dict(orient="records"), }, "inspection_persons_summary": { "date": str_now, "data": inspection_persons_summary_df.to_dict(orient="records"), }, } import json with open("data.json", "w", encoding="utf-8") as fw: json.dump(data, fw, ensure_ascii=False, indent=4)
