mirror of
https://github.com/ARUP-CAS/aiscr-qgis-amcr-viewer.git
synced 2026-10-10 12:57:36 +02:00
david-spacil
d398d2cd1b
Oficiální kontrola pyqgis4-checker (skript pyqt5_to_pyqt6.py z QGISu)
hlásila u pluginu 13 nekompatibilit a tři soubory vůbec nepřečetla.
Odstraněn UTF-8 BOM z amcr_codelists.py, amcr_dialog.py a amcr_tools.py.
Checker čte zdroják jako UTF-8 bez utf-8-sig a na BOM padá na
SyntaxError: invalid non-printable character U+FEFF, takže se ty soubory
nezkontrolovaly vůbec. Pythonu při běhu BOM nevadí, proto to nikdy
nevyskočilo. heslar.csv si BOM ponechává, tam je kvůli Excelu.
Enumy převedeny na plně kvalifikované tvary (13 míst):
Qgis.{Info,Warning,Critical} -> Qgis.MessageLevel.*,
QgsTask.CanCancel -> QgsTask.Flag.CanCancel,
QgsWkbTypes.*Geometry -> QgsWkbTypes.GeometryType.*Geometry.
Zkrácené tvary v QGIS 4.2.1 zatím fungují, ale kontrola je vytýká
a aliasy do budoucna mizí. Zbytek kódu už scoped tvary používal.
Do AGENTS.md doplněna sekce o kompatibilitě s Qt6 / QGIS 4 se závaznými
pravidly a příkazy na ověření.
Ověřeno spuštěním v QGIS 3.44.13 (Qt 5.15.18) i QGIS 4.2.1 (Qt 6.10.3):
importy, vytvoření tasku i všechny tři filtrační dialogy fungují shodně.
Log pyqgis4-checkeru je nově prázdný.
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.MessageLevel.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.MessageLevel.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.MessageLevel.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()
|