Das Ministerium für innere Angelegenheiten und Kommunikation eStat veröffentlicht Daten zu Regierungsstatistiken. So erhalten Sie Regierungsstatistikdaten
Es gibt. Dieser Artikel beschreibt hauptsächlich 2 (unter Verwendung von e-Stat API). Verwenden Sie "request" und "urllib" auf die gleiche Weise, wie Sie normalerweise eine API in Python verwenden würden. Detaillierte Spezifikationen der eStat-API (Version 3.0) finden Sie unter API-Spezifikationen (Version 3.0). Siehe 0).
Web-Scraping ist auf der e-Stat-Homepage möglich, da Daten aus der Datenbank mit einer strukturierten URL bereitgestellt werden. Beispiel: Die URL von e-Stat bezieht sich auf "Detaillierte Ergebnistabelle für Haushaltsumfragen / Haushaltsbilanzausgaben für Haushalte mit zwei oder mehr Personen".
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
Hat die folgende Entsprechung zwischen jedem Klassifizierungselement und der HTTP-GET-Parameterspezifikation ("& =
| Artikel | Wert | URL params |
|---|---|---|
| Name der Regierungsstatistik | Haushaltsbefragung | &toukei=00200561 |
| Bereitgestellter Statistikname | Haushaltsbefragung | &tstat=000000330001 |
| Angebotsklassifizierung 1 | Haushaltsbilanz | tclass1=000000330001 |
| Angebotsklassifizierung 2 | Haushalte mit zwei oder mehr Personen (Ergebnisse ohne Land-, Forst- und Fischerhaushalte) | tclass2=000000330002 |
| Angebotsklassifizierung 3 | Detaillierte Ergebnistabelle | tclass3=000000330003 |
| Angebotszyklus | Monatlich | &cycle=1 |
| Umfragedatum | Januar 2000 | &year=20000 &month=11010301 |
** Statistikcode der Regierung **
Die von eStat veröffentlichten Daten werden durch den statistischen Code der Regierung oder die ID der statistischen Tabelle identifiziert. Beispielsweise lautet der nationale Statistikcode der Volkszählung 00200521 und der Statistikcode der Regierung der Haushaltsumfrage 00200561.
Geben Sie beim Erfassen von Daten aus e-Stat zunächst den "Regierungsstatistikcode" an und führen Sie eine bedingte Suche nach der in der angegebenen Regierungsstatistik enthaltenen statistischen Tabelle durch. Es gibt andere vom Ministerium für innere Angelegenheiten und Kommunikation festgelegte Codes, wie Präfekturcodes und Stadt- / Gemeinde- / Stadt- / Dorfcodes. Der Code, auf den bei der Durchführung einer bedingten Suche verwiesen wird, wird hier beschrieben.
** Schritt 1: Rufen Sie die Anwendungs-ID der eStat-API ab. ** ** **
Greifen Sie auf die Startseite von eStats API zu und führen Sie den Vorgang unter "Benutzerregistrierung / -anmeldung" aus. Nach der Registrierung Ihres Namens, Ihrer E-Mail-Adresse und Ihres Passworts erhalten Sie Ihre Anwendungs-ID per E-Mail. API-Aufrufe können nicht ohne Anwendungs-ID durchgeführt werden.
** Schritt 2: Rufen Sie HTTP GET auf und empfangen Sie die Antwort (Daten) von der API. ** ** **
Einzelheiten zur Verwendung der API finden Sie unter API-Spezifikation (Version 3.0). Wenn Sie die folgenden zwei Anrufe tätigen können, gibt es grundsätzlich fast kein Problem.
Geben Sie dies unter Bezugnahme auf [Spezifikationen der e-Stat-API (Version 3.0.0)] an (https://www.e-stat.go.jp/api/api-info/e-stat-manual3-0). Ich habe ein Modul erstellt, um eine geeignete URL für die angegebenen Parameter zu generieren.
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 = "****************" #Geben Sie hier die Anwendungs-ID ein
else:
self.app_id = app_id
def getStatsListURL(self, params_dict, format="csv"):
"""
2.1 Holen Sie sich statistische Tabelleninformationen(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 Meta-Informationserfassung(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 Holen Sie sich statistische Daten(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 Datensatzregistrierung(HTTP POST)
"""
url = (
f"{self.base_url}/{self.api_version}"
"/app/postDataset"
)
return url
def refDataset(self, params_dict, format="xml"):
"""
2.5 Datensatzreferenz(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 Datenkataloginformationserfassung(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 Kollektive Erfassung statistischer Daten(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
Sozial- und Bevölkerungsstatistiksystem (Kommunen in Statistik), das jedes Jahr vom Ministerium für innere Angelegenheiten und Kommunikation bereitgestellt wird. Von wird jedes Element über die eStat-API mit der Stadt, der Gemeinde, der Stadt und dem Dorf als Gebietseinheit aggregiert und als lokale Datei gespeichert.
import os
from pprint import pprint
from estat_api import EstatRestAPI_URLParser
appId = "****************" #Geben Sie hier die Anwendungs-ID ein
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:Sprache(J:japanisch, E:Englisch)
- surveyYears:Umfragedatum(YYYYY or YYYYMM or YYYYMM-YYYYMM)
- openYears:Entspricht dem Umfragedatum
- statsField:Statistikfeld(2 Ziffern:Statistische Hauptklassifikation,4 Ziffern:Statistische Unterklasse)
- statsCode:Regierungsstatistikcode(8-stellig)
- searchWord:Suchbegriff
- searchKind:Datentyp(1:Statistiken, 2:Kleine Fläche / regionales Netz)
- collectArea:Aggregierte Gebietsklassifizierung(1:Bundesweit, 2:Präfekturen, 3:Gemeinde)
- explanationGetFlg:Vorhandensein von Kommentarinformationen(Y or N)
- ...
"""
appId = "65a9e884e72959615c2c7c293ebfaeaebffb6030" # Application ID
params_dict = {
"appId": appId,
"lang": "J",
"statsCode": "00200502",
"searchWord": "Soziales / demografisches System", # "Wie Stadt, Gemeinde, Stadt und Dorf in der Statistik dargestellt werden",
"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", #Sprache(J:japanisch, E:Englisch)
"statsDataId": str(table_id), #Statistische Tabellen-ID
"startPosition": start_pos, #Startlinie
"limit": csv_raw_size, #Anzahl der Datenerfassungen
"explanationGetFlg": "N", #Vorhandensein von Kommentarinformationen(Y or N)
"annotationGetFlg": "N", #Vorhandensein oder Nichtvorhandensein von Anmerkungsinformationen(Y or N)
"metaGetFlg": "N", #Vorhandensein oder Nichtvorhandensein von Metainformationen(Y or N)
"sectionHeaderFlg": "2", #CSV-Header-Flag(1:Erhalten, 2:Erhalten無)
}
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