Compare Ticket Prices Across Marketplaces with One API

"""Compare one event's live listings across ticket marketplaces.

    python compare.py "<event name or URL>" --quantity 2
    python compare.py "<event name or URL>" --quantity 2 --section 110

Set TICKETS_DEV_KEY to your API key. Accessible seating (wheelchair and
companion seats) is left out unless you pass --accessible.
"""
import argparse
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor

import requests

API = "https://api.tickets.dev/v1"
HEADERS = {"x-api-key": os.environ["TICKETS_DEV_KEY"]}

# Marketplaces name some sections differently: the same floor section can be
# "Floor 5" on one and "5" on another. Map names you have checked to one key,
# per marketplace. Anything unmapped is compared by its own name, lowercased.
SECTION_ALIASES = {
    # ("vividseats", "floor 5"): "5",
}


def find_event(query):
    """Step 1: one event, with its URL on every marketplace that lists it."""
    response = requests.get(
        f"{API}/events", params={"query": query}, headers=HEADERS, timeout=30
    )
    if not response.ok:
        raise SystemExit(f"Event lookup failed: {error_code(response)}")
    events = response.json()["events"]
    if not events:
        raise SystemExit(f"No upcoming event matches {query!r}")
    if len(events) > 1:
        print(f"{len(events)} events match, using the first. "
              "Pass a marketplace event URL to pick one exactly.")
    return events[0]


def error_code(response):
    try:
        return response.json()["error"]["code"]
    except (ValueError, KeyError, TypeError):
        return f"HTTP {response.status_code}"


def capture(source):
    """Step 2: one capture is one event on one marketplace. Retries what the
    API marks retryable, such as 429 rate_limited when capacity is full, and
    never raises, so one marketplace failing doesn't take the others down."""
    marketplace = source["marketplace"]
    wait = 0
    for attempt in range(3):
        time.sleep(wait)
        try:
            response = requests.get(
                f"{API}/capture", params={"url": source["url"]}, headers=HEADERS,
                timeout=120,
            )
        except requests.RequestException as error:
            failure, wait = type(error).__name__, 10
            continue
        if response.ok:
            return {"marketplace": marketplace, "snapshot": response.json()}
        failure = error_code(response)
        try:
            retryable = response.json()["error"]["retryable"]
        except (ValueError, KeyError, TypeError):
            retryable = False
        if not retryable:
            break
        retry_after = response.headers.get("Retry-After", "")
        wait = int(retry_after) if retry_after.isdigit() else 10
    return {"marketplace": marketplace, "error": failure}


def sellable(listing):
    """The quantities a buyer can actually buy. A listing of 3 that sells
    as "1,3" can't sell you 2. With no split rules published, assume only
    the whole listing."""
    quantities = listing["sellableQuantities"].replace(" ", "").split(",")
    return {int(q) for q in quantities if q.isdigit()} or {listing["quantity"]}


ACCESSIBLE = re.compile(r"accessib|wheelchair|companion", re.IGNORECASE)


def accessible(listing):
    """Wheelchair and companion seats: row WC, or said so in the listing."""
    text = f"{listing.get('ticketType') or ''} {listing.get('listingNotes') or ''}"
    return listing["row"].strip().upper() == "WC" or bool(ACCESSIBLE.search(text))


def normalize(marketplace, snapshot):
    """Step 3: one row per listing, with the fields the comparison needs."""
    rows = []
    for listing in snapshot["listings"]:
        section = listing["section"].strip().casefold()
        total = listing["totalPrice"]
        rows.append({
            "marketplace": marketplace,
            "section": listing["section"],
            "section_key": SECTION_ALIASES.get((marketplace, section), section),
            "row": listing["row"],
            "quantities": sellable(listing),
            # All-in per ticket. A missing total stays missing: falling back
            # to ticketPrice would rank a pre-fee price against all-in ones.
            "total": total if isinstance(total, (int, float)) and total > 0 else None,
            "fee_itemized": listing["fee"] > 0,
            "accessible": accessible(listing),
            "currency": snapshot["currency"],
        })
    return rows


def matches(row, quantity, section, row_label, include_accessible=False):
    """Step 4: listings that can sell this many tickets, where you want them."""
    if row["accessible"] and not include_accessible:
        return False
    if quantity not in row["quantities"]:
        return False
    if section and row["section_key"] != section.strip().casefold():
        return False
    if row_label and row["row"].strip().casefold() != row_label.strip().casefold():
        return False
    return True


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("event", help="event name, or any marketplace event URL")
    parser.add_argument("--quantity", type=int, default=2)
    parser.add_argument("--section")
    parser.add_argument("--row")
    parser.add_argument("--top", type=int, default=10)
    parser.add_argument("--accessible", action="store_true",
                        help="include wheelchair and companion seats")
    args = parser.parse_args()

    event = find_event(args.event)
    print(f"{event['name']} | {event['venue']['name']} | {event['eventDateLocal']}")

    # Plans cap how many captures run at once, so keep the pool small.
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(capture, event["sources"]))

    comparable = []
    print(f"\n{'Marketplace':<13} {'Captured (UTC)':<20} Matching listings")
    for result in results:
        marketplace = result["marketplace"]
        if "error" in result:
            print(f"{marketplace:<13} {'unavailable':<20} {result['error']}")
            continue
        snapshot = result["snapshot"]
        rows = [r for r in normalize(marketplace, snapshot)
                if matches(r, args.quantity, args.section, args.row, args.accessible)]
        priced = [r for r in rows if r["total"] is not None]
        comparable += priced
        detail = f"{len(rows)}"
        if len(priced) < len(rows):
            detail += f" ({len(rows) - len(priced)} without an all-in price)"
        if not snapshot["listings"] and snapshot["note"]:
            detail += f" | {snapshot['note'][:60]}"
        print(f"{marketplace:<13} {snapshot['capturedAt'][:19]:<20} {detail}")

    if not comparable:
        print("\nNo listing on any marketplace matches those filters.")

    # Step 5: rank by all-in price, one currency at a time.
    for currency in sorted({r["currency"] for r in comparable}):
        best = sorted((r for r in comparable if r["currency"] == currency),
                      key=lambda r: r["total"])[:args.top]
        print(f"\nCheapest {args.quantity} tickets, all-in, {currency}")
        print(f"{'Marketplace':<13} {'Section':<9} {'Row':<5} "
              f"{'Per ticket':>11} {f'For {args.quantity}':>11}  Fee itemized")
        for r in best:
            print(f"{r['marketplace']:<13} {r['section']:<9} {r['row']:<5} "
                  f"{r['total']:>11,.2f} {r['total'] * args.quantity:>11,.2f}  "
                  f"{'yes' if r['fee_itemized'] else 'no, included'}")


if __name__ == "__main__":
    main()