mpeltriaux
b0a15d9d1c
* adds command sync_codelist
* provides updating of all codes to the newest version (id)
* must be run once on staging, can be dropped afterwards since the root for the problem has been resolved on the codelist management application
108 lines
4.1 KiB
Python
108 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 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
|
|
) |