tickets.py

import asyncio
from datetime import date
from pathlib import Path

import httpx

from utils import (Cache, Config, Notifier, get_category_url,
                   to_bon_francais_date)


def matches_features(product_features: list[dict]) -> bool:
    required_features = Config.get("features", {})

    for feature_id, allowed_values in required_features.items():
        feature_value = next(
            (int(f["id_feature_value"]) for f in product_features if f["id"] == str(feature_id)), None
        )
        if feature_value and feature_value in allowed_values:
            return True

    return False


async def fetch_batch(client: httpx.AsyncClient, offset: int, limit: int, today: str):
    params = {
        "display": "full",  # Looking for [id,base_name,manufacturer_name,representation_date], but 'associations' key is only displayed using 'full'
        "output_format": "JSON",
        "filter[representation_date]": f">[{today}]",
        "limit": f"{offset},{limit}",
    }

    try:
        response = await client.get(
            "https://portail-culture-et-loisirs.ccas.fr/api/products",
            params=params,
            timeout=30,
        )
        response.raise_for_status()
        return response.json().get("products", [])
    except Exception:
        return []


async def process_product(product: dict, cache: Cache, notifier: Notifier):
    product_id = str(product.get("id"))
    if cache.hit(product_id):
        return

    features = product.get("associations", {}).get("product_features", [])
    if not matches_features(features):
        return

    name = product.get("base_name")
    location = product.get("manufacturer_name")
    day = product.get("representation_date", " ").split(" ")[0]

    category_id = product.get("id_category_default")
    category_url = get_category_url(category_id)
    link = f"https://portail-culture-et-loisirs.ccas.fr/{category_url}/{product_id}-{product.get('link_rewrite')}.html"

    match category_id:
        case 101:
            message = "🎫"
        case 102:
            message = "🎭"
        case 103:
            message = "🏛️"
        case 104:
            message = "🌳"
        case 105:
            message = "⚽"
        case _:
            message = "🎫"
    message += f"🎫 {name}\n{location}, {to_bon_francais_date(day)}\n{link}"
    await notifier.send(message)
    cache.add(product_id)


async def scrape():
    today = date.today().isoformat()
    LIMIT = 50
    OFFSET = 0

    cache = Cache(Path(__file__).with_name(".cache"))
    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"
    }
    async with httpx.AsyncClient(headers=headers) as client:
        while True:
            print(f"[{OFFSET}..{OFFSET + LIMIT}] Fetching")

            products = await fetch_batch(client, OFFSET, LIMIT, today)
            if not products:
                break

            await asyncio.gather(*[process_product(p, cache, notifier) for p in products])

            OFFSET += LIMIT
            await asyncio.sleep(10)


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


if __name__ == "__main__":
    main()