""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 23.08.21 """ import requests from django.core.management import BaseCommand from xml.etree import ElementTree as etree from codelist.models import KonovaCode, KonovaCodeList from codelist.settings import CODELIST_INTERVENTION_HANDLER_ID, CODELIST_CONSERVATION_OFFICE_ID, \ CODELIST_REGISTRATION_OFFICE_ID, CODELIST_BIOTOPES_ID, CODELIST_LAW_ID, CODELIST_HANDLER_ID, \ CODELIST_COMPENSATION_ACTION_ID, CODELIST_COMPENSATION_ACTION_CLASS_ID, CODELIST_COMPENSATION_ADDITIONAL_TYPE_ID, \ CODELIST_BASE_URL, CODELIST_PROCESS_TYPE_ID, CODELIST_BIOTOPES_EXTRA_CODES_ID, \ CODELIST_COMPENSATION_ACTION_DETAIL_ID from konova.management.commands.setup import BaseKonovaCommand from konova.settings import PROXIES bool_map = { "true": True, "false": False, } class Command(BaseKonovaCommand): help = "Updates internal codelist by external API" def handle(self, *args, **options): try: codelist_ids = [ CODELIST_INTERVENTION_HANDLER_ID, CODELIST_CONSERVATION_OFFICE_ID, CODELIST_REGISTRATION_OFFICE_ID, CODELIST_BIOTOPES_ID, CODELIST_BIOTOPES_EXTRA_CODES_ID, CODELIST_LAW_ID, CODELIST_HANDLER_ID, CODELIST_COMPENSATION_ACTION_ID, CODELIST_COMPENSATION_ACTION_CLASS_ID, CODELIST_COMPENSATION_ACTION_DETAIL_ID, CODELIST_COMPENSATION_ADDITIONAL_TYPE_ID, CODELIST_PROCESS_TYPE_ID, ] self._write_warning("Fetching codes...") for list_id in codelist_ids: _url = CODELIST_BASE_URL + "/" + str(list_id) result = requests.get(url=_url, proxies=PROXIES) if result.status_code != 200: self._write_error("Error on codelist url '{}'".format(_url)) continue self._write_warning("Fetch codes for list id {}".format(list_id)) content = result.content.decode("utf-8") root = etree.fromstring(content) items = root.findall("item") self._write_warning("Found {} codes. Process now...".format(len(items))) code_list = KonovaCodeList.objects.get_or_create( id=list_id, )[0] self._rec_xml_code_fetch( items=items, code_list=code_list, parent=None, ) except KeyboardInterrupt: self._break_line() exit(-1) def _rec_xml_code_fetch(self, items: list, code_list: KonovaCodeList, parent: KonovaCode): if items is None: return else: self._write_warning(" --- Found {} subcodes. Process now...".format(len(items))) for element in items: children = element.find("items") _id = element.find("id").text atom_id = element.find("atomid").text selectable = element.find("selectable").text.lower() selectable = bool_map.get(selectable, False) short_name = element.find("shortname").text long_name = element.find("longname").text is_archived = bool_map.get((element.find("archive").text.lower()), False) code = KonovaCode.objects.get_or_create( id=_id, ) code = code[0] code.atom_id = atom_id code.short_name = short_name code.long_name = long_name code.parent = parent code.is_selectable = selectable code.is_archived = is_archived code.is_leaf = children is None code.save() if code not in code_list.codes.all(): code_list.codes.add(code) self._rec_xml_code_fetch( items=children, code_list=code_list, parent=code )