mirror of
https://github.com/ARUP-CAS/aiscr-qgis-amcr-viewer.git
synced 2026-08-12 10:08:16 +02:00
334 lines
11 KiB
Python
334 lines
11 KiB
Python
# -*- 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()
|