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
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


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)
        description = (f"<p>{html.escape(meta)}</p>" if meta else "") + full_body_html(item, 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()