paraandClaude Sonnet 5
dac36592eb
Found by testing: unlike news items, the agenda @results batch listing always returns text_fr and items (attachments) as null for events - only description_fr (short teaser) was ever available from it. The site itself works around this by firing a second request per event (@results?UID=<uid>&fullobjects=1) when you open one; we now do the same, with a per-UID cache since a recurring event shares one UID across all its occurrences. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
276 lines
10 KiB
Python
276 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Genere deux flux RSS 2.0 (actualites + agenda) pour la commune de Silly (Belgique)
|
|
a partir de l'API JSON REST (Plone/imio.smartweb) utilisee par www.silly.be.
|
|
|
|
Aucun navigateur headless n'est necessaire : le HTML de www.silly.be/actualites et
|
|
/agenda contient les balises <smartweb-news> / <smartweb-events> dont l'attribut
|
|
query-url pointe directement vers l'API JSON (@results) qui retourne les items.
|
|
|
|
Sortie : deux fichiers XML (actualites.xml, agenda.xml) a servir statiquement.
|
|
"""
|
|
import html
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import unicodedata
|
|
from datetime import datetime, timezone
|
|
from email.utils import format_datetime
|
|
from xml.sax.saxutils import escape
|
|
|
|
# Le site www.silly.be est derriere Cloudflare, dont le WAF laisse passer
|
|
# curl (empreinte TLS/HTTP2 "navigateur") mais bloque silencieusement
|
|
# (reponse HTTP 200 avec un corps "{}") les requetes faites avec
|
|
# urllib/requests en Python. On shell-oute donc vers curl, qui est
|
|
# disponible par defaut sur toute machine Yunohost.
|
|
#
|
|
# Autre contrainte observee : le WAF ne laisse passer que b_size=20
|
|
# exactement (la valeur utilisee par le composant JS du site) ; toute
|
|
# autre valeur de b_size renvoie un corps vide. La pagination se fait
|
|
# donc avec b_size=20 fixe + b_start croissant.
|
|
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
SITE = "https://www.silly.be"
|
|
PAGE_SIZE = 20
|
|
MAX_PAGES = 2 # jusqu'a 40 items par flux
|
|
|
|
NEWS_API = f"{SITE}/actualites/@results"
|
|
EVENTS_API = f"{SITE}/agenda/@results"
|
|
|
|
# Base publique ou ce script publie ses flux une fois deploye (voir
|
|
# README.md, variante Yunohost/cron) - sert a l'atom:link rel="self".
|
|
# A adapter si vous deployez ailleurs.
|
|
SELF_BASE_URL = "https://akbar.nohost.me/silly-rss"
|
|
|
|
|
|
def fetch_json(url: str) -> dict:
|
|
result = subprocess.run(
|
|
["curl", "-s", "-A", USER_AGENT, "-H", "Accept: application/json", "--fail", url],
|
|
capture_output=True,
|
|
timeout=30,
|
|
check=True,
|
|
)
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def fetch_all_items(base_url: str) -> list[dict]:
|
|
items: list[dict] = []
|
|
for page in range(MAX_PAGES):
|
|
b_start = page * PAGE_SIZE
|
|
data = fetch_json(f"{base_url}?b_size={PAGE_SIZE}&b_start={b_start}")
|
|
page_items = data.get("items", [])
|
|
if not page_items:
|
|
break
|
|
items.extend(page_items)
|
|
if len(items) >= data.get("items_total", 0):
|
|
break
|
|
return items
|
|
|
|
|
|
def strip_html(text: str | None) -> str:
|
|
if not text:
|
|
return ""
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
text = html.unescape(text)
|
|
return re.sub(r"\s+", " ", text).strip()
|
|
|
|
|
|
def slugify(text: str) -> str:
|
|
"""
|
|
Reproduit le slug d'URL genere par le site (cosmetique : le routeur JS
|
|
de silly.be lit en realite le parametre `u` pour savoir quel item
|
|
afficher, pas ce segment de chemin - mais autant coller a l'URL
|
|
"officielle" telle qu'on la voit dans la barre d'adresse).
|
|
"""
|
|
ascii_text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii")
|
|
ascii_text = ascii_text.lower()
|
|
ascii_text = re.sub(r"[^a-z0-9]+", "-", ascii_text)
|
|
return ascii_text.strip("-")
|
|
|
|
|
|
def attachments_html(item: dict) -> str:
|
|
"""HTML des liens vers les pieces jointes (PDF, etc.) d'un item."""
|
|
parts = []
|
|
for sub in item.get("items") or []:
|
|
file_info = sub.get("file") or {}
|
|
url = file_info.get("download") or sub.get("targetUrl")
|
|
if not url:
|
|
continue
|
|
label = sub.get("title") or url.rsplit("/", 1)[-1]
|
|
parts.append(f'<p><a href="{html.escape(url)}">{html.escape(label)}</a></p>')
|
|
return "".join(parts)
|
|
|
|
|
|
def lead_image_html(item: dict, title: str) -> str:
|
|
"""Image principale de l'item, deja fournie par l'API (aucune requete
|
|
supplementaire necessaire). "paysage_affiche" est une taille
|
|
intermediaire raisonnable pour une image de flux RSS."""
|
|
url = item.get("image_affiche_scale") or item.get("image_full_scale")
|
|
if not url:
|
|
return ""
|
|
return f'<p><img src="{html.escape(url)}" alt="{html.escape(title)}" /></p>'
|
|
|
|
|
|
def full_body_html(item: dict, title: str) -> str:
|
|
"""Corps HTML complet (image + texte riche + pieces jointes), repli sur le resume court."""
|
|
text_fr = item.get("text_fr")
|
|
body = (text_fr.get("data") or "").strip() if isinstance(text_fr, dict) else ""
|
|
if not body:
|
|
summary = strip_html(item.get("description_fr") or item.get("description"))
|
|
body = f"<p>{html.escape(summary)}</p>" if summary else ""
|
|
return lead_image_html(item, title) + body + attachments_html(item)
|
|
|
|
|
|
def parse_dt(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def rfc822(dt: datetime | None) -> str:
|
|
if dt is None:
|
|
dt = datetime.now(timezone.utc)
|
|
return format_datetime(dt)
|
|
|
|
|
|
def build_news_items() -> list[str]:
|
|
entries = []
|
|
for item in fetch_all_items(NEWS_API):
|
|
item_id = item.get("id", "")
|
|
title = item.get("title_fr") or item.get("title") or "(sans titre)"
|
|
guid = item.get("UID") or item_id
|
|
link = f"{SITE}/actualites/{slugify(title)}?u={guid}"
|
|
description = full_body_html(item, title)
|
|
pub_dt = parse_dt(item.get("effective") or item.get("created"))
|
|
entries.append(
|
|
" <item>\n"
|
|
f" <title>{escape(title)}</title>\n"
|
|
f" <link>{escape(link)}</link>\n"
|
|
f" <guid isPermaLink=\"false\">{escape(guid)}</guid>\n"
|
|
f" <pubDate>{rfc822(pub_dt)}</pubDate>\n"
|
|
f" <description>{escape(description)}</description>\n"
|
|
" </item>"
|
|
)
|
|
return entries
|
|
|
|
|
|
_event_detail_cache: dict[str, dict | None] = {}
|
|
|
|
|
|
def fetch_event_detail(uid: str) -> dict | None:
|
|
"""
|
|
Contrairement aux actualites, l'API renvoie text_fr/items (texte complet
|
|
et pieces jointes) a null dans la liste paginee des evenements - il faut
|
|
une requete par UID pour les recuperer (c'est ce que fait le site
|
|
lui-meme quand on ouvre un evenement). Un evenement recurrent partage un
|
|
seul UID sur toutes ses occurrences, d'ou le cache pour eviter de
|
|
refaire la requete pour chaque occurrence.
|
|
"""
|
|
if uid in _event_detail_cache:
|
|
return _event_detail_cache[uid]
|
|
try:
|
|
data = fetch_json(f"{EVENTS_API}?UID={uid}&fullobjects=1")
|
|
detail = (data.get("items") or [None])[0]
|
|
except Exception:
|
|
detail = None
|
|
_event_detail_cache[uid] = detail
|
|
return detail
|
|
|
|
|
|
def format_event_meta(item: dict) -> str:
|
|
parts = []
|
|
start = parse_dt(item.get("start"))
|
|
end = parse_dt(item.get("end"))
|
|
if start:
|
|
if item.get("whole_day"):
|
|
parts.append(f"Le {start.strftime('%d/%m/%Y')}")
|
|
else:
|
|
parts.append(f"Le {start.strftime('%d/%m/%Y a %H:%M')}")
|
|
if end and end.date() != (start.date() if start else None):
|
|
parts.append(f"jusqu'au {end.strftime('%d/%m/%Y %H:%M')}")
|
|
address_bits = [item.get("street"), item.get("zipcode"), item.get("city")]
|
|
address = ", ".join(b for b in address_bits if b)
|
|
if address:
|
|
parts.append(f"Lieu : {address}")
|
|
return " — ".join(parts)
|
|
|
|
|
|
def build_event_items() -> list[str]:
|
|
entries = []
|
|
for item in fetch_all_items(EVENTS_API):
|
|
event_id = item.get("@id", "").rstrip("/").split("/")[-1]
|
|
title = item.get("title_fr") or item.get("title") or "(sans titre)"
|
|
guid = item.get("UID") or event_id
|
|
link = f"{SITE}/agenda/{slugify(title)}?u={guid}"
|
|
meta = format_event_meta(item)
|
|
detail = fetch_event_detail(guid) or item
|
|
description = (f"<p>{html.escape(meta)}</p>" if meta else "") + full_body_html(detail, title)
|
|
pub_dt = parse_dt(item.get("created") or item.get("start"))
|
|
# Un evenement recurrent partage le meme UID sur toutes ses
|
|
# occurrences : on suffixe le guid avec la date de debut pour
|
|
# rester unique par occurrence, comme l'exige la spec RSS.
|
|
occurrence_guid = f"{guid}-{item.get('start', '')}"
|
|
entries.append(
|
|
" <item>\n"
|
|
f" <title>{escape(title)}</title>\n"
|
|
f" <link>{escape(link)}</link>\n"
|
|
f" <guid isPermaLink=\"false\">{escape(occurrence_guid)}</guid>\n"
|
|
f" <pubDate>{rfc822(pub_dt)}</pubDate>\n"
|
|
f" <description>{escape(description)}</description>\n"
|
|
" </item>"
|
|
)
|
|
return entries
|
|
|
|
|
|
def write_feed(path: str, title: str, link: str, description: str, items: list[str], self_url: str) -> None:
|
|
now = rfc822(datetime.now(timezone.utc))
|
|
xml = (
|
|
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
|
'<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom">\n'
|
|
" <channel>\n"
|
|
f" <title>{escape(title)}</title>\n"
|
|
f" <link>{escape(link)}</link>\n"
|
|
f' <atom:link href="{escape(self_url)}" rel="self" type="application/rss+xml" />\n'
|
|
f" <description>{escape(description)}</description>\n"
|
|
" <language>fr-be</language>\n"
|
|
f" <lastBuildDate>{now}</lastBuildDate>\n"
|
|
+ "\n".join(items) + "\n"
|
|
" </channel>\n"
|
|
"</rss>\n"
|
|
)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(xml)
|
|
|
|
|
|
def main() -> None:
|
|
out_dir = sys.argv[1] if len(sys.argv) > 1 else "."
|
|
|
|
news_items = build_news_items()
|
|
write_feed(
|
|
f"{out_dir}/actualites.xml",
|
|
title="Silly.be - Actualités",
|
|
link=f"{SITE}/actualites",
|
|
description="Actualités de la commune de Silly (Belgique)",
|
|
items=news_items,
|
|
self_url=f"{SELF_BASE_URL}/actualites.xml",
|
|
)
|
|
print(f"[OK] actualites.xml : {len(news_items)} items")
|
|
|
|
event_items = build_event_items()
|
|
write_feed(
|
|
f"{out_dir}/agenda.xml",
|
|
title="Silly.be - Agenda",
|
|
link=f"{SITE}/agenda",
|
|
description="Agenda / événements de la commune de Silly (Belgique)",
|
|
items=event_items,
|
|
self_url=f"{SELF_BASE_URL}/agenda.xml",
|
|
)
|
|
print(f"[OK] agenda.xml : {len(event_items)} items")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|