Add RSS feed generator for commune de Silly (actualites + agenda)
Scrapes the imio.smartweb JSON API used by www.silly.be (found via the smartweb-news/smartweb-events web components in the raw HTML) instead of needing a headless browser. Shells out to curl since Cloudflare silently blocks Python's urllib/requests TLS fingerprint. Includes Yunohost cron and nginx deployment snippets. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
para
+198
@@ -0,0 +1,198 @@
|
||||
#!/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
|
||||
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"
|
||||
|
||||
NEWS_LINK_TMPL = f"{SITE}/actualites?news={{id}}"
|
||||
EVENT_LINK_TMPL = f"{SITE}/agenda?event={{id}}"
|
||||
|
||||
|
||||
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 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", "")
|
||||
link = NEWS_LINK_TMPL.format(id=item_id)
|
||||
title = item.get("title_fr") or item.get("title") or "(sans titre)"
|
||||
summary = strip_html(item.get("description_fr") or item.get("description"))
|
||||
body = strip_html(item.get("text_fr", {}).get("data") if isinstance(item.get("text_fr"), dict) else None)
|
||||
description = summary or body
|
||||
pub_dt = parse_dt(item.get("effective") or item.get("created"))
|
||||
guid = item.get("UID") or item_id
|
||||
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
|
||||
|
||||
|
||||
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]
|
||||
link = EVENT_LINK_TMPL.format(id=event_id)
|
||||
title = item.get("title_fr") or item.get("title") or "(sans titre)"
|
||||
summary = strip_html(item.get("description_fr") or item.get("description"))
|
||||
meta = format_event_meta(item)
|
||||
description = " | ".join(p for p in (meta, summary) if p)
|
||||
pub_dt = parse_dt(item.get("created") or item.get("start"))
|
||||
guid = item.get("UID") or event_id
|
||||
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
|
||||
|
||||
|
||||
def write_feed(path: str, title: str, link: str, description: str, items: list[str]) -> None:
|
||||
now = rfc822(datetime.now(timezone.utc))
|
||||
xml = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
'<rss version="2.0">\n'
|
||||
" <channel>\n"
|
||||
f" <title>{escape(title)}</title>\n"
|
||||
f" <link>{escape(link)}</link>\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,
|
||||
)
|
||||
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,
|
||||
)
|
||||
print(f"[OK] agenda.xml : {len(event_items)} items")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user