aiscr-qgis-amcr-viewer/amcr_viewer/amcr_codelists.py

# -*- 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', {}))
    NALEZ_KATEGORIE.clear()
    NALEZ_KATEGORIE.update(data.get('nalez_kategorie', {}))
    DRUH_NALEZU.clear()
    DRUH_NALEZU.update(data.get('druh_nalezu', {}))
    SPECIFIKACE.clear()
    SPECIFIKACE.update(data.get('specifikace', {}))
    NALEZOVE_OKOLNOSTI.clear()
    NALEZOVE_OKOLNOSTI.update(data.get('nalezove_okolnosti', {}))
    NALEZCE.clear()
    NALEZCE.update(data.get('nalezce', {}))


# 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 = {}
NALEZ_KATEGORIE = {}
DRUH_NALEZU = {}
SPECIFIKACE = {}
NALEZOVE_OKOLNOSTI = {}
NALEZCE = {}

refresh_globals()