Source code for election_day.views.opendata_swiss

from babel.dates import format_date
from io import BytesIO
from onegov.core.security import Public
from onegov.core.utils import normalize_for_url
from onegov.election_day import _
from onegov.election_day import ElectionDayApp
from onegov.election_day.layouts import DefaultLayout
from onegov.election_day.models import Election
from onegov.election_day.models import ElectionCompound
from onegov.election_day.models import Principal
from onegov.election_day.models import Vote
from sedate import as_datetime
from webob.exc import HTTPNotImplemented
from xml.etree.ElementTree import Element
from xml.etree.ElementTree import ElementTree
from xml.etree.ElementTree import SubElement


from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from onegov.election_day.request import ElectionDayRequest
    from translationstring import TranslationString
    from webob.response import Response


[docs] def sub( parent: Element, tag: str, attrib: dict[str, str] | None = None, text: str | None = None ) -> Element: element = SubElement(parent, tag, attrib=attrib or {}) element.text = text or '' return element
@ElectionDayApp.view( model=Principal, name='catalog.rdf', permission=Public )
[docs] def view_rdf(self: Principal, request: 'ElectionDayRequest') -> bytes: """ Returns an XML / RDF / DCAT-AP for Switzerland format for opendata.swiss. See https://handbook.opendata.swiss/de/content/glossar/bibliothek/ dcat-ap-ch.html and https://dcat-ap.ch/ for more information. """ principal_name = request.app.principal.name publisher_id = self.open_data.get('id') publisher_name = self.open_data.get('name') publisher_mail = self.open_data.get('mail') if not publisher_id or not publisher_name or not publisher_mail: raise HTTPNotImplemented() @request.after def set_headers(response: 'Response') -> None: response.headers['Content-Type'] = 'application/rdf+xml; charset=UTF-8' layout = DefaultLayout(self, request) domains = dict(self.domains_election) domains.update(self.domains_vote) locales = {k: k[:2] for k in request.app.locales} default_locale = request.default_locale rdf = Element('rdf:RDF', attrib={ 'xmlns:dct': 'http://purl.org/dc/terms/', 'xmlns:dc': 'http://purl.org/dc/elements/1.1/', 'xmlns:dcat': 'http://www.w3.org/ns/dcat#', 'xmlns:foaf': 'http://xmlns.com/foaf/0.1/', 'xmlns:xsd': 'http://www.w3.org/2001/XMLSchema#', 'xmlns:rdfs': 'http://www.w3.org/2000/01/rdf-schema#', 'xmlns:rdf': 'http://www.w3.org/1999/02/22-rdf-syntax-ns#', 'xmlns:vcard': 'http://www.w3.org/2006/vcard/ns#', 'xmlns:odrs': 'http://schema.theodi.org/odrs#', 'xmlns:schema': 'http://schema.org/', }) catalog = sub(rdf, 'dcat:Catalog') session = request.session items: list[Election | ElectionCompound | Vote] items = session.query(Election).all() # type:ignore[assignment] items.extend(session.query(ElectionCompound)) items.extend(session.query(Vote)) translations = request.app.translations def translate(text: 'TranslationString', locale: str) -> str: translator = translations.get(locale) if translator: return text.interpolate(translator.gettext(text)) return text.interpolate(text) for item in sorted(items, key=lambda i: i.date, reverse=True): if not item.completed: continue is_vote = isinstance(item, Vote) # IDs item_id = '{}-{}'.format('vote' if is_vote else 'election', item.id) ds = sub(catalog, 'dcat:dataset') ds = sub(ds, 'dcat:Dataset', { 'rdf:about': f'http://{publisher_id}/{item_id}' }) sub(ds, 'dct:identifier', {}, f'{item_id}@{publisher_id}') # Dates sub( ds, 'dct:issued', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#dateTime'}, as_datetime(item.date).isoformat() ) last_modified = item.last_modified or as_datetime(item.date) assert last_modified is not None sub( ds, 'dct:modified', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#dateTime'}, last_modified.replace(microsecond=0).isoformat() ) sub( ds, 'dct:accrualPeriodicity', {'rdf:resource': 'http://publications.europa.eu/resource/' 'authority/frequency/IRREG'} ) # Theme sub( ds, 'dcat:theme', {'rdf:resource': 'http://publications.europa.eu/resource/' 'authority/data-theme/GOVE'} ) # Landing page sub( ds, 'dcat:landingPage', {'rdf:resource': request.link(item, 'data')} ) # Keywords for keyword in ( _("Vote") if is_vote else _("Election"), domains[item.domain] ): for locale, lang in locales.items(): value = translate(keyword, locale).lower() sub(ds, 'dcat:keyword', {'xml:lang': lang}, value) # Title for locale, lang in locales.items(): title = item.get_title(locale, default_locale) or '' sub(ds, 'dct:title', {'xml:lang': lang}, title) # Description for locale, lang in locales.items(): locale = locale if is_vote: if item.domain == 'canton': des = _( "Final results of the cantonal vote \"${title}\", " "${date}, ${principal}, " "broken down by municipalities.", mapping={ 'title': ( item.get_title(locale, default_locale) or '' ), 'date': format_date( item.date, format='long', locale=locale ), 'principal': principal_name } ) else: des = _( "Final results of the federal vote \"${title}\", " "${date}, ${principal}, " "broken down by municipalities.", mapping={ 'title': ( item.get_title(locale, default_locale) or '' ), 'date': format_date( item.date, format='long', locale=locale ), 'principal': principal_name } ) else: if item.domain == 'canton': des = _( "Final results of the cantonal election \"${title}\", " "${date}, ${principal}, " "broken down by candidates and municipalities.", mapping={ 'title': ( item.get_title(locale, default_locale) or '' ), 'date': format_date( item.date, format='long', locale=locale ), 'principal': principal_name } ) elif item.domain in ('region', 'district', 'none'): des = _( "Final results of the regional election \"${title}\", " "${date}, ${principal}, " "broken down by candidates and municipalities.", mapping={ 'title': ( item.get_title(locale, default_locale) or '' ), 'date': format_date( item.date, format='long', locale=locale ), 'principal': principal_name } ) else: des = _( "Final results of the federal election \"${title}\", " "${date}, ${principal}, " "broken down by candidates and municipalities.", mapping={ 'title': ( item.get_title(locale, default_locale) or '' ), 'date': format_date( item.date, format='long', locale=locale ), 'principal': principal_name } ) translated_des = translate(des, locale) sub(ds, 'dct:description', {'xml:lang': lang}, translated_des) # Format description for locale, lang in locales.items(): label = translate(_("Format Description"), locale) url = layout.get_opendata_link(lang) fmt_des = sub(ds, 'dct:relation') fmt_des = sub(fmt_des, 'rdf:Description', {'rdf:about': url}) sub(fmt_des, 'rdfs:label', {}, label) # Publisher pub = sub(ds, 'dct:publisher') pub = sub(pub, 'foaf:Organization', { 'rdf:about': f'https://{publisher_id}' }) sub(pub, 'foaf:name', {}, publisher_name) # Contact point mail = sub(ds, 'dcat:contactPoint') mail = sub(mail, 'vcard:Organization') sub(mail, 'vcard:fn', {}, publisher_name) sub(mail, 'vcard:hasEmail', { 'rdf:resource': 'mailto:{}'.format(publisher_mail) }) # Date date = sub(ds, 'dct:temporal') date = sub(date, 'dct:PeriodOfTime') sub( date, 'schema:startDate', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#date'}, item.date.isoformat() ) sub( date, 'schema:endDate', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#date'}, item.date.isoformat() ) # Distributions for fmt, extension, media_type, party_result in ( ('csv', 'csv', 'text/csv', False), ('json', 'json', 'application/json', False), ('parties-csv', 'csv', 'text/csv', True), ('parties-json', 'json', 'application/json', True), ): if party_result: if not getattr(item, 'has_party_results', False): continue url = request.link(item, f'data-{fmt}') # IDs dist = sub(ds, 'dcat:distribution') dist = sub(dist, 'dcat:Distribution', { 'rdf:about': f'http://{publisher_id}/{item_id}/{fmt}' }) sub(dist, 'dct:identifier', {}, fmt) # Title for locale, lang in locales.items(): title = item.get_title(locale, default_locale) or item.id if party_result: title += ' ({})'.format(translate(_('Parties'), locale)) title = f'{normalize_for_url(title)}.{extension}' sub(dist, 'dct:title', {'xml:lang': lang}, title) # Dates sub( dist, 'dct:issued', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#dateTime'}, item.created.replace(microsecond=0).isoformat() ) sub( dist, 'dct:modified', {'rdf:datatype': 'http://www.w3.org/2001/XMLSchema#dateTime'}, last_modified.replace(microsecond=0).isoformat() ) # URLs sub(dist, 'dcat:accessURL', {'rdf:resource': url}) sub(dist, 'dcat:downloadURL', {'rdf:resource': url}) # Legal license = sub( dist, 'dct:license', {'rdf:about': 'http://dcat-ap.ch/vocabulary/licenses/terms_by'} ) sub(license, 'rdf:type', { 'rdf:resource': 'http://purl.org/dc/terms/RightsStatement' }) # Media Type sub(dist, 'dcat:mediaType', {}, media_type) sub(dist, 'dcat:mediaType', { 'rdf:resource': f'https://www.iana.org/assignments/' f'media-types/{media_type}' }) sub(dist, 'dcat:format', { 'rdf:resource': f'http://publications.europa.eu/resource/' f'authority/file-type/{extension.upper()}' }) out = BytesIO() ElementTree(rdf).write(out, encoding='utf-8', xml_declaration=True) return out.getvalue()