" + translatedOfficialPage[lang] +"
" + if item["collectedSignature"] == None: # avoid being redundant with the start date + item["text"] += getSignatureInfo(creationDate, lang) + else: + item["text"] += translatedInitiativeStarted[lang].format(date=creationDate.date()) + + # Status phrase + if item["withdrawn"] == True: + item["text"] += translatedInititiveWithdrawn[lang] + + elif item["invalid"] == True: + item["text"] += translatedInvalidInitiative[lang] + + elif item["acceptedInVote"] == False: + item["text"] += translatedRejectedInVote[lang] + + elif item["acceptedInVote"] == True: + item["text"] += translatedAcceptedInVote[lang] + + elif item["collectedSignature"] == True: + item["text"] += translatedSignatureThresholdReachedWaitingVote[lang] + + elif item["collectedSignature"] == False: + item["text"] += translatedEndDateAndInsufficientSignatures[lang] + + elif item["collectedSignature"] == None: + item["text"] += getSignatureInfo(creationDate, lang) + + else: + raise ValueError( + "Invalid initiative state combination: " + f"item[\"withdrawn\"]={item["withdrawn"]}, " + f"item[\"invalid\"]={item["invalid"]}, " + f"item[\"acceptedInVote\"]={item["acceptedInVote"]}, " + f"item[\"collectedSignature\"]={item["collectedSignature"]}, " + f"item[\"finished\"]={item["finished"]}" + ) + + # Constitutional change + item["text"] += getValue(entry, "text_" + lang) + + feeds[lang].append(copy.deepcopy(item)) + + logging.info("Parsed query. Generating feeds...") + generateFeed( + "Initiatives populaires fédérales", + "Flux RSS des initiatives populaires fédérales", + "initiativesPopulairesFederales", + "fr", + ["rss", "atom"], + feeds["fr"] + ) + + generateFeed( + "Eidgenössisch Volksinitiativen", + "RSS-Feed der föderalen Volksinitiativen", + "eidgenossischVolksinitiativen", + "de", + ["rss", "atom"], + feeds["de"] + ) + + generateFeed( + "Iniziative popolari federali", + "Feed RSS delle iniziative popolari federali", + "iniziativePopolariFederali", + "it", + ["rss", "atom"], + feeds["it"] + ) diff --git a/resspublica/main.py b/resspublica/main.py index de362b7..e4c6bd3 100644 --- a/resspublica/main.py +++ b/resspublica/main.py @@ -1,5 +1,4 @@ from SPARQLWrapper import SPARQLWrapper, JSON -from feedgen.feed import FeedGenerator from importlib.resources import files from datetime import datetime from dateutil.relativedelta import relativedelta @@ -11,331 +10,12 @@ import copy import argparse import logging +import .utils +import .translations +import .federalInitiativesFeeds + logger = logging.getLogger("resspublica") -def getSignatureInfo(start_date, lang="fr"): - MONTHS = { - "fr": ["janvier","février","mars","avril","mai","juin", - "juillet","août","septembre","octobre","novembre","décembre"], - "de": ["Januar","Februar","März","April","Mai","Juni", - "Juli","August","September","Oktober","November","Dezember"], - "it": ["gennaio","febbraio","marzo","aprile","maggio","giugno", - "luglio","agosto","settembre","ottobre","novembre","dicembre"] - } - - texts = { - "fr": ("La collecte des signatures a commencé le", "et se terminera vers
"), - "de": ("Die Unterschriftensammlung begann am", "und endet ungefähr im
"), - "it": ("La raccolta delle firme è iniziata il", "e terminerà circa nel
"), - } - if not isinstance(start_date, datetime): - start_date = datetime.fromisoformat(start_date).replace(tzinfo=ZoneInfo("Europe/Zurich")) - end_date = start_date + relativedelta(months=18) - - def format_full(d): - if lang == "de": - return f"{d.day}. {MONTHS[lang][d.month - 1]} {d.year}" - return f"{d.day} {MONTHS[lang][d.month - 1]} {d.year}" - - def format_month_year(d): - return f"{MONTHS[lang][d.month - 1]} {d.year}" - - start = format_full(start_date) - end = format_month_year(end_date) - - return f""" -- {texts[lang][0]} {start} {texts[lang][1]} {end}. -
- """.strip() -def getValue(row, key): - return row.get(key, {}).get("value") or None -def generateFeed(title, description, fileName, language, standards, entries): - time = datetime.now(ZoneInfo("Europe/Zurich")) - fg = FeedGenerator() - fg.title(title) - # We need to set a different link between atom and rss. As rss links are easier with feedgen we set fg.link for atom and fg.__rss_link for rss - # See https://github.com/lkiesow/python-feedgen/blob/main/feedgen/feed.py#L542 - fg.link(href="https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/atom/" + language + "/" + fileName, rel='self') - fg.__rss_link = "https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/rss/" + language + "/" + fileName - fg.id("https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/atom/" + language + "/" + fileName) - fg.description(description) - fg.language(language) - fg.updated(time) - fg.author({'name': 'ReSSPublica'}) - fg.docs("https://github.com/tomasriveral/ReSSPublica") - - for item in entries: - fe = fg.add_entry() - fe.title(item["title"]) - fe.guid(f"initiative-{item['id']}", permalink=False) - fe.pubDate(item["date"]) - fe.source({'url': item["url"], 'title': item["source"]}) - fe.content(item["text"], type="html") - if "rss" in standards: - fg.rss_file("./feed/rss/" + language + "/" + fileName + ".xml", pretty=True) - if "atom" in standards: - fg.atom_file("./feed/atom/" + language + "/" + fileName + ".atom", pretty=True) - -def generateFederalFeed(): - - logger.info("Making th SPARQL query") - sparql = SPARQLWrapper("https://cached.lindas.admin.ch/query") - query = files("resspublica.queries").joinpath("federalPopularInitiatives.sparql").read_text() - sparql.setQuery(query) - sparql.setReturnFormat(JSON) - - results = sparql.query().convert() - - logger.info("Query succesfully returned. Processing the feeds.") - logger.info(f"Lenght of the return: {len(results["results"]["bindings"])}\nShould be greater or equal than 550 if the query limit wasn't overriden.") - - now = datetime.now(ZoneInfo("Europe/Zurich")) - - # remove some of useless json data such as "head": {"vars": [...]} - data = results["results"]["bindings"] - - feeds = { - "fr": [], - "de": [], - "it": [] - } - - BASE_URLS = { - "fr": "https://www.bk.admin.ch/fr/details-initiatives-populaires/?initiative=", - "de": "https://www.bk.admin.ch/de/details-volksinitiativen/?initiative=", - "it": "https://www.bk.admin.ch/it/dettagli-iniziative-popolari/?initiative=" - } - BASE_SOURCE = { - "fr": "Chancellerie fédérale ChF", - "de": "Bundeskanzlei BK", - "it": "Cancelleria federale CaF" - } - BASE_PAGE = { - "fr": "Page officielle", - "de": "Offizielle Seite", - "it": "Pagina ufficiale" - } - - message_collecting = { - "fr": "La collecte des signatures a commencé et est en cours.
", - "de": "Die Sammlung der Unterschriften hat begonnen und läuft noch.
", - "it": "La raccolta delle firme è iniziata ed è in corso.
" - } - - message_threshold_reached_waiting_vote = { - "fr": "L’initiative a atteint 100'000 signatures et attend la votation.
", - "de": "Die Initiative hat 100'000 Unterschriften erreicht und wartet auf die Abstimmung.
", - "it": "L’iniziativa ha raggiunto 100'000 firme ed è in attesa della votazione.
" - } - - message_finished_insufficient_signatures = { - "fr": "L’initiative n’a pas atteint les 100'000 signatures et la procédure est terminée.
", - "de": "Die Initiative hat das erforderliche Quorum nicht erreicht und das Verfahren ist abgeschlossen.
", - "it": "L’iniziativa non ha raggiunto il numero richiesto di firme e la procedura è conclusa.
" - } - - message_withdrawn = { - "fr": "L’initiative a été retirée.
", - "de": "Die Initiative wurde zurückgezogen.
", - "it": "L’iniziativa è stata ritirata.
" - } - - message_invalid = { - "fr": "L’initiative a été déclarée invalide.
", - "de": "Die Initiative wurde für ungültig erklärt.
", - "it": "L’iniziativa è stata dichiarata non valida.
" - } - - message_accepted_vote = { - "fr": "L’initiative a été acceptée en votation.
", - "de": "Die Initiative wurde in der Abstimmung angenommen.
", - "it": "L’iniziativa è stata accettata in votazione.
" - } - - message_rejected_vote = { - "fr": "L’initiative a été rejetée en votation.
", - "de": "Die Initiative wurde in der Abstimmung abgelehnt.
", - "it": "L’iniziativa è stata respinta in votazione.
" - } - message_opened = { - "fr": "L’initiative a été lancée le {date}.
", - "de": "Die Initiative wurde am {date} gestartet.
", - "it": "L’iniziativa è stata lanciata il {date}.
" - } - - FEDERAL_DB_PATH = CACHE / "federalInitiatives.json" - db = TinyDB(FEDERAL_DB_PATH) - q = Query() - - for entry in data: - logger.debug(f"Popular initiative {getValue(entry, "title_fr")}") - creationDate = datetime.strptime(getValue(entry, "date"), "%Y-%m-%d").replace(tzinfo=ZoneInfo("Europe/Zurich")) - item = { - "id": getValue(entry, "id"), - # For status, zustand, zurueckgezogen, ungueltig and angenommen, we just get the URI as we can deduce the info we need from it and it makes queries faster. - # status_uri erledigt/haengig main lifecycle state - "finished": "erledigt" in getValue(entry, "status_uri"), - "date": getValue(entry, "date") - } - logger.debug(f"status_uri {getValue(entry, "status_uri")}") - logger.debug(f"finished {str(item["finished"])}") - - # zustand_uri nein/ja/Undefined valid initiative reached threshold - zustand = getValue(entry, "zustand_uri") - logger.debug(f"zustand_uri: {zustand}") - if "Undefined" in zustand: - item["collectedSignature"] = None - elif "ja" in zustand: - item["collectedSignature"] = True - elif "nein" in zustand: - item["collectedSignature"] = False - else: - raise ValueError("Unexpected value in zustand_uri: " + zustand + " (Expected are [...]nein, [...]ja and [...]Undefined)") - - logger.debug(f"collectedSignature {str(item["collectedSignature"])}") - - # zurueckgezogen_uri nein/ja/Undefined withdrawal - zurueckgezogen = getValue(entry, "zurueckgezogen_uri") - logger.debug(f"zurueckgezogen_uri {getValue(entry, "zurueckgezogen_uri")}") - if "Undefined" in zurueckgezogen: - item["withdrawn"] = None - elif "ja" in zurueckgezogen: - item["withdrawn"] = True - elif "nein" in zurueckgezogen: - item["withdrawn"] = False - else: - raise ValueError("Unexpected value in zurueckgezogen_uri: " + zurueckgezogen + " (Expected are [...]nein, [...]ja and [...]Undefined)") - logger.debug(f"withdrawn {str(item["withdrawn"])}") - - # I couldn't find an example for invalided initiatives. So if it is not Undefined we set to true. invalid initiative - logger.debug(f"ungueltig_uri {getValue(entry, "ungueltig_uri")}") - item["invalid"] = "Undefined" not in getValue(entry, "ungueltig_uri") - logger.debug(f"invalid {str(item["invalid"])}") - - # angenommen_uri nein/ja/Undefined/undefiniert For some reason two types of undefined. accepted in vote - angenommen = getValue(entry, "angenommen_uri") - logger.debug(f"angenommen_uri {getValue(entry, "angenommen_uri")}") - if "Undefined" in angenommen or "undefiniert" in angenommen: - item["acceptedInVote"] = None - elif "ja" in angenommen: - item["acceptedInVote"] = True - elif "nein" in angenommen: - item["acceptedInVote"] = False - else: - raise ValueError("Unexpected value in angenommen_uri: " + angenommen + " (Expected are [...]nein, [...]ja, [...]Undefined and [...]undefiniert)") - - logger.debug(f"acceptedInVote {str(item["acceptedInVote"])}") - - # if we already got the initiative in the db - if db.contains(q.id == item["id"]): - oldStatus = db.get(q.id == item["id"]) - needsUpdate = False - - if {k: oldStatus.get(k) for k in oldStatus if k != "date"} != {k: item.get(k) for k in item if k != "date"}: - needsUpdate = True - - if needsUpdate: - item["date"] = now.isoformat() - else: - item["date"] = oldStatus["date"] - db.upsert(item, q.id == item["id"]) - item["date"] = datetime.fromisoformat(item["date"]).replace(tzinfo=ZoneInfo("Europe/Zurich")) - else: - # if we don't have it: - # if still in signature phase -> creationDate - # if not -> now - item["date"] = creationDate.isoformat() - - db.upsert(item, q.id == item["id"]) - item["date"] = datetime.fromisoformat(item["date"]).replace(tzinfo=ZoneInfo("Europe/Zurich")) - - logger.debug(f"date {item["date"].isoformat()}") - for lang in ["fr", "de", "it"]: - item["title"] = getValue(entry, "title_" + lang) - if item["title"] == None: - item["title"] = "Missing data" - item["url"] = BASE_URLS.get(lang, BASE_URLS["de"]) + str(getValue(entry, "id")) - item["source"] = BASE_SOURCE[lang] - - # We need to construct the text part in three phases: - # 1. Hyperlink to official page and initial start date - # 2. status phrase - # 3. constitutional change - - - # Hyperlink to official page and initial start date - item["text"] = "" - if item["collectedSignature"] == None: # avoid being redundant with the start date - item["text"] += getSignatureInfo(creationDate, lang) - else: - item["text"] += message_opened[lang].format(date=creationDate.date()) - - # Status phrase - if item["withdrawn"] == True: - item["text"] += message_withdrawn[lang] - - elif item["invalid"] == True: - item["text"] += message_invalid[lang] - - elif item["acceptedInVote"] == False: - item["text"] += message_rejected_vote[lang] - - elif item["acceptedInVote"] == True: - item["text"] += message_accepted_vote[lang] - - elif item["collectedSignature"] == True: - item["text"] += message_threshold_reached_waiting_vote[lang] - - elif item["collectedSignature"] == False: - item["text"] += message_finished_insufficient_signatures[lang] - - elif item["collectedSignature"] == None: - item["text"] += getSignatureInfo(creationDate, lang) - - else: - raise ValueError( - "Invalid initiative state combination: " - f"item[\"withdrawn\"]={item["withdrawn"]}, " - f"item[\"invalid\"]={item["invalid"]}, " - f"item[\"acceptedInVote\"]={item["acceptedInVote"]}, " - f"item[\"collectedSignature\"]={item["collectedSignature"]}, " - f"item[\"finished\"]={item["finished"]}" - ) - - # Constitutional change - item["text"] += getValue(entry, "text_" + lang) - - feeds[lang].append(copy.deepcopy(item)) - - - generateFeed( - "Initiatives populaires fédérales", - "Flux RSS des initiatives populaires fédérales", - "initiativesPopulairesFederales", - "fr", - ["rss", "atom"], - feeds["fr"] - ) - - generateFeed( - "Eidgenössisch Volksinitiativen", - "RSS-Feed der föderalen Volksinitiativen", - "eidgenossischVolksinitiativen", - "de", - ["rss", "atom"], - feeds["de"] - ) - - generateFeed( - "Iniziative popolari federali", - "Feed RSS delle iniziative popolari federali", - "iniziativePopolariFederali", - "it", - ["rss", "atom"], - feeds["it"] - ) - def main(): parser = argparse.ArgumentParser() parser.add_argument("--verbose", action="store_true", help="Enable debug logging") diff --git a/resspublica/translations.py b/resspublica/translations.py new file mode 100644 index 0000000..3a15a8b --- /dev/null +++ b/resspublica/translations.py @@ -0,0 +1,71 @@ +# used in main.py +translatedFederalPopularInitiativesUrls = { + "fr": "https://www.bk.admin.ch/fr/details-initiatives-populaires/?initiative=", + "de": "https://www.bk.admin.ch/de/details-volksinitiativen/?initiative=", + "it": "https://www.bk.admin.ch/it/dettagli-iniziative-popolari/?initiative=" +} +translatedFederalChancellery = { + "fr": "Chancellerie fédérale ChF", + "de": "Bundeskanzlei BK", + "it": "Cancelleria federale CaF" +} +translatedOfficialPage = { + "fr": "Page officielle", + "de": "Offizielle Seite", + "it": "Pagina ufficiale" +} +translatedSignatureThresholdReachedWaitingVote = { + "fr": "L’initiative a atteint 100'000 signatures et attend la votation.
", + "de": "Die Initiative hat 100'000 Unterschriften erreicht und wartet auf die Abstimmung.
", + "it": "L’iniziativa ha raggiunto 100'000 firme ed è in attesa della votazione.
" +} + +translatedEndDateAndInsufficientSignatures = { + "fr": "L’initiative n’a pas atteint les 100'000 signatures et la procédure est terminée.
", + "de": "Die Initiative hat das erforderliche Quorum nicht erreicht und das Verfahren ist abgeschlossen.
", + "it": "L’iniziativa non ha raggiunto il numero richiesto di firme e la procedura è conclusa.
" +} + +translatedInititiveWithdrawn = { + "fr": "L’initiative a été retirée.
", + "de": "Die Initiative wurde zurückgezogen.
", + "it": "L’iniziativa è stata ritirata.
" +} + +translatedInvalidInitiative = { + "fr": "L’initiative a été déclarée invalide.
", + "de": "Die Initiative wurde für ungültig erklärt.
", + "it": "L’iniziativa è stata dichiarata non valida.
" +} + +translatedAcceptedInVote = { + "fr": "L’initiative a été acceptée en votation.
", + "de": "Die Initiative wurde in der Abstimmung angenommen.
", + "it": "L’iniziativa è stata accettata in votazione.
" +} + +translatedRejectedInVote = { + "fr": "L’initiative a été rejetée en votation.
", + "de": "Die Initiative wurde in der Abstimmung abgelehnt.
", + "it": "L’iniziativa è stata respinta in votazione.
" +} +translatedInitiativeStarted = { + "fr": "L’initiative a été lancée le {date}.
", + "de": "Die Initiative wurde am {date} gestartet.
", + "it": "L’iniziativa è stata lanciata il {date}.
" +} +# used in utils.py +translatedMonths = { + "fr": ["janvier","février","mars","avril","mai","juin", + "juillet","août","septembre","octobre","novembre","décembre"], + "de": ["Januar","Februar","März","April","Mai","Juni", + "Juli","August","September","Oktober","November","Dezember"], + "it": ["gennaio","febbraio","marzo","aprile","maggio","giugno", + "luglio","agosto","settembre","ottobre","novembre","dicembre"] +} + +translatedSignatureDates = { + "fr": ("La collecte des signatures a commencé le", "et se terminera vers
"), + "de": ("Die Unterschriftensammlung begann am", "und endet ungefähr im
"), + "it": ("La raccolta delle firme è iniziata il", "e terminerà circa nel
"), +} diff --git a/resspublica/utils.py b/resspublica/utils.py new file mode 100644 index 0000000..453f6cf --- /dev/null +++ b/resspublica/utils.py @@ -0,0 +1,60 @@ +from feedgen.feed import FeedGenerator + + +def generateFeed(title, description, fileName, language, standards, entries): + logger.info(f"Generating feed {title}") + time = datetime.now(ZoneInfo("Europe/Zurich")) + fg = FeedGenerator() + fg.title(title) + # We need to set a different link between atom and rss. As rss links are easier with feedgen we set fg.link for atom and fg.__rss_link for rss + # See https://github.com/lkiesow/python-feedgen/blob/main/feedgen/feed.py#L542 + fg.link(href="https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/atom/" + language + "/" + fileName, rel='self') + fg.__rss_link = "https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/rss/" + language + "/" + fileName + fg.id("https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/feed/atom/" + language + "/" + fileName) + fg.description(description) + fg.language(language) + fg.updated(time) + fg.author({'name': 'ReSSPublica'}) + fg.docs("https://github.com/tomasriveral/ReSSPublica") + + for item in entries: + fe = fg.add_entry() + fe.title(item["title"]) + fe.guid(f"initiative-{item['id']}", permalink=False) + fe.pubDate(item["date"]) + fe.source({'url': item["url"], 'title': item["source"]}) + fe.content(item["text"], type="html") + if "rss" in standards: + fg.rss_file("./feed/rss/" + language + "/" + fileName + ".xml", pretty=True) + logger.debug("Wrote to") + logger.debug("./feed/rss/" + language + "/" + fileName + ".xml") + if "atom" in standards: + fg.atom_file("./feed/atom/" + language + "/" + fileName + ".atom", pretty=True) + logger.debug("Wrote to") + logger.debug("./feed/atom/" + language + "/" + fileName + ".atom") + + logger.debug("Finished generating feeds") +def getSignatureInfo(start_date, lang="fr"): + if not isinstance(start_date, datetime): + start_date = datetime.fromisoformat(start_date).replace(tzinfo=ZoneInfo("Europe/Zurich")) + end_date = start_date + relativedelta(months=18) + + def format_full(d): + if lang == "de": + return f"{d.day}. {translatedMonths[lang][d.month - 1]} {d.year}" + return f"{d.day} {translatedMonths[lang][d.month - 1]} {d.year}" + + def format_month_year(d): + return f"{translatedMonths[lang][d.month - 1]} {d.year}" + + start = format_full(start_date) + end = format_month_year(end_date) + + return f""" ++ {translatedSignatureDates[lang][0]} {start} {translatedSignatureDates[lang][1]} {end}. +
+ """.strip() + +def getValue(row, key): + return row.get(key, {}).get("value") or None