# -*- coding: utf-8 -*- import os import csv import requests import xml.etree.ElementTree as ET # nosec import time from qgis.core import QgsMessageLog, Qgis # Define paths for the plugin and its codelists directory PLUGIN_DIR = os.path.dirname(__file__) CODELISTS_DIR = os.path.join(PLUGIN_DIR, 'codelists') BASE_URL_AMCR = "https://api.aiscr.cz/2.2/oai" BASE_URL_DA = "https://digiarchiv.aiscr.cz/api/search/query" OUTPUT_FILE = os.path.join(CODELISTS_DIR, 'heslar.csv') slovnicek = { 'obdobi': (BASE_URL_AMCR, 'heslo:obdobi'), 'typ_akce': (BASE_URL_AMCR, 'heslo:akce_typ'), 'areal': (BASE_URL_AMCR, 'heslo:areal'), 'kraj': (BASE_URL_AMCR, 'ruian_kraj'), 'organizace': (BASE_URL_AMCR, 'organizace'), 'okres': (BASE_URL_AMCR, 'ruian_okres'), 'katastr': (BASE_URL_AMCR, 'ruian_katastr'), 'pian_presnost': (BASE_URL_AMCR, 'heslo:pian_presnost'), 'typ_lokality': (BASE_URL_AMCR, 'heslo:lokalita_typ'), 'druh_lokality': (BASE_URL_AMCR, 'heslo:lokalita_druh'), 'jistota': (BASE_URL_AMCR, 'heslo:jistota_urceni'), 'lokalita_zachovalost': (BASE_URL_AMCR, 'heslo:stav_dochovani'), 'pristupnost': (BASE_URL_AMCR, 'heslo:pristupnost'), 'nalez_kategorie': (BASE_URL_AMCR, 'heslo:predmet_druh_kat'), 'druh_nalezu': (BASE_URL_AMCR, 'heslo:predmet_druh'), 'specifikace': (BASE_URL_AMCR, 'heslo:predmet_specifikace'), 'nalezove_okolnosti': (BASE_URL_AMCR, 'heslo:nalezove_okolnosti'), 'vedouci': (BASE_URL_DA, 'f_vedouci'), 'nalezce': (BASE_URL_DA, 'f_nalezce'), } NS = { 'oai': 'http://www.openarchives.org/OAI/2.0/', 'dc': 'http://purl.org/dc/elements/1.1/', 'oai_dc': 'http://www.openarchives.org/OAI/2.0/oai_dc/' } def ensure_codelists_dir(): """Creates the codelists directory if it does not exist.""" if not os.path.exists(CODELISTS_DIR): os.makedirs(CODELISTS_DIR) def parse_codelist_file(filename, target_dict=None): """ Reads a CSV codelist file and populates the target dictionary grouped by categories. """ if target_dict is None: target_dict = {} path = os.path.join(CODELISTS_DIR, filename) # Return early if the file doesn't exist to avoid missing file errors if not os.path.exists(path): return target_dict try: # Open the file using standard UTF-8 encoding with open(path, 'r', encoding='utf-8') as f: reader = csv.reader(f, delimiter=';') # Skip the CSV header row next(reader, None) # Iterate through rows and extract label, code, and category for row in reader: if len(row) >= 3: label = row[0].strip() code = row[1].strip() cat = row[2].strip() clean = code if code else None # Initialize a new dictionary for a category if encountered # for the first time if cat not in target_dict: target_dict[cat] = {} # Assign the extracted code to the corresponding label # within the category target_dict[cat][label] = clean except Exception as e: QgsMessageLog.logMessage( f"AMČR Codelist Read Error for {filename}: {e}", "AMČR", Qgis.Critical) return target_dict def load_all_data(): """Loads the codelist during plugin startup.""" ensure_codelists_dir() categorized_data = {k: {} for k in slovnicek.keys()} parse_codelist_file('heslar.csv', categorized_data) return categorized_data def fetch_set(base_url, internal_name, api_set, task=None): dataset = [] params_amcr = { "verb": "ListRecords", "metadataPrefix": "oai_dc", "set": api_set } params_da = { "entity": "samostatny_nalez" if internal_name == "nalezce" else "akce", "rows": 0, "noFacets": "false", "onlyFacets": "true" } while True: # Check for cancellation at each iteration if task and task.isCanceled(): return None try: if "digiarchiv" not in base_url: response = requests.get(base_url, params=params_amcr, timeout=30) response.raise_for_status() root = ET.fromstring(response.content) # nosec records = root.findall('.//oai:record', NS) for rec in records: metadata = rec.find('.//oai_dc:dc', NS) if metadata is not None: # Code (identifier) identifier_el = metadata.find('dc:identifier', NS) kod = ( identifier_el.text if identifier_el is not None else "" ) # Title – filter out system labels "AMČR - ..." titles = metadata.findall('dc:title', NS) nazev = "" for t in titles: if ( t.text and not t.text.startswith("AMČR -") and not t.text.startswith(" AMČR -") ): nazev = t.text break # If no title passed the filter, fall back # to the first available one if not nazev and titles: nazev = titles[0].text specialni_pripady = ['okres', 'katastr'] if internal_name in specialni_pripady: kod = nazev if internal_name == 'pristupnost': kod = next( ( t.text for t in titles if t.text and len(t.text) == 1 and t.text.isalpha() ), None ) # Skip records without a valid one-letter code – # a None code would end up in the CSV and later # in the API filter as the string "None" if not kod: continue dataset.append({ 'Název': nazev, 'Kód': kod, 'Kategorie': internal_name }) # Pagination token = root.find('.//oai:resumptionToken', NS) if token is not None and token.text: params_amcr = { "verb": "ListRecords", "resumptionToken": token.text } time.sleep(0.5) else: break else: response = requests.get(base_url, params=params_da, timeout=30) response.raise_for_status() data_json = response.json() records = data_json['facet_counts']['facet_fields'][api_set] for r in records: nazev = r["name"] dataset.append({ 'Název': nazev, 'Kód': nazev, 'Kategorie': internal_name }) break except Exception as e: QgsMessageLog.logMessage( f"Chyba u setu {api_set}: {e}", "AMČR", Qgis.Warning) break return dataset def download_heslare(task=None): """Fetches the codelists from the AMČR API and saves it to a CSV file.""" ensure_codelists_dir() all_data = [] total_sets = len(slovnicek) # index, (interni, api_nazev) for index, (key, value) in enumerate(slovnicek.items()): base_url = value[0] interni = key api_nazev = value[1] # Check if the user cancelled the task via the QGIS taskbar if task and task.isCanceled(): return False QgsMessageLog.logMessage( f"Zpracovávám kategorii: {interni}...", "AMČR", Qgis.Info) # Pass the task correctly to the updated fetch function data = fetch_set(base_url, interni, api_nazev, task=task) if data is None: return False # Cancelled mid-download all_data.extend(data) # Report progress (0-100) if task: progress = (index + 1) / total_sets * 100 task.setProgress(progress) # Save to CSV with open(OUTPUT_FILE, 'w', newline='', encoding='utf-8-sig') as f: fieldnames = ['Název', 'Kód', 'Kategorie'] writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=';') writer.writeheader() writer.writerows(all_data) return True def refresh_globals(): """Reloads data from files into the global variables.""" data = load_all_data() OBDOBI.clear() OBDOBI.update(data.get('obdobi', {})) TYP_AKCE.clear() TYP_AKCE.update(data.get('typ_akce', {})) AREAL.clear() AREAL.update(data.get('areal', {})) KRAJE.clear() KRAJE.update(data.get('kraj', {})) ORGANIZACE.clear() ORGANIZACE.update(data.get('organizace', {})) OKRESY.clear() OKRESY.update(data.get('okres', {})) KATASTRY.clear() KATASTRY.update(data.get('katastr', {})) VEDOUCI.clear() VEDOUCI.update(data.get('vedouci', {})) PIAN_PRESNOST.clear() PIAN_PRESNOST.update(data.get('pian_presnost', {})) TYP_LOKALITY.clear() TYP_LOKALITY.update(data.get('typ_lokality', {})) DRUH_LOKALITY.clear() DRUH_LOKALITY.update(data.get('druh_lokality', {})) JISTOTA.clear() JISTOTA.update(data.get('jistota', {})) LOKALITA_ZACHOVALOST.clear() LOKALITA_ZACHOVALOST.update(data.get('lokalita_zachovalost', {})) PRISTUPNOST.clear() PRISTUPNOST.update(data.get('pristupnost', {})) # Initialize empty dicts that will be populated immediately below OBDOBI = {} TYP_AKCE = {} AREAL = {} KRAJE = {} ORGANIZACE = {} OKRESY = {} KATASTRY = {} VEDOUCI = {} PIAN_PRESNOST = {} TYP_LOKALITY = {} DRUH_LOKALITY = {} JISTOTA = {} LOKALITA_ZACHOVALOST = {} PRISTUPNOST = {} refresh_globals()