Le ministère des Affaires intérieures et des Communications eStat publie des données sur les statistiques gouvernementales. Comment obtenir des données statistiques gouvernementales
Il y a. Cet article décrit principalement 2 (en utilisant e-Stat API). Utilisez request et ʻurllib` de la même manière que vous utiliseriez normalement une API en Python. Pour les spécifications détaillées de l'API eStat (version 3.0), voir Spécifications de l'API (ver 3.0). Voir 0).
Le web scraping est possible sur la page d'accueil e-Stat car les données sont fournies par la base de données avec une URL structurée. À titre d'exemple, l'URL d'e-Stat liée au "Tableau des résultats détaillés de l'enquête sur les ménages / Balance des ménages pour les ménages de deux personnes ou plus"
https://www.e-stat.go.jp/stat-search/files?page=1&layout=datalist&toukei=00200561&tstat=000000330001&cycle=1&year=20200&month=12040606&tclass1=000000330001&tclass2=000000330004&tclass3=000000330005&stat_infid=000031969452&result_back=1
A la correspondance suivante entre chaque élément de classification et la spécification de paramètre HTTP GET (" & <param> = <valeur> ").
| article | valeur | URL params |
|---|---|---|
| Nom des statistiques gouvernementales | Sondage de foyer | &toukei=00200561 |
| Nom des statistiques fourni | Sondage de foyer | &tstat=000000330001 |
| Classification de l'offre 1 | Solde du ménage | tclass1=000000330001 |
| Classification de l'offre 2 | Ménages de deux personnes ou plus (résultats hors agriculture, sylviculture et ménages de pêcheurs) | tclass2=000000330002 |
| Classification de l'offre 3 | Tableau de résultats détaillé | tclass3=000000330003 |
| Cycle de l'offre | Mensuel | &cycle=1 |
| Date de l'enquête | Janvier 2000 | &year=20000 &month=11010301 |
** Code des statistiques gouvernementales **
Les données publiées par eStat sont identifiées par le code statistique du gouvernement ou l'ID du tableau statistique. Par exemple, le code des statistiques gouvernementales du recensement national est 00200521 et le code des statistiques gouvernementales de l'enquête auprès des ménages est 00200561.
Lors de l'acquisition de données à partir d'e-Stat, spécifiez d'abord le «code des statistiques gouvernementales» et effectuez une recherche conditionnelle du tableau statistique inclus dans les statistiques gouvernementales spécifiées. Il existe d'autres codes définis par le ministère de l'Intérieur et des Communications, tels que les codes de préfecture et les codes de ville / quartier / ville / village. Le code auquel se référer lors de l'exécution d'une recherche conditionnelle est décrit ici.
** Étape 1: Obtenez l'ID d'application de l'API eStat. ** **
Accédez à la page d'accueil de l 'API eStats et effectuez la procédure à partir de «Enregistrement / connexion de l'utilisateur». Après avoir enregistré votre nom, votre adresse e-mail et votre mot de passe, votre identifiant de candidature vous sera envoyé par e-mail. Les appels API ne peuvent pas être effectués sans ID d'application.
** étape 2: appelez HTTP GET et recevez la réponse (données) de l'API. ** **
Pour plus de détails sur l'utilisation de l'API, reportez-vous à Spécification API (ver 3.0). En gros, si vous pouvez passer les deux appels suivants, il n'y aura pratiquement aucun problème.
méthode getStatsListURL () de la classe`. Voir Spécifications 2.1
--Paramètre: Voir Spécification 3.2 méthode getStatsDataURL () de la classe `. Voir Spécification 2.3
--Paramètre: Voir Spécification 3.4.Précisez en vous référant aux spécifications de l'API e-Stat (version 3.0.0) J'ai créé un module pour générer une URL appropriée pour les paramètres spécifiés.
import urllib
import requests
class EstatRestAPI_URLParser:
"""
This is a simple python module class for e-Stat API (ver.3.0).
See more details at https://www.e-stat.go.jp/api/api-info/e-stat-manual3-0
"""
def __init__(self, api_version=None, app_id=None):
# base url
self.base_url = "https://api.e-stat.go.jp/rest"
# e-Stat REST API Version
if api_version is None:
self.api_version = "3.0"
else:
self.api_version = api_version
# Application ID
if app_id is None:
self.app_id = "****************" #Entrez l'ID de l'application ici
else:
self.app_id = app_id
def getStatsListURL(self, params_dict, format="csv"):
"""
2.1 Obtenir des informations de tableau statistique(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getStatsList?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/getStatsList?{params_str}"
)
elif format == "jsonp":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/jsonp/getStatsList?{params_str}"
)
elif format == "csv":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getSimpleStatsList?{params_str}"
)
return url
def getMetaInfoURL(self, params_dict, format="csv"):
"""
2.2 Acquisition de méta-informations(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getMetaInfo?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/getMetaInfo?{params_str}"
)
elif format == "jsonp":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/jsonp/getMetaInfo?{params_str}"
)
elif format == "csv":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getSimpleMetaInfo?{params_str}"
)
return url
def getStatsDataURL(self, params_dict, format="csv"):
"""
2.3 Obtenez des données statistiques(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getStatsData?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/getStatsData?{params_str}"
)
elif format == "jsonp":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/jsonp/getStatsData?{params_str}"
)
elif format == "csv":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getSimpleStatsData?{params_str}"
)
return url
def postDatasetURL(self):
"""
2.4 Enregistrement de l'ensemble de données(HTTP POST)
"""
url = (
f"{self.base_url}/{self.api_version}"
"/app/postDataset"
)
return url
def refDataset(self, params_dict, format="xml"):
"""
2.5 Référence du jeu de données(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
+ f"/app/refDataset?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/refDataset?{params_str}"
)
elif format == "jsonp":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/jsonp/refDataset?{params_str}"
)
return url
def getDataCatalogURL(self, params_dict, format="xml"):
"""
2.6 Acquisition des informations du catalogue de données(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getDataCatalog?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/getDataCatalog?{params_str}"
)
elif format == "jsonp":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/jsonp/getDataCatalog?{params_str}"
)
return url
def getStatsDatasURL(self, params_dict, format="xml"):
"""
2.7 Acquisition collective de données statistiques(HTTP GET)
"""
params_str = urllib.parse.urlencode(params_dict)
if format == "xml":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getStatsDatas?{params_str}"
)
elif format == "json":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/json/getStatsDatas?{params_str}"
)
elif format == "csv":
url = (
f"{self.base_url}/{self.api_version}"
f"/app/getSimpleStatsDatas?{params_str}"
)
return url
import csv
import json
import xlrd
import zipfile
import requests
import functools
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
def get_json(url):
"""
Request a HTTP GET method to the given url (for REST API)
and return its response as the dict object.
Args:
====
url: string
valid url for REST API
"""
try:
print("HTTP GET", url)
r = requests.get(url)
json_dict = r.json()
return json_dict
except requests.exceptions.RequestException as error:
print(error)
def download_json(url, filepath):
"""
Request a HTTP GET method to the given url (for REST API)
and save its response as the json file.
Args:
url: string
valid url for REST API
filepath: string
valid path to the destination file
"""
try:
print("HTTP GET", url)
r = requests.get(url)
json_dict = r.json()
json_str = json.dumps(json_dict, indent=2, ensure_ascii=False)
with open(filepath, "w") as f:
f.write(json_str)
except requests.exceptions.RequestException as error:
print(error)
def download_csv(url, filepath, enc="utf-8", dec="utf-8", logging=False):
"""
Request a HTTP GET method to the given url (for REST API)
and save its response as the csv file.
Args:
=====
url: string
valid url for REST API
filepathe: string
valid path to the destination file
enc: string
encoding type for a content in a given url
dec: string
decoding type for a content in a downloaded file
dec = 'utf-8' for general env
dec = 'sjis' for Excel on Win
dec = 'cp932' for Excel with extended JP str on Win
logging: True/False
flag whether putting process log
"""
try:
if logging:
print("HTTP GET", url)
r = requests.get(url, stream=True)
with open(filepath, 'w', encoding=enc) as f:
f.write(r.content.decode(dec))
except requests.exceptions.RequestException as error:
print(error)
def download_all_csv(
urls,
filepathes,
max_workers=10,
enc="utf-8",
dec="utf-8"):
"""
Request some HTTP GET methods to the given urls (for REST API)
and save each response as the csv file.
(!! This method uses multi threading when calling HTTP GET requests
and downloading files in order to improve the processing speed.)
Args:
=====
urls: list of strings
valid urls for REST API
filepathes: list of strings
valid pathes to the destination file
max_workers: int
max number of working threads of CPUs within executing this method.
enc: string
encoding type for a content in a given url
dec: string
decoding type for a content in a downloaded file
dec = 'utf-8' for general env
dec = 'sjis' for Excel on Win
dec = 'cp932' for Excel with extended JP str on Win
logging: True/False
"""
func = functools.partial(download_csv, enc=enc, dec=dec)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(
tqdm(executor.map(func, urls, filepathes), total=len(urls))
)
del results
Système de statistiques sociales et démographiques (municipalités dans les statistiques) fourni par le ministère de l'Intérieur et des Communications chaque année. À partir de, chaque élément est agrégé via l'API eStat avec la ville, le quartier, la ville et le village comme unité de zone, et enregistré en tant que fichier local.
import os
from pprint import pprint
from estat_api import EstatRestAPI_URLParser
appId = "****************" #Entrez l'ID de l'application ici
estatapi_url_parser = EstatRestAPI_URLParser() # URL Parser
def search_tables():
"""
Prams (dictionary) to search eStat tables.
For more details, see also
https://www.e-stat.go.jp/api/api-info/e-stat-manual3-0#api_3_2
- appId: Application ID (*required)
- lang:Langue(J:Japonais, E:Anglais)
- surveyYears:Date de l'enquête(YYYYY or YYYYMM or YYYYMM-YYYYMM)
- openYears:Identique à la date de l'enquête
- statsField:Champ Statistiques(2 chiffres:Classification statistique majeure,4 chiffres:Sous-classe statistique)
- statsCode:Code des statistiques gouvernementales(8 chiffres)
- searchWord:Mot-clé de recherche
- searchKind:Type de données(1:Statistiques, 2:Petite zone / maillage régional)
- collectArea:Classification des zones agrégées(1:À l'échelle nationale, 2:Préfectures, 3:Municipalité)
- explanationGetFlg:Existence d'informations de commentaire(Y or N)
- ...
"""
appId = "65a9e884e72959615c2c7c293ebfaeaebffb6030" # Application ID
params_dict = {
"appId": appId,
"lang": "J",
"statsCode": "00200502",
"searchWord": "Système social / démographique", # "Comment la ville, le quartier, la ville et le village sont vus dans les statistiques",
"searchKind": 1,
"collectArea": 3,
"explanationGetFlg": "N"
}
url = estatapi_url_parser.getStatsListURL(params_dict, format="json")
json_dict = get_json(url)
# pprint(json_dict)
if json_dict['GET_STATS_LIST']['DATALIST_INF']['NUMBER'] != 0:
tables = json_dict["GET_STATS_LIST"]["DATALIST_INF"]["TABLE_INF"]
else:
tables = []
return tables
def parse_table_id(table):
return table["@id"]
def parse_table_raw_size(table):
return table["OVERALL_TOTAL_NUMBER"]
def parse_table_urls(table_id, table_raw_size, csv_raw_size=100000):
urls = []
for j in range(0, int(table_raw_size / csv_raw_size) + 1):
start_pos = j * csv_raw_size + 1
params_dict = {
"appId": appId, # Application ID
"lang": "J", #Langue(J:Japonais, E:Anglais)
"statsDataId": str(table_id), #ID de la table statistique
"startPosition": start_pos, #Ligne de départ
"limit": csv_raw_size, #Nombre d'acquisitions de données
"explanationGetFlg": "N", #Existence d'informations de commentaire(Y or N)
"annotationGetFlg": "N", #Présence ou absence d'informations d'annotation(Y or N)
"metaGetFlg": "N", #Présence ou absence de méta-informations(Y or N)
"sectionHeaderFlg": "2", #Indicateur d'en-tête CSV(1:Avoir, 2:Avoir無)
}
url = estatapi_url_parser.getStatsDataURL(params_dict, format="csv")
urls.append(url)
return urls
if __name__ == '__main__':
CSV_RAW_SIZE = 100000
# list of tables
tables = search_tables()
# extract all table ids
if len(tables) == 0:
print("No tables were found.")
elif len(tables) == 1:
table_ids = [parse_table_id(tables[0])]
else:
table_ids = list(map(parse_table_id, tables))
# list of urls
table_urls = []
table_raw_size = list(map(parse_table_raw_size, tables))
for i, table_id in enumerate(table_ids):
table_urls = table_urls + parse_table_urls(table_id, table_raw_size[i])
# list of filepathes
filepathes = []
for i, table_id in enumerate(table_ids):
table_name = tables[i]["TITLE_SPEC"]["TABLE_NAME"]
table_dir = f"./downloads/tmp/{table_name}_{table_id}"
os.makedirs(table_dir, exist_ok=True)
for j in range(0, int(table_raw_size[i] / CSV_RAW_SIZE) + 1):
filepath = f"{table_dir}/{table_name}_{table_id}_{j}.csv"
filepathes.append(filepath)
download_all_csv(table_urls, filepathes, max_workers=30)
Recommended Posts