busapi.py

"""
Client for mauritius-buses.com — an (unofficial) community journey-planner for
Mauritian buses. It exposes the only structured A->B routing data we could find
for the island, returning bus line numbers, ordered intermediate stops with
cumulative travel time, per-line frequency (headway) and fares.

Two operations:
  - get_stops()          : the full catalogue of selectable bus stops (id, name)
  - plan(day, frm, to)   : a structured itinerary between two stop ids

The site has no documented API, so we drive the same POST the search form makes
and parse the resulting HTML. We cache aggressively and send a descriptive
User-Agent to stay a polite consumer.
"""

import json
import os
import re
import time

import requests
from bs4 import BeautifulSoup

BASE = "https://www.mauritius-buses.com"
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
STOPS_CACHE = os.path.join(DATA_DIR, "stops.json")

UA = (
    "MauritiusBusPlanner/1.0 (personal route-planning demo; "
    "contact: syadone@gmail.com)"
)

_session = requests.Session()
_session.headers.update({"User-Agent": UA})


# --------------------------------------------------------------------------- #
# Stops catalogue
# --------------------------------------------------------------------------- #
def get_stops(force=False):
    """Return [{'id': '58', 'name': 'Port Louis - Victoria Square'}, ...].

    Cached to disk; the catalogue is effectively static so we only refetch when
    the cache is missing or `force` is set.
    """
    if not force and os.path.exists(STOPS_CACHE):
        with open(STOPS_CACHE, encoding="utf-8") as fh:
            return json.load(fh)

    html = _session.get(BASE, timeout=30).text
    soup = BeautifulSoup(html, "html.parser")
    select = soup.find("select", {"name": "searchroute[from]"})
    stops = []
    for opt in select.find_all("option"):
        val = (opt.get("value") or "").strip()
        name = opt.get_text(strip=True)
        if val.isdigit() and name:
            stops.append({"id": val, "name": name})
    stops.sort(key=lambda s: s["name"].lower())

    os.makedirs(DATA_DIR, exist_ok=True)
    with open(STOPS_CACHE, "w", encoding="utf-8") as fh:
        json.dump(stops, fh, ensure_ascii=False, indent=0)
    return stops


# --------------------------------------------------------------------------- #
# Journey planning
# --------------------------------------------------------------------------- #
_STAGE_RE = re.compile(r"\((\d+)\.\s*stage(?:,\s*([\d.]+)\s*min)?\)", re.I)
_FREQ_RE = re.compile(r"every\s+(\d+)\s*min", re.I)
_TOTAL_TIME_RE = re.compile(r"~?\s*(\d+)\s*minutes.*?net:\s*(\d+)\s*minutes", re.I)
_PRICE_RE = re.compile(r"Total price:\s*([\d./\s]+?)\s*MUR", re.I)


def _parse_substages(descr_cell):
    """Pull the ordered stop list out of a bus leg's <div class="substages">."""
    div = descr_cell.find("div", class_="substages")
    stops = []
    if not div:
        return stops
    # Each stop is separated by <br/>; turn breaks into newlines then split.
    for br in div.find_all("br"):
        br.replace_with("\n")
    last = 0.0
    for raw in div.get_text("\n").split("\n"):
        line = re.sub(r"\s+", " ", raw).strip()
        if not line:
            continue
        m = _STAGE_RE.search(line)
        if m and m.group(2):  # "(N. stage, M.M min)"
            last = float(m.group(2))
        stage_min = last  # carry forward if a min is missing
        name = _STAGE_RE.sub("", line).strip()  # also strips bare "(N. stage)"
        if name:
            stops.append({"name": name, "stageMin": stage_min})
    return stops


def _parse_bus_leg(tr):
    type_div = tr.find("div", class_="type")
    style = (type_div.get("style") if type_div else "") or ""
    color_m = re.search(r"background:\s*([^;]+)", style)
    text_m = re.search(r"color:\s*([^;]+)", style)
    route = type_div.get_text(strip=True).replace("Bus", "").strip() if type_div else ""

    descr = tr.find("td", class_="descr")
    stops = _parse_substages(descr)
    frm = stops[0]["name"] if stops else ""
    to = stops[-1]["name"] if stops else ""

    # Frequency lives in the last <td>.
    freq = None
    tds = tr.find_all("td")
    if tds:
        fm = _FREQ_RE.search(tds[-1].get_text(" "))
        if fm:
            freq = int(fm.group(1))

    return {
        "type": "bus",
        "route": route,
        "color": color_m.group(1).strip() if color_m else "#2c7fb8",
        "textColor": text_m.group(1).strip() if text_m else "white",
        "from": frm,
        "to": to,
        "frequencyMin": freq,
        "stops": stops,
    }


def _parse_walk_leg(tr):
    descr = tr.find("td", class_="descr")
    text = descr.get_text(" ", strip=True) if descr else "Walk"
    to = re.sub(r"^Walk to\s*", "", text).strip()
    return {"type": "walk", "to": to, "text": text}


def _parse_route_block(route_div):
    """Parse one <div class="route"> into a structured itinerary option."""
    option = {"legs": [], "price": None, "totalTimeMin": None, "netTimeMin": None}

    info = route_div.find("div", class_="info_box")
    if info:
        info_text = info.get_text(" ", strip=True)
        tm = _TOTAL_TIME_RE.search(info_text)
        if tm:
            option["totalTimeMin"] = int(tm.group(1))
            option["netTimeMin"] = int(tm.group(2))
        pm = _PRICE_RE.search(info_text)
        if pm:
            parts = [p.strip() for p in pm.group(1).split("/") if p.strip()]
            nums = [int(re.sub(r"\D", "", p)) for p in parts if re.sub(r"\D", "", p)]
            if nums:
                keys = ["adult", "child", "senior"]
                option["price"] = {keys[i]: nums[i] for i in range(min(3, len(nums)))}

    table = route_div.find("table", class_="itinerary")
    if table:
        for tr in table.find_all("tr"):
            classes = tr.get("class") or []
            if "bus_entry" in classes:
                option["legs"].append(_parse_bus_leg(tr))
            elif "walk_entry" in classes:
                option["legs"].append(_parse_walk_leg(tr))
    return option


def search_payload(day, frm, to):
    return {
        "searchroute[day]": str(day),
        "searchroute[from]": str(frm),
        "searchroute[from_village]": "1",
        "searchroute[to]": str(to),
        "searchroute[to_village]": "1",
    }


def parse_plan(html):
    """Parse a /routing/request HTML page into a list of itinerary options."""
    soup = BeautifulSoup(html, "html.parser")
    options = []
    for route_div in soup.find_all("div", class_="route"):
        opt = _parse_route_block(route_div)
        if opt["legs"]:
            options.append(opt)
    return options


def plan(day, frm, to, session=None):
    """Return a structured itinerary between two stop ids.

    `day`: 0=Weekdays, 1=Saturdays, 2=Sundays & Public Holidays.
    Returns {'options': [...], 'from': id, 'to': id, 'day': day}. `options` is a
    list of itineraries; each has legs (bus/walk), fares and times.
    """
    sess = session or _session
    # The POST replies 302 -> /routing/request (result held in the session).
    resp = sess.post(
        BASE + "/", data=search_payload(day, frm, to), timeout=30, allow_redirects=True
    )
    options = parse_plan(resp.text)

    error = None
    if not options:
        body = BeautifulSoup(resp.text, "html.parser").get_text(" ", strip=True).lower()
        error = (
            "No route found between these stops."
            if any(x in body for x in ("no route", "sorry", "not found"))
            else "No itinerary could be parsed from the planner."
        )

    return {
        "from": str(frm),
        "to": str(to),
        "day": int(day),
        "options": options,
        "error": error,
    }


if __name__ == "__main__":
    # Smoke test: Albion (Terminus)=313 -> Curepipe Ian Palach North=31
    stops = get_stops()
    print(f"{len(stops)} stops loaded")
    result = plan(0, 313, 31)
    print(json.dumps(result, ensure_ascii=False, indent=2)[:3000])