holidays.py

import asyncio
import re
from pathlib import Path

import httpx
from bs4 import BeautifulSoup

from utils import Cache, Config, Notifier


async def fetch_page(client: httpx.AsyncClient, page: int):
    params = {
        "typesej[]": "rouge",
        "page": page + 1,
    }

    try:
        response = await client.get(
            "https://gdscatalogueur.ccas.fr/search",
            params=params,
            timeout=30,
        )
        response.raise_for_status()
        return response.text
    except Exception:
        return None


def extract_stay_links(content: str) -> set[str]:
    pattern = r'["\']([^"\']*?\/sejour\/fiche\/[^"\']*?\/neo[^"\']*?)["\']'
    matches = re.findall(pattern, content)
    return {f"https://gdscatalogueur.ccas.fr{match}" for match in matches}


async def fetch_stay_content(client: httpx.AsyncClient, link: str) -> str:
    try:
        response = await client.get(link, timeout=30)
        response.raise_for_status()
        return response.text
    except Exception:
        return ""


def parse_stay_data(soup: BeautifulSoup) -> dict | None:
    data = {}

    title = soup.find("h1", class_="titleStayOne")
    if not title:
        return None
    data["title"] = title.get_text(strip=True)

    sessions = []
    date_pattern = r"(\d{2}/\d{2}/\d{4})"
    sessions_container = soup.find_all("div", class_="bull")
    for session in sessions_container:
        sess_date = session.find(class_="sessDate")
        if not sess_date:
            continue

        text = sess_date.get_text(strip=True)
        dates = re.findall(date_pattern, text)
        if len(dates) < 2:
            continue

        session_info = {"fmt": f"{dates[0]} - {dates[1]}", "places": None}

        places_text = session.find(string=re.compile(r"places disponibles"))
        if places_text:
            match = re.search(r"(\d+)\s*places disponibles", places_text)
            if match:
                session_info["places"] = int(match.group(1))

        sessions.append(session_info)

    if not sessions:
        return None
    data["sessions"] = sessions

    tarif = soup.find("b", string=re.compile(r"Tarif Référence\s*:"))
    data["tarif_ref"] = _extract_next(tarif)

    participation = soup.find(
        "b", string=re.compile(r"Votre participation selon votre coefficient social\s*:")
    )
    data["participation"] = _extract_next(participation)

    restauration = soup.find("b", string=re.compile(r"Type de Restauration\s*:"))
    data["restauration"] = _extract_next(restauration)

    return data


def _extract_next(element) -> str | None:
    if not element:
        return None
    next_text = element.next_sibling
    if not next_text:
        return None
    return next_text.strip() if isinstance(next_text, str) else next_text.get_text(strip=True)


def format_message(data: dict, link: str) -> str:
    message = "🌍" if data.get("tarif_ref") else "🏕️"
    message += f" [{data['title']}]({link})"
    if data.get("restauration"):
        message += f" - {data['restauration']}"
    message += "\n"
    message += f"{len(data['sessions'])} session(s) disponible(s):\n"
    for session in data.get("sessions", []):
        message += f"- {session['fmt']}"
        if session["places"]:
            message += f" ({session['places']} place(s))"
        message += "\n"

    if data.get("tarif_ref"):
        message += f"Tarif de référence: {data['tarif_ref']}\n"
    if data.get("participation"):
        message += f"Votre participation (selon coef.social): {data['participation']}"

    return message


async def process_stay(
    client: httpx.AsyncClient, link: str, cache: Cache, notifier: Notifier, semaphore: asyncio.Semaphore
):
    stay_id = f"{link.split('/')[-2]}/{link.split('/')[-4]}"
    if cache.hit(stay_id):
        return

    async with semaphore:
        content = await fetch_stay_content(client, link)
        if not content:
            return

        soup = BeautifulSoup(content, "html.parser")
        data = parse_stay_data(soup)
        if not data:
            return

        message = format_message(data, link)
        await notifier.send(message)
        cache.add(stay_id)
        await asyncio.sleep(1)


async def scrape():
    PAGE = 0
    processed_links = set()

    cache = Cache(Path(__file__).with_name(".cache_holidays"))
    notifier = Notifier(Config.get("apprise_urls"))

    headers = {
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"
    }
    semaphore = asyncio.Semaphore(2)
    async with httpx.AsyncClient(headers=headers) as client:
        while True:
            print(f"[p. {PAGE}] Fetching...")
            page = await fetch_page(client, PAGE)
            if not page:
                break

            sejour_links = extract_stay_links(page)
            if not sejour_links:
                break

            links_to_process = sejour_links - processed_links
            if not links_to_process:
                PAGE += 1
                await asyncio.sleep(5)
                continue

            tasks = [process_stay(client, link, cache, notifier, semaphore) for link in links_to_process]
            await asyncio.gather(*tasks)

            processed_links.update(links_to_process)
            PAGE += 1
            await asyncio.sleep(5)


def main():
    Config.load(["apprise_urls"])
    asyncio.run(scrape())


if __name__ == "__main__":
    main()