#!/usr/bin/env python3 namespaces = { 'gml': 'http://www.opengis.net/gml/3.2', 'xlink': 'http://www.w3.org/1999/xlink', 'xsi': 'http://www.w3.org/2001/XMLSchema-instance', 'ami': 'urn:cz:isvs:ruian:schemas:AdrMistoIntTypy:v1', 'base': 'urn:cz:isvs:ruian:schemas:BaseTypy:v1', 'coi': 'urn:cz:isvs:ruian:schemas:CastObceIntTypy:v1', 'com': 'urn:cz:isvs:ruian:schemas:CommonTypy:v1', 'kui': 'urn:cz:isvs:ruian:schemas:KatUzIntTypy:v1', 'kri': 'urn:cz:isvs:ruian:schemas:KrajIntTypy:v1', 'mci': 'urn:cz:isvs:ruian:schemas:MomcIntTypy:v1', 'mpi': 'urn:cz:isvs:ruian:schemas:MopIntTypy:v1', 'obi': 'urn:cz:isvs:ruian:schemas:ObecIntTypy:v1', 'oki': 'urn:cz:isvs:ruian:schemas:OkresIntTypy:v1', 'opi': 'urn:cz:isvs:ruian:schemas:OrpIntTypy:v1', 'pai': 'urn:cz:isvs:ruian:schemas:ParcelaIntTypy:v1', 'pui': 'urn:cz:isvs:ruian:schemas:PouIntTypy:v1', 'rsi': 'urn:cz:isvs:ruian:schemas:RegSouIntiTypy:v1', 'spi': 'urn:cz:isvs:ruian:schemas:SpravObvIntTypy:v1', 'sti': 'urn:cz:isvs:ruian:schemas:StatIntTypy:v1', 'soi': 'urn:cz:isvs:ruian:schemas:StavObjIntTypy:v1', 'uli': 'urn:cz:isvs:ruian:schemas:UliceIntTypy:v1', 'vci': 'urn:cz:isvs:ruian:schemas:VuscIntTypy:v1', 'vf': 'urn:cz:isvs:ruian:schemas:VymennyFormatTypy:v1', 'zji': 'urn:cz:isvs:ruian:schemas:ZsjIntTypy:v1', 'voi': 'urn:cz:isvs:ruian:schemas:VOIntTypy:v1', } """ Novou zakladni URL ziskej na https://vdp.cuzk.cz/vdp/ruian/vymennyformat/vyhledej Ve vyhledávání zvolte: - Platnost údajů: Platné - Časový rozsah: Úplná kopie - Územní prvky: Obec a podřazené - Datová sada: Základní - Výběr z údajů: např. Základní údaje - Územní omezení: ČR Seznam obci uloz do souboru s nazvem `id_obce`; jeden radek = jedno ID obce. - Pouzij ciselniky CSU, napr. 554782 = Praha. Vysledkem skriptu je tab-separated soubor `ruian-byty.tsv` se sloupci: - id_obce (kod obce z ciselniku CSU) - ruian (RUIAN ID adresy) - stavebni_objekt (jeden stavebni objekt muze mit vic RUIAN ID) - byty_dum (celkovy pocet bytu v jednom stavebnim objektu) - byty_vchod (pocet bytu v jednom vchode/RUIAN ID; nekdy v datech chybi) - zpusob_vyuziti """ BASE_URL = 'https://vdp.cuzk.cz/vymenny_format/soucasna/20220331_OB_{}_UZSZ.xml.zip' OUTFILE = 'ruian-byty.tsv' import gzip import sys import os import io import zipfile from urllib.request import urlopen from io import StringIO from xml.dom.minidom import parse, parseString from xml.etree.ElementTree import ElementTree from pprint import pprint def parse_tea(id_objektu, vchod, vchody): e_byty = vchod.findall('./soi:PocetBytu', namespaces) if e_byty: byty = int(e_byty[0].text) else: # jsou stavebni vchody, ktere v ruian nemaji pocty bytu, napr. 22055843 v Praze # tyto vchody ignorujeme byty = -1 e_ruian = vchod.findall('./soi:AdresniMistoKod/base:Kod', namespaces) # objekt muze mit vic kodu RUIAN, takze ulozme vsechny if e_ruian: for e_ruian_item in e_ruian: ruian = int(e_ruian_item.text) if id_objektu not in vchody: vchody[id_objektu] = {} vchody[id_objektu][ruian] = byty else: print('ruian NENALEZEN') pprint(id_objektu) sys.exit(2) def parse_so(e, vchody, domy): e_id_objektu = e.findall('./soi:Kod', namespaces) if not e_id_objektu: return id_objektu = int(e_id_objektu[0].text) e_vchody = e.findall('./soi:DetailniTEA/soi:DetailniTEA', namespaces) if e_vchody: for vchod in e_vchody: parse_tea(id_objektu, vchod, vchody) if id_objektu not in domy: domy[id_objektu] = {} byty = e.findall('./soi:PocetBytu', namespaces) if byty: if len(byty) > 1: print('vice nez jeden zaznam o poctu bytu bez TEA') sys.exit(1) else: domy[id_objektu]['byty'] = int(byty[0].text) else: domy[id_objektu]['byty'] = -1 vyuziti = e.findall('./soi:ZpusobVyuzitiKod', namespaces) domy[id_objektu]['vyuziti'] = int(vyuziti[0].text) if vyuziti else -1 def parse_am(e, adresnimista): e_id_objektu = e.findall('./ami:Kod', namespaces) if not e_id_objektu: return id_objektu = int(e_id_objektu[0].text) e_so = e.findall('./ami:StavebniObjekt/soi:Kod', namespaces) if e_so: so_id = int(e_so[0].text) else: so_id = -1 adresnimista[id_objektu] = so_id def parse_file(file_name): print('Opening file...') mydoc = ElementTree(file=file_name) vchody = {} domy = {} adresnimista = {} print('Parsing...') for e in mydoc.findall('./vf:Data/vf:StavebniObjekty/vf:StavebniObjekt', namespaces): parse_so(e, vchody, domy) for e in mydoc.findall('./vf:Data/vf:AdresniMista/vf:AdresniMisto', namespaces): parse_am(e, adresnimista) print('Parsing complete.') return (vchody, domy, adresnimista) def pocet_bytu(ruian, so_id, vchody, domy): if so_id in vchody: byty_vchod = vchody[so_id][ruian] if ruian in vchody[so_id] else -2 else: byty_vchod = -3 if so_id in domy: vyuziti = domy[so_id]['vyuziti'] if 'vyuziti' in domy[so_id] else -2 byty_dum = domy[so_id]['byty'] if 'byty' in domy[so_id] else -2 else: vyuziti = -3 byty_dum = -3 return byty_dum, byty_vchod, vyuziti def byty(id_obce, adresnimista, vchody, domy, outfile): byty = {} print('Writing...') for misto in adresnimista: ruian = misto so_id = adresnimista[ruian] byty_dum, byty_vchod, vyuziti = pocet_bytu(ruian, so_id, vchody, domy) outfile.write("{}\t{}\t{}\t{}\t{}\t{}\n".format(id_obce, ruian, so_id, byty_dum, byty_vchod, vyuziti)) print('Writing complete.') def download_gz(id_obce, outfile_path): url = BASE_URL.format(id_obce) print('Downloading {}...'.format(url)) response = urlopen(url) compressedFile = StringIO.StringIO() compressedFile.write(response.read()) # # Set the file's current position to the beginning # of the file so that gzip.GzipFile can read # its contents from the top. # compressedFile.seek(0) decompressedFile = gzip.GzipFile(fileobj=compressedFile, mode='rb') with open(outfile_path, 'w') as outfile: outfile.write(decompressedFile.read()) print('Download complete.') def download_zip(id_obce, outfile_path): url = BASE_URL.format(id_obce) print('Downloading {}...'.format(url)) response = urlopen(url) compressedData = io.BytesIO(response.read()) zipdata = zipfile.ZipFile(compressedData) decompressed_file_name = [name for name in zipdata.namelist()][0] with open(outfile_path, 'wb') as outfile: outfile.write(zipdata.read(decompressed_file_name)) print('Download complete.') def xml_to_bytlist(id_obce, outfile_path): tempfile = '/tmp/parse.xml' outfile = open(outfile_path, 'a') download_zip(id_obce, tempfile) (vchody, domy, adresnimista) = parse_file(tempfile) byty(id_obce, adresnimista, vchody, domy, outfile) os.unlink(tempfile) outfile.close() print('Job complete.') def load_obce(): # return [500496] # Olomouc # return [554782] # Praha seznam_obci = [] with open('id_obce', 'r') as obcefile: for line in obcefile: seznam_obci.append(int(line)) return seznam_obci def main(): print('Start processing...') if os.path.isfile(OUTFILE): os.unlink(OUTFILE) # write header with open(OUTFILE, 'a') as outfile: outfile.write('id_obce\truian\tstavebni_objekt\tbyty_dum\tbyty_vchod\tzpusob_vyuziti\n') # write data for obce in load_obce(): xml_to_bytlist(obce, OUTFILE) print('Processing complete.') main()