From 64d7a4e2e8c08bf4c8ac5befe3f690dbe5a6d728 Mon Sep 17 00:00:00 2001 From: Bernhard Koschicek-Krombholz Date: Wed, 7 Oct 2026 15:26:33 +0200 Subject: [PATCH] refactor get_info() functions of external reference systems --- openatlas/api/external/cadaster.py | 77 ++++++++--------- openatlas/api/external/chronontology.py | 77 +++++++++-------- openatlas/api/external/doi.py | 84 +++++++++---------- openatlas/api/external/gnd.py | 41 +++++---- openatlas/api/external/openatlas_api.py | 43 ++++++---- openatlas/api/external/viaf.py | 60 +++++++------- openatlas/api/external/wikidata.py | 105 +++++++++--------------- 7 files changed, 243 insertions(+), 244 deletions(-) diff --git a/openatlas/api/external/cadaster.py b/openatlas/api/external/cadaster.py index e72567190..96e9a1366 100644 --- a/openatlas/api/external/cadaster.py +++ b/openatlas/api/external/cadaster.py @@ -12,7 +12,6 @@ class Cadaster(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: endpoint = 'gst/' if '/' in id_ else 'kgnr/' - info = {} try: data = requests.get( f'https://kataster.bev.gv.at/api/{endpoint}{id_}', @@ -21,42 +20,44 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: timeout=10).json() except Exception: # pragma: no cover return {} - try: - info['katestralgemeinde'] = data['properties']['kg'] - except KeyError: # pragma: no cover - pass - try: - info['bundesland'] = data['properties']['bl'] - except KeyError: # pragma: no cover - pass - try: - info['grundstücksnummer'] = data['properties']['gnr'] - except KeyError: # pragma: no cover - pass - try: - info['einlagezahl'] = data['properties']['ez'] - except KeyError: # pragma: no cover - pass - try: - info['rstatus'] = data['properties']['rstatus'] - except KeyError: # pragma: no cover - pass - try: + + info: dict[str, object] = {} + + if error_msg := data.get('message'): + info['Error'] = error_msg + return info + + properties = data.get('properties', {}) + + if kg := properties.get('kg'): + info['katestralgemeinde'] = kg + + if bl := properties.get('bl'): + info['bundesland'] = bl # pragma: no cover + + if gnr := properties.get('gnr'): + info['grundstücksnummer'] = gnr + + if ez := properties.get('ez'): + info['einlagezahl'] = ez + + if rstatus := properties.get('rstatus'): + info['rstatus'] = rstatus + + if nutzungen := properties.get('nutzungen'): info['nutzungen'] = [ - f'Allocation: {usage['nutzung']}, Area: {usage['fl']}m²' - for usage in data['properties']['nutzungen']] - except KeyError: # pragma: no cover - pass - try: - geometry = shape(data['geometry']).centroid - zoom = '18.1' if '/' in id_ else '12.9' - url = f'https://kataster.bev.gv.at/#/center/' \ - f'{geometry.x},{geometry.y}/zoom/{zoom}/vermv/0.6' - info['geometry'] = link(url, url, external=True) - except KeyError: # pragma: no cover - pass - try: - info['Error'] = data['message'] - except KeyError: # pragma: no cover - pass + f'Allocation: {usage.get("nutzung", "")}, ' \ + f'Area: {usage.get("fl", "")}m²' + for usage in nutzungen] + + if geometry_data := data.get('geometry'): + try: + geom = shape(geometry_data).centroid + zoom = '18.1' if '/' in id_ else '12.9' + url = f'https://kataster.bev.gv.at/#/center/' \ + f'{geom.x},{geom.y}/zoom/{zoom}/vermv/0.6' + info['geometry'] = link(url, url, external=True) + except Exception: # pragma: no cover + pass + return info diff --git a/openatlas/api/external/chronontology.py b/openatlas/api/external/chronontology.py index 00e52fd14..0925fe44c 100644 --- a/openatlas/api/external/chronontology.py +++ b/openatlas/api/external/chronontology.py @@ -14,8 +14,6 @@ class ChronOntology(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - info: dict[str, object] = {} - try: response = requests.get( f'https://chronontology.dainst.org/data/period/{id_}', @@ -27,28 +25,30 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: response.raise_for_status() data = response.json() except Exception: # pragma: no cover - return info + return {} + info: dict[str, object] = {} resource = data.get('resource', {}) names = resource.get('names', {}) - title = None - if 'en' in names and names['en']: - title = names['en'][0] - elif 'de' in names and names['de']: # pragma: no cover - title = names['de'][0] + if en_names := names.get('en'): + info['title'] = en_names[0] + elif de_names := names.get('de'): # pragma: no cover + info['title'] = de_names[0] + else: + info['title'] = f"ChronOntology Period {id_}" # pragma: no cover - info['title'] = title or f"ChronOntology Period {id_}" + if definition := resource.get('definition'): + info['definition'] = definition - info['definition'] = resource.get('definition') + if (timespans := resource.get('hasTimespan')) \ + and isinstance(timespans, list): + if time_orig := timespans[0].get('timeOriginal'): + info['timespan'] = time_orig - timespans = resource.get('hasTimespan', []) - if timespans and isinstance(timespans, list): - info['timespan'] = timespans[0]['timeOriginal'] - - types = resource.get('types', []) - if types: - info['period types'] = ', '.join(types) + if types := resource.get('types'): + info['period types'] = ', '.join(types) \ + if isinstance(types,list) else str(types) if gazetteer := ChronOntology.get_gazetteer_links(data.get('related')): info['gazetteer'] = gazetteer @@ -57,40 +57,39 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: @staticmethod def get_gazetteer_links(related: Any) -> list[str]: - links: list[str] = [] if not isinstance(related, dict): - return links # pragma: no cover + return [] # pragma: no cover + links: list[str] = [] for key, place in related.items(): - if 'gazetteer.dainst.org/place/' not in str(key): - continue # pragma: no cover - if not isinstance(place, dict): + if 'gazetteer.dainst.org/place/' not in str(key) \ + or not isinstance(place, dict): continue # pragma: no cover english_name = '' + for name in place.get('names', []): - if not isinstance(name, dict): - continue # pragma: no cover - if name.get('language') == 'eng' and name.get('title'): - english_name = str(name['title']) + if isinstance(name, dict) \ + and name.get('language') == 'eng' \ + and (title := name.get('title')): + english_name = str(title) break - if not english_name \ - and isinstance(place.get('prefName'), dict) \ - and place['prefName'].get('language') == 'eng' \ - and place['prefName'].get('title'): - english_name = str(place['prefName']['title']) + if not english_name: + pref = place.get('prefName', {}) + if isinstance(pref, dict) \ + and pref.get('language') == 'eng' \ + and (title := pref.get('title')): + english_name = str(title) if not english_name: continue # pragma: no cover place_id = str(key).rstrip('/').rsplit('/', maxsplit=1)[-1] - if not place_id: - continue # pragma: no cover - - links.append(link( - english_name, - f'https://gazetteer.dainst.org/place/{place_id}', - external=True)) + if place_id: + links.append(link( + english_name, + f'https://gazetteer.dainst.org/place/{place_id}', + external=True)) - return links + return links \ No newline at end of file diff --git a/openatlas/api/external/doi.py b/openatlas/api/external/doi.py index 2655b72a7..d31225f10 100644 --- a/openatlas/api/external/doi.py +++ b/openatlas/api/external/doi.py @@ -12,7 +12,6 @@ class DOI(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - info: dict[str, object] = {} try: data = requests.get( f'https://doi.org/{id_}', @@ -22,59 +21,56 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: proxies=app.config['PROXIES'], timeout=10).json() except Exception: # pragma: no cover - return info + return {} + + info: dict[str, object] = {} - if 'title' in data and data['title']: - info['title'] = data['title'] + if title := data.get('title'): + info['title'] = title - if 'author' in data: + if raw_authors := data.get('author', []): authors = [] - for author in data['author']: + for author in raw_authors: name = [] - if 'family' in author: - name.append(author['family']) - if 'given' in author: - name.append(author['given']) - elif 'literal' in author: # pragma: no cover - name.append(author['literal']) - authors.append(', '.join(name)) - info['authors'] = '; '.join(authors) - - if 'container-title' in data and data['container-title']: - info['container'] = data['container-title'] - - if 'publisher' in data: - info['publisher'] = data['publisher'] - - year = None + if family := author.get('family'): + name.append(family) + if given := author.get('given'): + name.append(given) + elif literal := author.get('literal'): # pragma: no cover + name.append(literal) + if name: + authors.append(', '.join(name)) + if authors: + info['authors'] = '; '.join(authors) + + if container := data.get('container-title'): + info['container'] = container + + if publisher := data.get('publisher'): + info['publisher'] = publisher + for date_field in [ - 'issued', 'published-print', 'published-online', 'published']: - if date_field in data and 'date-parts' in data[date_field]: - try: - year = data[date_field]['date-parts'][0][0] - if year: + 'issued', 'published-print', 'published-online', 'published']: + if date_obj := data.get(date_field): + parts = date_obj.get('date-parts', []) + if parts and isinstance(parts[0], list) and parts[0]: + if year := parts[0][0]: + info['year'] = year break - except (IndexError, TypeError): # pragma: no cover - continue - if year: - info['year'] = year - if 'DOI' in data: - info['DOI'] = link( - data['DOI'], - f'https://doi.org/{data["DOI"]}', - external=True) + if doi := data.get('DOI'): + info['DOI'] = link(doi, f'https://doi.org/{doi}', external=True) - if 'type' in data: - info['type'] = data['type'].replace('-', ' ').title() + if doc_type := data.get('type'): + info['type'] = doc_type.replace('-', ' ').title() - if 'page' in data: - info['page'] = data['page'] + if page := data.get('page'): + info['page'] = page - if 'volume' in data: - info['volume'] = data['volume'] + if volume := data.get('volume'): + info['volume'] = volume - if 'issue' in data: - info['issue'] = data['issue'] + if issue := data.get('issue'): + info['issue'] = issue return info diff --git a/openatlas/api/external/gnd.py b/openatlas/api/external/gnd.py index d1049f27c..c250bcf3e 100644 --- a/openatlas/api/external/gnd.py +++ b/openatlas/api/external/gnd.py @@ -12,31 +12,40 @@ class GND(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - info: dict[str, object] = {} try: data = requests.get( f'{system.resolver_url}{id_}.json', proxies=app.config['PROXIES'], timeout=10).json() except Exception: # pragma: no cover - return info + return {} def print_values(values: list[dict[str, str]]) -> str: return '
'.join( [link(i['label'], i['id'], external=True) for i in values]) - if 'preferredName' in data: - info['preferred name'] = data['preferredName'] - if 'gender' in data: - info['gender'] = print_values(data['gender']) - if 'dateOfBirth' in data: - info['date of birth'] = data['dateOfBirth'] - if 'placeOfBirth' in data: - info['place of birth'] = print_values(data['placeOfBirth']) - if 'dateOfDeath' in data: - info['date of death'] = data['dateOfDeath'] - if 'placeOfDeath' in data: - info['place of death'] = print_values(data['placeOfDeath']) - if 'type' in data: - info['type'] = '
'.join(data['type']) + info: dict[str, object] = {} + + if pref_name := data.get('preferredName'): + info['preferred name'] = pref_name + + if gender := data.get('gender'): + info['gender'] = print_values(gender) + + if dob := data.get('dateOfBirth'): + info['date of birth'] = dob + + if pob := data.get('placeOfBirth'): + info['place of birth'] = print_values(pob) + + if dod := data.get('dateOfDeath'): + info['date of death'] = dod + + if pod := data.get('placeOfDeath'): + info['place of death'] = print_values(pod) + + if types := data.get('type'): + info['type'] = '
'.join(types) \ + if isinstance(types, list) else str(types) + return info diff --git a/openatlas/api/external/openatlas_api.py b/openatlas/api/external/openatlas_api.py index e1337466a..56b4c8376 100644 --- a/openatlas/api/external/openatlas_api.py +++ b/openatlas/api/external/openatlas_api.py @@ -11,7 +11,6 @@ class OpenAtlas(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - info: dict[str, object] = {} try: api_url = f'{system.website_url}/api/entity_presentation_view/' data = requests.get( @@ -20,20 +19,36 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: proxies=app.config['PROXIES'], timeout=10).json() except Exception: # pragma: no cover - return info + return {} + + info: dict[str, object] = {} + + if title := data.get('title'): + info['name'] = title + + if sys_class := data.get('systemClass'): + info['OpenAtlas class'] = sys_class + + if aliases := data.get('aliases'): + info['aliases'] = '
'.join(aliases) # pragma: no cover - info['name'] = data.get('title') - info['OpenAtlas class'] = data.get('systemClass') - info['aliases'] = '
'.join(data.get('aliases')) for type_ in data.get('types', []): - if type_.get('isStandard'): - info['type'] = type_.get('title') - info['begin from'] = data['when']['start'].get('earliest') - info['begin to'] = data['when']['start'].get('latest') - info['begin comment'] = data['when']['start'].get('comment') - info['end from'] = data['when']['end'].get('earliest') - info['end to'] = data['when']['end'].get('latest') - info['end comment'] = data['when']['end'].get('comment') - info['description'] = data.get('description') + if type_.get('isStandard') and type_.get('title'): + info['type'] = type_['title'] + break + + if when := data.get('when', {}): + start = when.get('start', {}) + if earliest := start.get('earliest'): info['begin from'] = earliest + if latest := start.get('latest'): info['begin to'] = latest + if comment := start.get('comment'): info['begin comment'] = comment + + end = when.get('end', {}) + if earliest := end.get('earliest'): info['end from'] = earliest + if latest := end.get('latest'): info['end to'] = latest + if comment := end.get('comment'): info['end comment'] = comment + + if desc := data.get('description'): + info['description'] = desc # pragma: no cover return info diff --git a/openatlas/api/external/viaf.py b/openatlas/api/external/viaf.py index f9a59275d..e07394ce2 100644 --- a/openatlas/api/external/viaf.py +++ b/openatlas/api/external/viaf.py @@ -12,7 +12,6 @@ class VIAF(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - info: dict[str, object] = {} try: data = requests.get( f'https://viaf.org/viaf/{id_}', @@ -22,34 +21,37 @@ def get_info(id_: str, system: Entity) -> dict[str, object]: proxies=app.config['PROXIES'], timeout=10).json() except Exception: # pragma: no cover - return info - - if 'ns1:VIAFCluster' in data: - viaf_data = data['ns1:VIAFCluster'] - - if 'ns1:viafID' in viaf_data: - viaf_id = str(viaf_data['ns1:viafID']) - info['VIAF ID'] = link( - viaf_id, - f'https://viaf.org/viaf/{str(viaf_data['ns1:viafID'])}', - external=True) - - headings = viaf_data.get( - 'ns1:mainHeadings', {}).get('ns1:data', []) - if isinstance(headings, list) and headings: - info['title'] = headings[0].get('ns1:text', '') - elif isinstance(headings, dict): # pragma: no cover - info['title'] = headings.get('ns1:text', '') - - if 'ns1:nameType' in viaf_data: - info['type'] = viaf_data['ns1:nameType'] - - if 'ns1:birthDate' in viaf_data \ - and str(viaf_data['ns1:birthDate']) != '0': - info['birth date'] = str(viaf_data['ns1:birthDate']) - if 'ns1:deathDate' in viaf_data \ - and str(viaf_data['ns1:deathDate']) != '0': - info['death date'] = str(viaf_data['ns1:deathDate']) + return {} + + viaf_data = data.get('ns1:VIAFCluster', {}) + if not viaf_data: + return {} # pragma: no cover + + info: dict[str, object] = {} + if viaf_id := viaf_data.get('ns1:viafID'): + info['VIAF ID'] = link( + str(viaf_id), + f'https://viaf.org/viaf/{viaf_id}', + external=True) + + headings = viaf_data.get('ns1:mainHeadings', {}).get('ns1:data', []) + if isinstance(headings, list) and headings: + if text := headings[0].get('ns1:text'): + info['title'] = text + elif isinstance(headings, dict): # pragma: no cover + if text := headings.get('ns1:text'): + info['title'] = text + + if name_type := viaf_data.get('ns1:nameType'): + info['type'] = name_type + + if (birth_date := viaf_data.get('ns1:birthDate')) is not None \ + and str(birth_date) != '0': + info['birth date'] = str(birth_date) + + if (death_date := viaf_data.get('ns1:deathDate')) is not None \ + and str(death_date) != '0': + info['death date'] = str(death_date) return info diff --git a/openatlas/api/external/wikidata.py b/openatlas/api/external/wikidata.py index da0dc8579..b35d2c290 100644 --- a/openatlas/api/external/wikidata.py +++ b/openatlas/api/external/wikidata.py @@ -1,3 +1,4 @@ +from typing import Any import requests from openatlas import app @@ -10,18 +11,11 @@ class Wikidata(ExternalApi): # pylint: disable=too-few-public-methods @staticmethod def get_info(id_: str, system: Entity) -> dict[str, object]: - def add_resolver_url(wikidata_id: str) -> str: - return link( - f'Q{wikidata_id}', - f'{system.resolver_url}Q{wikidata_id}', - external=True) - params = { 'action': 'wbgetentities', 'ids': id_, 'format': 'json', 'languages': 'en'} - info = {} try: data = requests.get( 'https://www.wikidata.org/w/api.php', @@ -29,65 +23,48 @@ def add_resolver_url(wikidata_id: str) -> str: params=params, proxies=app.config['PROXIES'], timeout=10).json() + wiki = data['entities'][id_] except Exception: # pragma: no cover return {} - try: - info['title'] = data['entities'][id_]['labels']['en']['value'] - except KeyError: # pragma: no cover - pass - try: - info['aliases'] = [ - ' ' + v['value'] - for v in data['entities'][id_]['aliases']['en']] - except KeyError: # pragma: no cover - pass - try: - info['description'] = \ - data['entities'][id_]['descriptions']['en']['value'] - except KeyError: # pragma: no cover - pass - try: + + def add_resolver_url(numeric_id: int | str) -> str: + return link( + f'Q{numeric_id}', + f'{system.resolver_url}Q{numeric_id}', + external=True) + + def claim_values(property_: str) -> list[Any]: + values = [] + claims = wiki.get('claims', {}).get(property_, []) + for claim in claims: + if 'mainsnak' in claim and 'datavalue' in claim['mainsnak']: + values.append(claim['mainsnak']['datavalue']['value']) + return values + + info: dict[str, object] = {} + + if label := wiki.get('labels', {}).get('en', {}).get('value'): + info['title'] = label + if aliases := [ + a['value'] for a in wiki.get('aliases', {}).get('en', [])]: + info['aliases'] = [f' {a}' for a in aliases] + if desc := wiki.get('descriptions', {}).get('en', {}).get('value'): + info['description'] = desc + if founders := claim_values('P112'): info['founded by'] = [ - add_resolver_url( - v['mainsnak']['datavalue']['value']['numeric-id']) - for v in data['entities'][id_]['claims']['P112']] - except KeyError: # pragma: no cover - pass - try: - info['nick names'] = [ - v['mainsnak']['datavalue']['value']['text'] - for v in data['entities'][id_]['claims']['P1449']] - except KeyError: - pass - try: + add_resolver_url(f['numeric-id']) for f in founders] + if nicknames := claim_values('P1449'): # pragma: no cover + info['nick names'] = [n['text'] for n in nicknames] + if websites := claim_values('P856'): info['official websites'] = [ - ' ' + link( - v['mainsnak']['datavalue']['value'], - v['mainsnak']['datavalue']['value'], - external=True) - for v in data['entities'][id_]['claims']['P856']] - except KeyError: # pragma: no cover - pass - try: + f' {link(w, w, external=True)}' for w in websites] + if categories := claim_values('P910'): info['categories'] = [ - add_resolver_url( - v['mainsnak']['datavalue']['value']['numeric-id']) - for v in data['entities'][id_]['claims']['P910']] - except KeyError: # pragma: no cover - pass - try: - info['inception'] = \ - data['entities'][id_]['claims']['P571'][0]['mainsnak'][ - 'datavalue']['value']['time'] - except KeyError: # pragma: no cover - pass - try: - info['latitude'] = \ - data['entities'][id_]['claims']['P625'][0]['mainsnak'][ - 'datavalue']['value']['latitude'] - info['longitude'] = \ - data['entities'][id_]['claims']['P625'][0]['mainsnak'][ - 'datavalue']['value']['longitude'] - except KeyError: # pragma: no cover - pass - return info + add_resolver_url(c['numeric-id']) for c in categories] + if inception := claim_values('P571'): + info['inception'] = inception[0]['time'] + if coords := claim_values('P625'): + info['latitude'] = coords[0]['latitude'] + info['longitude'] = coords[0]['longitude'] + + return info \ No newline at end of file