Pythonで国土数値情報のWeb APIを使って全データを一括でダウンロードする方法

さて、今回も国土数値情報のWeb APIについて書いてみようと思います。先日、本件に関するエントリーを書きましたが、この時は一部データのダウンロードしかしませんでした。その時に全データを一括でダウンロードできたらいいなぁと思ったので、今回はそれについて書いてみようと思います。

国土数値情報のWeb APIとは？

以下エントリーで紹介しています。興味のある方はぜひ読んでみてください。

www.gis-py.com

ダウンロード対象

国土数値情報は以下の画像のように各カテゴリごとにデータをダウンロードできます。

f:id:sanvarie:20190215154354p:plain

そして、各カテゴリごとに年度、都道府県ごとにデータをダウンロードすることができます（カテゴリによっては一つの年度のみ、または、全国単位でのみダウンロード可能）。
f:id:sanvarie:20190215154622p:plain

今回は・・・

①大分類（例：1. 国土（水・土地））
②中分類（例：＜水域＞）
③小分類（例：海岸線）
④年度（最新の年度）

ごとにフォルダを作成してその中にダウンロードしたデータを格納していこうと思います（カテゴリによっては中分類がなかったりします）。

また、最初はすべての年度のデータをダウンロードしようと思ったのですが、どんでもないデータ量になりそうだったのであきらめました（笑）。サンプルコードには全ての年度のデータをとってくるコードを記載しますので（コメントアウトしています）、興味がある方はぜひ試してみてください。

環境

Python2.7.3
Windows10

サンプルコード

DATA_DESTINATIONに任意のディレクトリを指定してください。そこにデータをダウンロードします。

# -*- coding: utf-8 -*-
import zipfile
import os
import os.path
import requests
from lxml import etree

#任意の保存先
DATA_DESTINATION = ur"D:\gis-py\WebAPI"

#国土数値情報の概要情報取得
URL_OVERVIEW = "http://nlftp.mlit.go.jp/ksj/api/1.0b/index.php/app/getKSJSummary.xml" \
               "?appId=ksjapibeta1&lang=J&dataformat=1"

#国土数値情報取得のURL情報取得
url_detail = "http://nlftp.mlit.go.jp/ksj/api/1.0b/index.php/app/getKSJURL.xml" \
              "?appId=ksjapibeta1&lang=J&dataformat=1"

dic = {}

def get_url(url, identifier = 0):
    """
    国土数値情報のWEBAPIを使用して各情報をXMLで取得します。
    """
    if identifier == 0:
        resp = requests.get(url, timeout=10)
    else:
        url = url + "&identifier=%s" % (identifier)
        resp = requests.get(url, timeout=10)

    return etree.fromstring(resp.content)

def download_zip():
    """
    指定したURLからZIPをダウンロードします。
    """
    year = ""
    for key, value in dic.items():
        xml = get_url((url_detail), key)

        for x in xml.iter():

            if x.tag == "year":
                year = x.text

            if x.tag == "zipFileUrl":

                if year == value[0]:
                    year = ""

                    filename = x.text.split('/')[-1]
                    r = requests.get(x.text, stream=True)
                    with open(filename, 'wb') as f:
                        for chunk in r.iter_content(chunk_size=1024):
                            if chunk:
                                f.write(chunk)
                                f.flush()

                    uncompress_zip(filename, value[1])

def uncompress_zip(filename, destination):
    """
    ZIP ファイルを指定したディレクトリに展開します。
    """
    zfile = zipfile.ZipFile(filename)
    zfile.extractall(destination)

def make_directory(xml):
    """
    DATA_DESTINATIONの下に各フォルダを作成します。
    """
    element_list = []
    new_directry_list = []

    for x in xml.iter():

        if x.tag == "identifier":
            element_list.append(x.text)
        elif x.tag == "title":
            element_list.append(x.text)
        elif x.tag == "field1":
            element_list.append(x.text)
        elif x.tag == "field2":
            element_list.append(x.text)
        elif x.tag == "areaType":
            element_list.append(x.text)
            new_directry_list.append(element_list)
            element_list = []
            continue

    for n in new_directry_list:
        new_directry = ""
        year_list = []

        new_directry = os.path.join(DATA_DESTINATION, n[2])

        if n[3] != " " and n[3] != "-": #field2が不要なデータがあるため
            new_directry = os.path.join(new_directry, n[3])

        new_directry = os.path.join(new_directry, n[1])

        for x in get_url((url_detail), n[0]).iter("year"):
            year_list.append(x.text)

        year_list = list(set(year_list))
        max_year = max(year_list)

        new_dir = os.path.join(new_directry, max_year)

        if os.path.exists(new_dir) == 0: #なかったら作る
            os.makedirs(new_dir)
        dic[n[0]] = [max_year,new_dir]

        # ↓全ての年度のデータを取得したい場合は以下を使用する
        #year_list = list(set(year_list))
        #year_list.sort()
        #
        #for l in max(year_list):

        #    new_dir = os.path.join(new_directry, l)

        #    if os.path.exists(new_dir) == 0: #なかったら作る
        #        os.makedirs(new_dir)
        #    dic[n[0]] = [l,new_dir]

if __name__ == '__main__':
    make_directory(get_url(URL_OVERVIEW, 0))
    download_zip()