Pobieranie danych z rejestru TERYT

GUS udostępnia usługę sieciową TERYT ws1.
Dane przesyłane są w formacie XML protokołem SOAP.
Potrzebujemy więc klienta, który obsłuży ten protokół, np. zeep:

pip install zeep

Aby aplikacja mogła pobierać dane z rejestru musimy przejść uwierzytelnienie.
Dane do logowania zapiszemy w słowniku:

CREDENTIALS = {
    'wsdl': 'https://uslugaterytws1test.stat.gov.pl/wsdl/terytws1.wsdl',
    'username': 'TestPubliczny',
    'password': '1234abcd'
}

Są to dane do środowiska testowego.
Aby korzystać z usługi na produkcji,
należy wysłać maila do GUSu z prośbą o założenie prywatnego konta.

Tworzymy nową instancję klienta:

from zeep import Client
from zeep.wsse.username import UsernameToken

token = UsernameToken(
    username=CREDENTIALS['username'],
    password=CREDENTIALS['password']
)
client = Client(wsdl=CREDENTIALS['wsdl'], wsse=token)

Sprawdzamy czy uda się nawiązać połączenie z usługą:

print(client.service.CzyZalogowany())

Jeżeli wszystko jest ok, powinno wypisać True.

Update: W przypadku wystąpienia błędu:

zeep.exceptions.XMLParseError: The namespace defined on the xsd:import doesn't match the imported targetNamespace located at 'https://uslugaterytws1test.stat.gov.pl/wsdl/xsd1.xsd' (https://uslugaterytws1test.stat.gov.pl/wsdl/terytws1.wsdl:53)

należy pobrać wskazany w komunikacie plik terytws1.wsdl i w linii nr 53 zmienić odwołanie z xsd1.xsd na xsd2.xsd, a następnie w słowniku CREDENTIALS dla klucza wsdl podać ścieżkę do zmodyfikowanego pliku terytws1.wsdl.

Kiedy mamy już do dyspozycji obiekt klienta,
możemy na nim wywoływać metody dostępne w TERYT ws1
(pełna lista metod w instrukcji).

Wiele z tych metod wymaga podania daty jako argumentu. Zatem:

from datetime import datetime

STATE_DATE = datetime.now()

Spróbujmy pobrać listę województw:

client.service.PobierzListeWojewodztw(STATE_DATE)

Wywołanie zwróci listę obiektów słownikopodobnego typu JednostkaTerytorialna.
Oto jeden z nich:

{
    'GMI': None,
    'NAZWA': 'DOLNOŚLĄSKIE',
    'NAZWA_DOD': 'województwo',
    'POW': None,
    'RODZ': None,
    'STAN_NA': '1/2/2018 12:00:00 AM',
    'WOJ': '02'
}

Możemy więc, powołując się na klucz 'NAZWA', stworzyć listę złożoną z nazw:

[e['NAZWA'] for e in client.service.PobierzListeWojewodztw(STATE_DATE)]

I voilà:

['DOLNOŚLĄSKIE', 'KUJAWSKO-POMORSKIE', 'LUBELSKIE', 'LUBUSKIE', 'ŁÓDZKIE', 
'MAŁOPOLSKIE', 'MAZOWIECKIE', 'OPOLSKIE', 'PODKARPACKIE', 'PODLASKIE', 
'POMORSKIE', 'ŚLĄSKIE', 'ŚWIĘTOKRZYSKIE', 'WARMIŃSKO-MAZURSKIE', 
'WIELKOPOLSKIE', 'ZACHODNIOPOMORSKIE']

Wyszukiwanie jednostek

Jednostki terytorialne możemy wyszukiwać m.in. po nazwie:

client.service.WyszukajJPT(nazwa='Warszawa')

[{
    'GmiNazwa': None,
    'GmiNazwaDodatkowa': 'miasto stołeczne, na prawach powiatu',
    'GmiRodzaj': None,
    'GmiSymbol': None,
    'PowSymbol': '65',
    'Powiat': 'Warszawa',
    'WojSymbol': '14',
    'Wojewodztwo': 'MAZOWIECKIE'
}, {
    'GmiNazwa': 'Warszawa',
    'GmiNazwaDodatkowa': 'gmina miejska, miasto stołeczne',
    'GmiRodzaj': '1',
    'GmiSymbol': '01',
    'PowSymbol': '65',
    'Powiat': 'Warszawa',
    'WojSymbol': '14',
    'Wojewodztwo': 'MAZOWIECKIE'
}]

Wyszukiwanie z użyciem identyfikatora TERC
wymaga stworzenia specjalnego obiektu klasy identyfikatory.
Posłuży nam do tego fabryka, którą dostarcza klient:

factory = client.type_factory('ns2')

Prefix 'ns2' wskazuje na określoną przestrzeń nazw,
w której zadeklarowane zostały klasy obiektów.
Ale skąd wiemy jakiego namespace użyć? Oto ściągawka:

print(client.wsdl.dump())

Powyższe polecenie wyświetli wszystkie dostępne prefixy, klasy, metody,
jednym słowem cały schemat usługi.

Tworzymy nowy identyfikator podając string z numerem TERC:

identyfikator = factory.identyfikatory(terc='1465011')

Ale to nie wszystko. Metoda WyszukajJednostkeWRejestrze()
wymaga podania listy identyfikatorów, a nie pojedynczego numeru.

Uruchamiamy fabrykę ponownie:

array = factory.ArrayOfidentyfikatory(identyfikator)

Do pełni szczęścia pozostaje nam wskazać kategorię szukanej jednostki.
Zgodnie z instrukcją "0" oznacza wyszukiwanie wśród wszystkich rodzajów.

client.service.WyszukajJednostkeWRejestrze(identyfiks=array, kategoria=0, DataStanu=STATE_DATE)

[{
    'GmiNazwa': 'Warszawa',
    'GmiNazwaDodatkowa': 'gmina miejska, miasto stołeczne',
    'GmiRodzaj': '1',
    'GmiSymbol': '01',
    'PowSymbol': '65',
    'Powiat': 'Warszawa',
    'WojSymbol': '14',
    'Wojewodztwo': 'MAZOWIECKIE'
}]

Prościej wygląda sprawa z miejscowościami. Można od razu użyć numeru SIMC:

client.service.WyszukajMiejscowosc(identyfikatorMiejscowosci='0329898')

[{
    'GmiRodzaj': '2',
    'GmiSymbol': '04',
    'Gmina': 'Pcim',
    'Nazwa': 'Pcim',
    'PowSymbol': '1209',
    'Powiat': 'myślenicki',
    'Symbol': '0329898',
    'WojSymbol': '12',
    'Wojewodztwo': 'MAŁOPOLSKIE'
}]

Pobieranie katalogów

Usługa umożliwia również pobieranie całych zbiorów danych.
W odpowiedzi na żądanie otrzymujemy obiekt klasy PlikKatalog
posiadającej właściwości:

• nazwa_pliku – string z sugerowaną nazwą pliku
• plik_zawartosc – string z zakodowaną w Base64 treścią pliku zip
• opis – string z opisem pliku.

A zatem przesłany zostaje plik binarny,
podobnie jak załączniki w poczcie elektronicznej.

Pobierzemy teraz katalog miejscowości:

catalog = client.service.PobierzKatalogSIMC(STATE_DATE)

Jego właściwości zapiszemy do zmiennych:

filename = catalog['nazwa_pliku']
content = catalog['plik_zawartosc']

Można oczywiście użyć własnej nazwy pliku, np.: filename='katalog.zip',
a także zmienić ścieżkę: filename=os.path.expanduser('~/Desktop/katalog.zip').

Zawartość pliku odkodujemy używając funkcji b64decode()
z modułu base64 z biblioteki standardowej:

from base64 import b64decode

decoded = b64decode(content)

A tak wygląda treść zakodowana i odkodowana (fragment):

CONTENT: c0bANbUuAEcAAAAU0lNQ19VcnplZG9
DECODED: b'\x01\x1c\x00\x00\x00SIMC_Urzedowy_2018-11-23.'

Odkodowaną treść zapiszemy jako plik na dysku pod wskazaną nazwą:

with open(filename, 'wb') as file:
    file.write(decoded)
    file.close()

Plik zip został "zmaterializowany" i można go otworzyć.
Ale nie będziemy przecież otwierać tego ręcznie.
Biblioteka standardowa oferuje odpowiednie narzędzia:

from zipfile import ZipFile

zf = ZipFile(filename, 'r')

Otrzymaliśmy pythonową reprezentację zapisanego wcześniej pliku.
Sprawdźmy co znajduje się wewnątrz:

print(zf.namelist())

Widzimy, że znajdują się tam dwa pliki, XML i CSV:

['SIMC_Urzedowy_2018-11-23.xml', 'SIMC_Urzedowy_2018-11-23.csv']

Przeczytajmy teraz pierwszy z nich (ale nie wszystko, kilobajt tylko, stąd n=1024):

with zf.open(zf.namelist()[0]) as xml_file:
    print(xml_file.read(n=1024))

Oczywiście parsowanie XML to temat na osobny artykuł,
ale wylistujmy sobie chociaż nazwy miejscowości jakimś prostym narzędziem:

from xml.dom import minidom

with zf.open(zf.namelist()[0]) as xml_file:
    DOMTree = minidom.parse(xml_file)

    children = DOMTree.childNodes
    for row in children[0].getElementsByTagName('row'):
        print(row.getElementsByTagName('NAZWA')[0].childNodes[0].toxml())

Konsola zaczyna wyrzucać kolejne nazwy miejscowości:

Zagórze
Zacisze
Dobrzyków
Dzwonów Dolny
Dzwonów Górny
...

Zwróćmy uwagę, że metoda ZipFile.open() zwraca objekt klasy ZipExtFile,
która dziedziczy po io.BufferedIOBase.
O ile parser XML nie miał problemu z obsługą tego typu danych,
to w przypadku CSV sprawa się komplikuje:

import csv

with zf.open(zf.namelist()[1]) as csv_file:
    csv_reader = csv.reader(csv_file, delimiter=";")
    for row in csv_reader:
        print(row)

Przy wykonywaniu pętli rzuciło wyjątkiem:

_csv.Error: iterator should return strings, not bytes (did you open the file in text mode?)

Dzieje się tak, ponieważ obiekty klasy io.BufferedIOBase,
a co za tym idzie również ZipExtFile,
reprezentują strumienie binarne, a nie tekstowe.

W komunikacie jest podpowiedź, aby plik otworzyć w trybie tekstowym.
Jak czytamy w opisie metody ZipFile.open() w dokumentacji Pythona:

Use io.TextIOWrapper for reading compressed text files in universal newlines mode.

Nic, tylko zastosować:

import csv
import io

with zf.open(zf.namelist()[1]) as csv_file:
    text_file = io.TextIOWrapper(csv_file)
    csv_reader = csv.reader(text_file, delimiter=";")
    for row in csv_reader:
        print(row)

I już możemy cieszyć się widokiem kolejnych rekordów:

['\ufeffWOJ', 'POW', 'GMI', 'RODZ_GMI', 'RM', 'MZ', 'NAZWA', 'SYM', 'SYMPOD', 'STAN_NA']
['02', '16', '01', '5', '03', '1', 'Zagórze', '0363122', '0363100', '2018-01-02']
['02', '16', '01', '5', '03', '1', 'Zacisze', '0363168', '0363145', '2018-01-02']
['02', '16', '01', '5', '03', '1', 'Dobrzyków', '0363263', '0363257', '2018-01-02']
['02', '09', '02', '2', '03', '1', 'Dzwonów Dolny', '0363330', '0363323', '2018-01-02']
['02', '09', '02', '2', '03', '1', 'Dzwonów Górny', '0363346', '0363323', '2018-01-02']

Po skończonej pracy nie zapomnijmy zamknąć archiwum:

zf.close()

Niby taka błaha sprawa jak odpytanie API,
a ile się można nowych rzeczy nauczyć 🙂.