You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
konova/codelist/management/commands/update_codelist.py

106 lines
4.1 KiB
Python

"""
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_COMPENSATION_HANDLER_ID, \
CODELIST_COMPENSATION_ACTION_ID, CODELIST_COMPENSATION_ACTION_CLASS_ID, CODELIST_COMPENSATION_ADDITIONAL_TYPE_ID, \
CODELIST_COMPENSATION_FUNDING_ID, CODELIST_BASE_URL, CODELIST_PROCESS_TYPE_ID
from konova.management.commands.setup import BaseKonovaCommand
bool_map = {
"true": True,
"false": False,
}
class Command(BaseKonovaCommand):
help = "Performs test on collisions using the identifier generation"
def handle(self, *args, **options):
try:
codelist_ids = [
CODELIST_INTERVENTION_HANDLER_ID,
CODELIST_CONSERVATION_OFFICE_ID,
CODELIST_REGISTRATION_OFFICE_ID,
CODELIST_BIOTOPES_ID,
CODELIST_LAW_ID,
CODELIST_COMPENSATION_HANDLER_ID,
CODELIST_COMPENSATION_ACTION_ID,
CODELIST_COMPENSATION_ACTION_CLASS_ID,
CODELIST_COMPENSATION_ADDITIONAL_TYPE_ID,
CODELIST_COMPENSATION_FUNDING_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)
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
)