#!/usr/bin/env python3
"""Thin client for the ExpertFlyer API service.

The browser automation, the ExpertFlyer credential and the minted session all
live in `jbaruch/expertflyer-api`, a service container that runs no LLM. This
container holds none of them — it makes HTTP calls and relays the answers.

Stdlib only: the work here is a request and an error mapping, which does not
earn a dependency.

Output: one JSON object on stdout. Exit non-zero on failure, with the service's
own diagnostic on stderr.
"""

from __future__ import annotations

import argparse
import json
import os
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import seat_quality  # noqa: E402

URL_ENV = "EXPERTFLYER_API_URL"
TOKEN_ENV = "EXPERTFLYER_API_TOKEN"
# The service runs beside the agent container on the same host. Address the
# docker bridge gateway by IP, NOT host.docker.internal: that alias sits in
# nanoclaw's AGENT_PROXY_BYPASS_HOSTS, so a request to the hostname skips the
# OneCLI gateway — and the gateway is what swaps the real bearer in for the
# `onecli-managed` placeholder this container holds. Using the hostname would
# send the placeholder through unswapped and earn a 401.
DEFAULT_URL = "http://172.17.0.1:8090"
TIMEOUT_SECONDS = 180


def _base_url() -> str:
    return os.environ.get(URL_ENV, DEFAULT_URL).rstrip("/")


def _request(method: str, path: str, params: dict | None = None, body: dict | None = None):
    url = f"{_base_url()}{path}"
    if params:
        cleaned = {k: v for k, v in params.items() if v is not None}
        url = f"{url}?{urllib.parse.urlencode(cleaned)}"

    data = json.dumps(body).encode() if body is not None else None
    request = urllib.request.Request(url, data=data, method=method)
    if data is not None:
        request.add_header("Content-Type", "application/json")
    token = os.environ.get(TOKEN_ENV)
    if token:
        request.add_header("Authorization", f"Bearer {token}")

    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as exc:
        # The service distinguishes an expired session (503) from the upstream
        # bot wall (502); both look like 403 to it, so the distinction is not
        # re-derivable here. Relay it rather than flattening it.
        raw = exc.read().decode(errors="replace")
        try:
            detail = json.loads(raw).get("detail", raw)
        except json.JSONDecodeError:
            detail = raw
        if isinstance(detail, dict):
            return {"error": detail.get("error", "upstream"), "detail": detail.get("detail", raw)}
        return {"error": "upstream", "detail": detail, "status": exc.code}
    except urllib.error.URLError as exc:
        # A TLS verification failure means the service ANSWERED and its
        # certificate could not be validated — telling the operator to check
        # whether it is running sends them to the wrong place entirely.
        if isinstance(exc.reason, ssl.SSLCertVerificationError):
            return {
                "error": "tls",
                "detail": (
                    f"TLS verification failed for {_base_url()} "
                    f"({exc.reason.verify_message or exc.reason.reason}) — the service "
                    "responded but its certificate chain could not be verified. On a "
                    "host whose Python does not read the system trust store (macOS), "
                    "point SSL_CERT_FILE at the system CA bundle: "
                    "SSL_CERT_FILE=/etc/ssl/cert.pem on macOS, "
                    "/etc/ssl/certs/ca-certificates.crt on Debian"
                ),
            }
        return {
            "error": "unreachable",
            "detail": (
                f"ExpertFlyer API not reachable at {_base_url()} ({exc.reason}) — "
                f"check the service is running and {URL_ENV} points at it"
            ),
        }


# The schedule stamps UTC, so a departure late in the local evening falls on
# the next UTC day and the service finds no such flight. Retrying the previous
# day is fixed logic, not judgement, so it lives here rather than in the skill.
ROUTE_UNRESOLVED = "could not resolve a route"


def _previous_day(date: str) -> str:
    return (datetime.strptime(date, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")


def _looks_like_wrong_date(result: dict) -> bool:
    detail = str(result.get("detail", "")).lower()
    return "error" in result and ROUTE_UNRESOLVED in detail


def _rank(result: dict) -> dict:
    """Order the service's bookable seats by the operator's preferences.

    The service reports what each seat IS; ranking decides what it is WORTH,
    which is why it happens here rather than upstream. `matching` is the
    service's own criteria filter and is left untouched.
    """
    if "error" in result or not isinstance(result.get("seats"), list):
        return result
    cabin = result.get("cabin")
    # Recline is derived from the cabin's exit-row layout, so descriptions need
    # the same tiers ranking used. Without them a derived reclining row renders
    # as a plain "exit row", because service seats carry no recline field.
    # `exit_rows` is the cabin's full layout when the service supplies it.
    # Without it the seats list holds bookable seats only, so an occupied rear
    # exit row is invisible and nothing is claimed to recline.
    layout = result.get("exit_rows")
    try:
        tiers = seat_quality.exit_tiers(result["seats"], layout)
        ranked = seat_quality.rank_seats(result["seats"], cabin, layout)
        result["ranked"] = [{**s, "why": seat_quality.describe(s, cabin, tiers)} for s in ranked]
        best = ranked[0] if ranked else None
        result["best"] = seat_quality.describe(best, cabin, tiers) if best else None
        # Every open seat may still be unacceptable — a cabin of middles ranks
        # to nothing even though `available_total` is non-zero.
        result["acceptable_total"] = len(ranked)
    except seat_quality.SeatQualityError as exc:
        # A seat the ranker refuses to order is a reportable answer, not a
        # traceback: the service replied, and the operator needs to read WHICH
        # seat could not be ranked. Dropping `best`/`ranked` is deliberate —
        # a partial ranking would read as a complete one.
        result.pop("ranked", None)
        result.pop("best", None)
        result.pop("acceptable_total", None)
        result["error"] = "unrankable"
        result["detail"] = str(exc)
    return result


# How far up the cabin ladder to look by default. Each rung is one more request
# to a bot-walled service, and one rung already covers the case a single-cabin
# check structurally cannot see: a Comfort+ seat opening while the operator
# sits in Main.
DEFAULT_SCAN_RUNGS = 1

# How far up the ladder an alert offer reaches from the held cabin. A wide
# sweep sees what exists; it does not widen what the operator would actually
# move into, and an alert on a cabin they cannot buy into is noise.
ALERT_RUNGS = 1

VERDICT_OPTIMAL = "optimal"
VERDICT_UPGRADE = "upgrade"
VERDICT_NO_HELD_SEAT = "no_held_seat"
VERDICT_POSITION_UNKNOWN = "held_position_unknown"
VERDICT_NOTHING_OPEN = "nothing_open"
VERDICT_CABIN_MISMATCH = "held_cabin_mismatch"
VERDICT_CABIN_UNRESOLVED = "held_cabin_unresolved"

# Why the cabin could not be read off the layout. Each takes the operator
# somewhere different, so the verdict alone does not say enough.
REASON_SHARED_ROW = "shared_row"
REASON_NO_SUCH_ROW = "no_such_row"
REASON_ROWS_UNAVAILABLE = "rows_unavailable"


def _has_row(response: dict, row: int) -> bool:
    """Whether any bookable seat in this cabin's response sits in `row`."""
    return any(int(seat.get("row", -1)) == row for seat in response.get("seats", []))


def _held_seat(label: str, cabin: str, position: str | None, cabin_seats, exit_rows) -> dict:
    """The seat already occupied, shaped so it can be ranked against open ones.

    The service reports bookable seats, so the held seat is absent from every
    response by definition — it is occupied, by the operator. Its row and
    column come from the designator; whether it is an exit row comes from the
    cabin's layout; its position is either stated or read off the columns of
    the open seats around it.
    """
    row, column = seat_quality.parse_seat_label(label)
    source = "stated"
    if position is None:
        position = seat_quality.column_positions(cabin_seats).get(column)
        source = "seat-map"
    return {
        "label": f"{row}{column}",
        "row": row,
        "column": column,
        # None when the column is stated nowhere the sweep can read. The caller
        # reports that rather than ranking a seat whose position it invented.
        "position": position,
        "position_source": source if position else None,
        "cabin": cabin,
        "isExitRow": int(row) in {int(r) for r in (exit_rows or [])},
    }


def parse_args(argv=None):
    p = argparse.ArgumentParser(description="Query the ExpertFlyer API service.")
    sub = p.add_subparsers(dest="action", required=True)

    seats = sub.add_parser("seats", help="Bookable seats matching the wanted positions")
    seats.add_argument("--airline", required=True)
    seats.add_argument("--flight", required=True)
    seats.add_argument("--date", required=True, help="YYYY-MM-DD")
    seats.add_argument("--cabin", required=True, help="e.g. 'premium economy', 'comfort+', W")
    seats.add_argument("--want", default="non-middle")
    seats.add_argument("--origin")
    seats.add_argument("--destination")
    seats.add_argument(
        "--date-fallback",
        action="store_true",
        help=(
            "Retry the previous day when the flight is not found on --date. "
            "Only for dates derived from the UTC travel schedule, where a late "
            "local departure lands on the next UTC day. Off by default: for a "
            "date the operator named, a flight that does not operate that day "
            "must report that, not seats from another day's flight."
        ),
    )

    assess = sub.add_parser("assess", help="Judge the held seat against everything open")
    assess.add_argument("--airline", required=True)
    assess.add_argument("--flight", required=True)
    assess.add_argument("--date", required=True, help="YYYY-MM-DD")
    assess.add_argument(
        "--held",
        help=(
            "The seat currently assigned, e.g. 21F. Without it there is no "
            "verdict: an open seat can only be better or worse than something."
        ),
    )
    assess.add_argument(
        "--held-cabin",
        help=(
            "The cabin the held seat is in, e.g. 'comfort+', 'first', W. Omit "
            "it and the cabin is resolved from the aircraft's own row extents, "
            "which is what the operator would otherwise have to look up."
        ),
    )
    assess.add_argument(
        "--held-position",
        choices=("window", "aisle", "middle"),
        help=(
            "Whether the held seat is a window, an aisle or a middle. Omit it "
            "and the column is read off the open seats in the same cabin; that "
            "fails when no open seat shares the column, which reports "
            "held_position_unknown rather than guessing."
        ),
    )
    assess.add_argument(
        "--scan-up",
        type=int,
        default=DEFAULT_SCAN_RUNGS,
        help=(
            "How many cabins above the held one to include. Each is one more "
            f"request to a bot-walled service. Default {DEFAULT_SCAN_RUNGS}; "
            "0 checks the held cabin alone."
        ),
    )
    assess.add_argument("--origin")
    assess.add_argument("--destination")
    assess.add_argument("--date-fallback", action="store_true")

    fare = sub.add_parser("fare-class", help="Fare-class inventory for a flight")
    fare.add_argument("--origin", required=True)
    fare.add_argument("--destination", required=True)
    fare.add_argument("--date", required=True, help="YYYY-MM-DD")
    fare.add_argument("--airline", required=True)
    fare.add_argument("--class", dest="fare_class", required=True)
    fare.add_argument("--flight")
    fare.add_argument("--include-codeshares", action="store_true")

    sub.add_parser("alerts", help="Every alert on the account")

    create = sub.add_parser("create-alert", help="Create a seat or fare-class alert")
    create.add_argument("--kind", required=True, choices=("seat", "fare-class"))
    create.add_argument("--airline", required=True)
    create.add_argument("--flight", required=True)
    create.add_argument("--date", required=True, help="YYYY-MM-DD")
    create.add_argument("--origin", required=True)
    create.add_argument("--destination", required=True)
    create.add_argument("--cabin")
    create.add_argument("--want", default="non-middle")
    create.add_argument("--class", dest="fare_class")
    create.add_argument("--force", action="store_true")

    delete = sub.add_parser("delete-alert", help="Delete one alert by id")
    delete.add_argument("--id", dest="alert_id", required=True, type=int)

    return p.parse_args(argv)


def _seats_in_cabin(args, cabin: str, want: str) -> dict:
    """One cabin's bookable seats, with the schedule-date retry applied.

    Shared by `seats` and `assess`: the previous-day retry is a property of a
    schedule-derived date, not of which command asked.
    """

    def seats_on(date: str) -> dict:
        return _request(
            "GET",
            "/seats",
            {
                "airline": args.airline,
                "flight": args.flight,
                "date": date,
                "cabin": cabin,
                "want": want,
                "origin": args.origin,
                "destination": args.destination,
            },
        )

    result = seats_on(args.date)
    # Opt-in only. The retry is sound when the date came from the UTC
    # schedule; against a date the operator named it would answer about a
    # different day's flight and present it as the requested one.
    if not (_looks_like_wrong_date(result) and args.date_fallback):
        return result
    try:
        fallback = _previous_day(args.date)
    except ValueError:
        return {
            "error": "bad_request",
            "detail": (
                f"--date {args.date!r} is not YYYY-MM-DD, so the "
                "previous-day retry cannot be computed — pass the "
                "departure date as e.g. 2026-08-31"
            ),
        }
    retried = seats_on(fallback)
    # Report the retry's own outcome. Returning the first error instead
    # would hide what actually went wrong the second time — an expired
    # session or an unreachable service reported as "no such flight".
    if "error" not in retried:
        retried["date_fallback_applied"] = fallback
    else:
        retried["date_fallback_attempted"] = fallback
    return retried


def _assess(args) -> dict:
    """Judge the seat already held against everything open worth moving to.

    This is the question the operator actually asks — "are my seats the best"
    — and it is not the one a cabin scan answers. A cabin scan reports what is
    open; only a comparison against the held seat reports whether any of it is
    better. Without the held seat there is no verdict to give, so the absence
    is reported rather than answered around.
    """
    if not args.held:
        return {
            "verdict": VERDICT_NO_HELD_SEAT,
            "detail": (
                "no seat given, so nothing can be called better or worse than it — "
                "pass --held with the seat currently assigned, e.g. --held 21F"
            ),
        }
    try:
        row, column = seat_quality.parse_seat_label(args.held)
        held_cabin = seat_quality.cabin_code(args.held_cabin) if args.held_cabin else None
    except seat_quality.SeatQualityError as exc:
        return {"error": "bad_request", "detail": str(exc)}

    scanned: dict[str, dict] = {}
    absent: list[str] = []
    fetched: dict[str, dict] = {}

    def fetch(cabin: str) -> dict:
        if cabin not in fetched:
            # `want=any`: the criteria filter shapes `matching`, and a
            # comparison against the held seat has to see every open seat, not
            # the subset that already matched a wanted position.
            fetched[cabin] = _seats_in_cabin(args, cabin, "any")
        return fetched[cabin]

    def requested_so_far() -> list[str]:
        """The cabins this run has asked for, best first.

        Resolution has no planned list — finding the cabin is what it is for —
        so the contract's `cabins_requested` is what it got through.
        """
        return sorted(fetched, key=lambda c: seat_quality.CABIN_SCORE[c], reverse=True)

    cabin_source = "stated"
    if held_cabin is None:
        # The seat's cabin is a fact about the aircraft, not something the
        # operator should have to look up. `rows` is every row of a cabin, sold
        # out or not, so the cabin holding this row can be found by reading
        # from the bottom of the ladder up — where most seats are, so the
        # common case costs one request.
        cabin_source = "resolved"
        holders: list[str] = []
        for candidate in reversed(seat_quality.cabins_at_or_above(seat_quality.MAIN_CABIN, 4)):
            response = fetch(candidate)
            if "error" in response:
                return {
                    "error": response["error"],
                    "detail": f"{candidate}: {response.get('detail', response['error'])}",
                    "cabin_failed": candidate,
                    "cabins_requested": requested_so_far(),
                }
            if response.get("cabin_present") is False:
                absent.append(candidate)
                continue
            rows = response.get("rows")
            if rows is None:
                return {
                    "verdict": VERDICT_CABIN_UNRESOLVED,
                    "reason": REASON_ROWS_UNAVAILABLE,
                    "cabins_absent": absent,
                    "detail": (
                        f"{candidate} reports no rows, so the cabin holding seat "
                        f"{args.held!r} cannot be read off the layout. Pass --held-cabin."
                    ),
                }
            if row in {int(r) for r in rows}:
                holders.append(candidate)
                # A cabin boundary can fall mid-row — the 739's Comfort+ ends a
                # row later on the right — so one row number belongs to two
                # ADJACENT cabins. Reading on past the first match is what
                # tells them apart; anything further up cannot share this row.
                above = seat_quality.cabins_above(candidate)
                if not above:
                    break
                neighbour_cabin = above[-1]
                neighbour = fetch(neighbour_cabin)
                # A neighbour that did not answer is not evidence the row is
                # unshared. Reading it that way assigns the seat to the lower
                # cabin off a failure, and at --scan-up 0 the sweep never
                # fetches it again to notice.
                if "error" in neighbour:
                    return {
                        "error": neighbour["error"],
                        "detail": (
                            f"{neighbour_cabin}: {neighbour.get('detail', neighbour['error'])} "
                            f"— read to check whether row {row} is shared with {candidate}"
                        ),
                        "cabin_failed": neighbour_cabin,
                        "cabins_requested": requested_so_far(),
                    }
                if neighbour.get("cabin_present") is not False:
                    if neighbour.get("rows") is None:
                        return {
                            "verdict": VERDICT_CABIN_UNRESOLVED,
                            "reason": REASON_ROWS_UNAVAILABLE,
                            "cabins_absent": absent,
                            "detail": (
                                f"{neighbour_cabin} reports no rows, so whether row {row} is "
                                f"shared with {candidate} cannot be established. Pass "
                                "--held-cabin."
                            ),
                        }
                    if row in {int(r) for r in neighbour["rows"]}:
                        holders.append(neighbour_cabin)
                break
        if len(holders) != 1:
            return {
                "verdict": VERDICT_CABIN_UNRESOLVED,
                "reason": REASON_SHARED_ROW if holders else REASON_NO_SUCH_ROW,
                "cabins_absent": absent,
                "row_in_cabins": holders,
                "detail": (
                    (
                        f"row {row} runs through {' and '.join(holders)} on this aircraft, so "
                        f"which cabin holds seat {args.held!r} cannot be read off the layout"
                    )
                    if holders
                    else (
                        f"no cabin on this aircraft has a row {row}, so seat {args.held!r} is "
                        "not on it — check the seat and the flight"
                    )
                )
                + ". Pass --held-cabin to assess it against a named cabin.",
            }
        held_cabin = holders[0]

    cabins = seat_quality.cabins_at_or_above(held_cabin, args.scan_up)
    absent = [c for c in absent if c in cabins]
    for cabin in cabins:
        # `want=any`: the criteria filter shapes `matching`, and a comparison
        # against the held seat has to see every open seat, not the subset that
        # already matched a wanted position.
        response = fetch(cabin)
        if "error" in response:
            # A cabin that failed to load could be holding the upgrade, so a
            # partial sweep must not report "nothing better is open".
            return {
                "error": response["error"],
                "detail": f"{cabin}: {response.get('detail', response['error'])}",
                "cabin_failed": cabin,
                "cabins_requested": cabins,
            }
        if response.get("cabin_present") is False:
            absent.append(cabin)
            continue
        scanned[cabin] = response

    if held_cabin not in scanned:
        return {
            "error": "bad_request",
            "detail": (
                f"the aircraft has no {held_cabin} cabin, so seat {args.held!r} "
                "cannot be in it — check the cabin the operator actually flies"
            ),
            "cabins_absent": absent,
        }

    held = _held_seat(
        args.held,
        held_cabin,
        args.held_position,
        scanned[held_cabin].get("seats", []),
        scanned[held_cabin].get("exit_rows"),
    )
    open_seats = [seat for response in scanned.values() for seat in response.get("seats", [])]
    common = {
        "flight": scanned[held_cabin].get("flight"),
        "route": scanned[held_cabin].get("route"),
        "cabins_scanned": sorted(scanned, key=lambda c: seat_quality.CABIN_SCORE[c], reverse=True),
        "cabins_absent": absent,
        # `optimal` is only ever true of the cabins actually read. Naming the
        # ones the sweep stopped short of keeps the verdict from being heard
        # as "nothing anywhere on this aircraft beats your seat".
        "cabins_unscanned": seat_quality.cabins_above(cabins[0]),
        "seats_compared": len(open_seats),
        # How the cabin was arrived at, distinct from how it was corroborated:
        # "stated" came from --held-cabin, "resolved" was read off the row
        # extents so the operator never had to know it.
        "held_cabin_from": cabin_source,
        # Per cabin, how many open seats were worth taking at all. `optimal`
        # says nothing open beat the held seat; it never says the held seat
        # outranks a cabin. A cabin at 0 had nothing to beat it WITH, and that
        # distinction is the difference between "Comfort+ was empty" and the
        # false "this seat beats Comfort+".
        "acceptable_by_cabin": {
            code: len(seat_quality.rank_seats(response.get("seats", []), code))
            for code, response in scanned.items()
        },
    }

    # The held seat is occupied and never appears in a response. Its row can
    # still show up — under another passenger's seat in the same row — and that
    # is worth surfacing. It is NOT worth refusing on: `/seats` reports bookable
    # seats, so a row whose every seat is taken is missing from the held cabin's
    # response while still being in it. Cabins also split mid-row (the 739's
    # Comfort+ ends a row later on the right), so one row number legitimately
    # appears in two cabins. Absence here is not disproof of membership, and
    # disproving it needs the cabin's row layout the service does not yet
    # report (jbaruch/expertflyer-api#20).
    # Rendered before the early returns: a response that carries `held` with a
    # known position carries its description too. Only an unknown position has
    # nothing to render, and that shape says so.
    layout = sorted({int(r) for c in scanned.values() for r in (c.get("exit_rows") or [])})
    if held["position"] is not None:
        try:
            held["why"] = seat_quality.describe(
                held, None, seat_quality.exit_tiers([held], layout or None)
            )
        except seat_quality.SeatQualityError as exc:
            return {**common, "error": "unrankable", "detail": str(exc), "held": held}

    held_rows = scanned[held_cabin].get("rows")
    if held_rows is not None:
        # The cabin's full extent, sold out or not. Membership is decidable
        # from the held cabin's own response, so nothing else has to be read.
        common["held_cabin_source"] = "rows"
        common["held_cabin_corroborated"] = row in {int(r) for r in held_rows}
        if not common["held_cabin_corroborated"]:
            elsewhere = sorted(
                code
                for code, response in scanned.items()
                if code != held_cabin and row in {int(r) for r in (response.get("rows") or [])}
            )
            extent = f"rows {min(held_rows)}-{max(held_rows)}" if held_rows else "no rows"
            return {
                **common,
                "verdict": VERDICT_CABIN_MISMATCH,
                "held": held,
                "detail": (
                    f"seat {args.held!r} was assessed as a {held_cabin} seat, and {held_cabin} "
                    f"runs {extent} on this aircraft, so row {row} is not in it"
                    + (f" — it is in {', '.join(elsewhere)}" if elsewhere else "")
                    + ". Re-run with the right --held-cabin. The cabin decides the ladder rung "
                    "and the exit-row layout, so every part of the verdict depends on it."
                ),
            }
    else:
        # A service without `rows` (expertflyer-api before #20) can corroborate
        # and never disprove: `/seats` reports bookable seats, so a row whose
        # every seat is taken is missing from the cabin it is in.
        common["held_cabin_source"] = "seats"
        common["held_cabin_corroborated"] = _has_row(scanned[held_cabin], row) or None
        if common["held_cabin_corroborated"] is None:
            # A hint for the agent to confirm the cabin, never a verdict.
            common["row_seen_in"] = sorted(
                code
                for code, response in scanned.items()
                if code != held_cabin and _has_row(response, row)
            )

    # A sweep that saw nothing has nothing to compare. `optimal` off an empty
    # evidence base is true the way "no counterexample was found" is true after
    # looking in no drawers — and it reads as a comparison that happened.
    if not open_seats:
        return {
            **common,
            "verdict": VERDICT_NOTHING_OPEN,
            "held": held,
            "detail": (
                f"no open seat was found in {', '.join(common['cabins_scanned'])}, so nothing was "
                f"compared against seat {args.held!r} — there is no seat to move to, better or "
                "worse. Widen the sweep with --scan-up, or check the cabin the operator flies: a "
                f"sold-out {held_cabin} and a seat that is not in {held_cabin} look identical here."
            ),
        }
    if held["position"] is None:
        return {
            **common,
            "verdict": VERDICT_POSITION_UNKNOWN,
            "held": held,
            "detail": (
                f"seat {args.held!r} is a {held_cabin} seat in column {column} of row {row}, "
                f"and no open seat in that cabin sits in column {column} — so whether it is a "
                "window, an aisle or a middle cannot be read off the seat map. Pass "
                "--held-position window|aisle|middle."
            ),
        }

    # Exit rows are numbered on the aircraft, not per cabin, so recline is
    # derived from every layout the sweep saw rather than one cabin's slice.
    try:
        beats_held = [
            seat for seat in open_seats if seat_quality.is_upgrade(seat, held, None, layout or None)
        ]
        tiers = seat_quality.exit_tiers(beats_held, layout or None)

        def describe_all(seats):
            ranked = seat_quality.rank_seats(seats, None, layout or None)
            return [{**s, "why": seat_quality.describe(s, None, tiers)} for s in ranked]

        # A seat in the cabin the operator is ticketed into can be selected in
        # the airline's app. One in a better cabin cannot: it is a fare change
        # or an upgrade clearance, which is Step 1's question, not a seat
        # change. Reporting both as "upgrades" tells the operator to go take a
        # seat that is not theirs to take.
        described = describe_all(
            [s for s in beats_held if seat_quality.seat_cabin(s, None) == held_cabin]
        )
        openings = describe_all(
            [s for s in beats_held if seat_quality.seat_cabin(s, None) != held_cabin]
        )
    except seat_quality.SeatQualityError as exc:
        return {**common, "error": "unrankable", "detail": str(exc), "held": held}

    # An alert is worth setting on the cabins the operator would actually move
    # into. A wide sweep is for seeing what exists; it does not widen what is
    # worth watching.
    #
    # Check first, alert only if absent: a cabin already holding a seat worth
    # taking has nothing to wait for, and a watch on it fires the moment it is
    # created. Only a cabin with nothing acceptable open is worth watching.
    watchable = [
        cabin
        for cabin in seat_quality.cabins_at_or_above(held_cabin, ALERT_RUNGS)
        if cabin in scanned and common["acceptable_by_cabin"].get(cabin, 0) == 0
    ]

    return {
        **common,
        # `optimal` is scoped to `cabins_scanned`, never to the whole aircraft,
        # and to seats the operator can select — a better cabin is reported
        # separately because taking it is a different transaction.
        "verdict": VERDICT_UPGRADE if described else VERDICT_OPTIMAL,
        "held": held,
        "upgrades": described,
        "best_upgrade": described[0]["why"] if described else None,
        "cabin_openings": openings,
        # Nothing selectable beats the held seat, so watching is the only move
        # left. A seat worth taking is taken now, not watched — and there has
        # to be a cabin worth watching.
        "alert_recommended": not described and bool(watchable),
        "alert_cabins": sorted(watchable, key=lambda c: seat_quality.CABIN_SCORE[c], reverse=True),
    }


def run(args) -> dict:
    if args.action == "seats":
        return _rank(_seats_in_cabin(args, args.cabin, args.want))
    if args.action == "assess":
        return _assess(args)
    if args.action == "fare-class":
        return _request(
            "GET",
            "/fare-class",
            {
                "origin": args.origin,
                "destination": args.destination,
                "date": args.date,
                "airline": args.airline,
                "class": args.fare_class,
                "flight": args.flight,
                "include_codeshares": str(args.include_codeshares).lower(),
            },
        )
    if args.action == "alerts":
        return _request("GET", "/alerts")
    if args.action == "create-alert":
        return _request(
            "POST",
            "/alerts",
            body={
                "kind": args.kind,
                "airline": args.airline,
                "flight": args.flight,
                "date": args.date,
                "origin": args.origin,
                "destination": args.destination,
                "cabin": args.cabin,
                "want": args.want,
                "class": args.fare_class,
                "force": args.force,
            },
        )
    if args.action == "delete-alert":
        return _request("DELETE", f"/alerts/{args.alert_id}")
    raise ValueError(f"unknown action {args.action!r}")


# Verdicts that decline to answer. They are not service faults, but treating
# them as success invites the caller to read a missing verdict as "nothing
# better is open" — the exact misreport this command exists to prevent.
UNANSWERED = frozenset(
    {
        VERDICT_NO_HELD_SEAT,
        VERDICT_POSITION_UNKNOWN,
        VERDICT_NOTHING_OPEN,
        VERDICT_CABIN_MISMATCH,
        VERDICT_CABIN_UNRESOLVED,
    }
)


def main(argv=None) -> int:
    result = run(parse_args(argv))
    print(json.dumps(result))
    if "error" in result or result.get("verdict") in UNANSWERED:
        print(f"expertflyer: {result.get('detail', result.get('error'))}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())