silly-rss/silly_rss.py

#!/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()