API Reference

def clear_token_cache() -> None:
    """Delete any cached OAuth tokens."""
    path = _cache_path()
    if path.exists():
        path.unlink()

def get_bearer_token(config: Config) -> str:
    """Return a valid Bearer token for the ServiceNow API.

    Reads from the token cache, refreshes silently if possible, and falls
    back to the full browser-based Authorization Code + PKCE flow when needed.

    Parameters
    ----------
    config : Config
        Must have ``servicenow_client_id``, ``servicenow_client_secret``,
        ``servicenow_oauth_scope``, and ``servicenow_oauth_redirect_port`` set.

    Returns
    -------
    str
        A valid access token string.
    """
    cached = _load_cache()

    if cached and not _token_is_expired(cached):
        return cached["access_token"]

    if cached and cached.get("refresh_token"):
        refreshed = _refresh_access_token(config, cached["refresh_token"])
        if refreshed:
            _save_cache(refreshed)
            return refreshed["access_token"]

    ## Full browser flow
    token = _run_authorization_code_flow(config)
    _save_cache(token)
    return token["access_token"]

def get_id_token() -> str | None:
    """Return the id_token from the cached OAuth token response, or None if absent.

    Requires the ``openid`` scope to have been requested during authentication.
    """
    cached = _load_cache()
    return cached.get("id_token") if cached else None

@cli.command()
def browse_ticket(ctx: typer.Context, ticket_number: str):
    """Opens a ticket in your default the browser (e.g. RES0001234)."""
    url = command.get_ticket_url(ticket_number=ticket_number, config=_config(ctx))
    print(f"Opening ticket: {url}")
    webbrowser.open(url)

@cli.command()
def display_all(ctx: typer.Context, long_list: bool = False):
    """Display all unassigned tickets from ServiceNow."""
    command.display_all_tickets(long_list=long_list, config=_config(ctx))

@cli.command()
def display_ticket(ctx: typer.Context, ticket_number: str):
    """Display the details of a specific ticket (e.g. RES0001234)."""
    command.display_single_ticket(ticket_number=ticket_number, config=_config(ctx))

@cli.command()
def evaluate_all(
    ctx: typer.Context,
    in_dir: Annotated[Path, typer.Argument(help="Directory of pre-downloaded HTML ticket files.")],
    out_dir: Annotated[Path, typer.Argument(help="Directory to write the JSON results file.")],
    num_runs: Annotated[int, typer.Option(help="Number of LLM runs per ticket (for majority-vote analysis).")] = 1,
):
    """Evaluate LLM performance against a directory of pre-downloaded HTML tickets.

    The output filename is derived from the configured LLM model name and the
    number of runs, e.g. ``gpt-4o_3runs.json``.
    """
    if not out_dir.exists():
        msg = f"Output directory does not exist: {out_dir}"
        raise typer.BadParameter(msg, param_hint="out_dir")

    config = _config(ctx)
    ## Sanitise model name for use in a filename (replace / and : which appear in some model IDs)
    safe_model = config.llm_model.replace("/", "-").replace(":", "-")
    out_file = out_dir / f"{safe_model}_{num_runs}runs.json"

    if out_file.exists():
        msg = f"Output file already exists: {out_file}. Delete it or choose a different output directory."
        raise typer.BadParameter(msg, param_hint="out_dir")

    command.batch_evaluate_tickets(in_dir=in_dir, out_file=out_file, num_runs=num_runs, config=config)

@cli.command()
def login(ctx: typer.Context) -> None:
    """Authorise rcpond with ServiceNow via OAuth (browser-based flow).

    Opens a browser, completes the Authorization Code + PKCE flow, and caches
    the resulting tokens. Subsequent commands will use the cached token
    automatically without prompting again.
    """
    from rcpond.auth import get_bearer_token

    get_bearer_token(_config(ctx))
    print("[green]Login successful.[/green] Token cached.")

@cli.command()
def process_all(
    ctx: typer.Context,
    dry_run: bool = False,
    reply_mode: ReplyMode = ReplyMode.default,
    ## Single flag name (no "--flag/--no-flag" form) suppresses Typer's auto-generated
    ## negative, which would otherwise produce the unreadable `--no-yes-i-am-sure`.
    yes_i_am_sure: Annotated[
        bool, typer.Option("--yes-i-am-sure", help="Confirm processing all unassigned tickets.")
    ] = False,
):
    """Review all unassigned tickets via the LLM."""
    if yes_i_am_sure:
        command.batch_process_tickets(dry_run=dry_run, reply_mode=reply_mode, config=_config(ctx))
    else:
        msg = "The [bold cyan]--yes-i-am-sure[/bold cyan] option MUST be specified when using the [bold]process-all[/bold] subcommand."
        print(msg)
        typer.echo(ctx.get_help())

@cli.command()
def process_next(
    ctx: typer.Context,
    dry_run: bool = False,
    reply_mode: Annotated[ReplyMode, typer.Option(help=_REPLY_MODE_HELP)] = ReplyMode.default,
):
    """Review an arbitrarily selected unassigned ticket via the LLM."""
    command.process_next_ticket(dry_run=dry_run, reply_mode=reply_mode, config=_config(ctx))

@cli.command()
def process_ticket(
    ctx: typer.Context,
    ticket_number: str,
    dry_run: bool = False,
    reply_mode: Annotated[ReplyMode, typer.Option(help=_REPLY_MODE_HELP)] = ReplyMode.default,
):
    """Review a specific ticket (e.g. RES0001234) via the LLM."""
    command.process_specific_ticket(
        ticket_number=ticket_number, dry_run=dry_run, reply_mode=reply_mode, config=_config(ctx)
    )

@cli.command()
def whoami(ctx: typer.Context) -> None:
    """Show the identity of the currently authenticated OAuth user."""
    from rcpond.servicenow import ServiceNow

    sn = ServiceNow(_config(ctx))
    if not sn._is_oauth:
        print("[yellow]Static token authentication — user identity not available.[/yellow]")
        return
    claims = sn._fetch_current_user_claims()
    if not claims:
        print("[red]Unable to determine user identity.[/red]")
        raise typer.Exit(1)
    print(f"[bold]Name:[/bold]      {claims.get('name', '?')}")
    print(f"[bold]Username:[/bold]  {claims.get('user_name', '?')}")
    print(f"[bold]sys_id:[/bold]    {claims.get('sub', '?')}")

def batch_evaluate_tickets(in_dir: Path, out_file: Path, num_runs: int = 1, config: Config | None = None):
    """Process all Azure tickets in an offline HTML directory, across multiple runs.

    Used for evaluating the performance of the LLM in reviewing tickets. Results
    are written as ``dict[str, list[LLMResponse]]`` keyed by ticket number, so
    that each ticket's responses across all runs are grouped together. Non-Azure
    tickets are skipped.

    Parameters
    ----------
    in_dir : Path
        Directory containing pre-downloaded HTML ticket files.
    out_file : Path
        Path to write the JSON results. Must not already exist.
    num_runs : int
        Number of times to run the LLM over all tickets (for majority-vote analysis).
    config : Config | None
        Configuration to use. If None, Config() is constructed from the environment.
    """
    try:
        from rcpond.html_servicenow import HtmlServiceNow
    except ImportError as e:
        msg = "The 'html' optional dependencies are required for this command. Install them with: pip install rcpond[html]"
        raise ImportError(msg) from e

    config = config or Config()
    service_now: HtmlServiceNow = HtmlServiceNow(in_dir)
    llm: LLM = LLM(config)

    ## Pre-filter to Azure tickets only
    all_tickets = service_now.get_tickets(long_list=True)
    azure_tickets = []
    for ticket in all_tickets:
        # TODO: Temporary, an messy way to limit tickets to only those related to Azure
        # Find a better solution
        full_ticket = service_now.get_full_ticket(ticket)

        if full_ticket.which_service != "Azure":
            print(f"skipping non-Azure ticket: {ticket.number}")
        else:
            azure_tickets.append(ticket)

    ## Run the LLM num_runs times, accumulating responses per ticket
    results: dict[str, list[LLMResponse]] = {t.number: [] for t in azure_tickets}
    for run in range(num_runs):
        print(f"\n--- Run {run + 1}/{num_runs} ---")
        for ticket in azure_tickets:
            resp = _process_ticket(
                ticket=ticket,
                dry_run=True,
                config=config,
                service_now=service_now,
                llm=llm,
                reply_mode=ReplyMode.always,
            )
            results[ticket.number].append(resp)
            print()

    with open(out_file, "w") as f:
        json.dump({k: [vars(r) for r in v] for k, v in results.items()}, f, indent=2)

def batch_process_tickets(dry_run: bool, reply_mode: ReplyMode, config: Config | None = None):
    """Process all unassigned ServiceNow tickets via the LLM.

    Each ticket is reviewed individually. The LLM response and reasoning are
    displayed for each. If the LLM recommends actions and ``dry_run`` is False,
    the actions are performed.

    Parameters
    ----------
    dry_run : bool
        If True, planned tool calls are not executed.
    reply_mode : ReplyMode
        Controls when to skip a ticket based on prior RCPond activity.
    config : Config | None
        Configuration to use. If None, Config() is constructed from the environment.
    """
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    llm: LLM = LLM(config)
    all_tickets = service_now.get_tickets()

    if len(all_tickets) > 0:
        for num, ticket in enumerate(all_tickets, start=1):
            print(f"Processing ticket {num} of {len(all_tickets)}")
            resp: LLMResponse = _process_ticket(ticket, dry_run, config, service_now, llm, reply_mode)
            display_short_ticket(ticket)
            display_response(resp)
    else:
        print("No tickets that have not been previously processed by RCPond")

def display_all_tickets(long_list: bool, config: Config | None = None):
    """Display the list of unassigned tickets from ServiceNow to the user."""
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    display_multi_tickets(service_now.get_tickets(long_list=long_list))

def display_single_ticket(ticket_number: str, config: Config | None = None):
    """Display the details of a specific ticket."""
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    ticket = service_now.get_ticket(ticket_number)
    display_full_ticket(service_now.get_full_ticket(ticket))

def get_ticket_url(ticket_number: str, config: Config | None = None) -> str:
    """Return the ServiceNow Web UI URL for a specific ticket.

    Parameters
    ----------
    ticket_number : str
        The ticket number to look up (e.g. ``"RES0001234"``).
    config : Config | None
        Configuration to use. If None, Config() is constructed from the environment.
    """
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    ticket = service_now.get_ticket(ticket_number)
    return service_now.web_url(ticket)

def process_next_ticket(dry_run: bool, reply_mode: ReplyMode, config: Config | None = None):
    """Process an arbitrarily selected ServiceNow ticket via the LLM.

    The LLM response and reasoning are displayed to the user. If the LLM
    recommends an action and ``dry_run`` is False, the action is performed.

    Parameters
    ----------
    dry_run : bool
        If True, planned tool calls are not executed.
    reply_mode : ReplyMode
        Controls when to skip a ticket based on prior RCPond activity.
    config : Config | None
        Configuration to use. If None, Config() is constructed from the environment.
    """
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    llm: LLM = LLM(config)
    tickets: list[Ticket] = service_now.get_tickets()
    next_ticket = tickets.pop()
    resp: LLMResponse = _process_ticket(next_ticket, dry_run, config, service_now, llm, reply_mode)
    display_short_ticket(next_ticket)
    display_response(resp)

def process_specific_ticket(ticket_number: str, dry_run: bool, reply_mode: ReplyMode, config: Config | None = None):
    """Process the given ServiceNow ticket via the LLM.

    The LLM response and reasoning are displayed to the user. If the LLM
    recommends an action and ``dry_run`` is False, the action is performed.

    Parameters
    ----------
    ticket_number : str
        The ticket number (e.g. ``"RES0001234"``) to process.
    dry_run : bool
        If True, planned tool calls are not executed.
    reply_mode : ReplyMode
        Controls when to skip a ticket based on prior RCPond activity.
    config : Config | None
        Configuration to use. If None, Config() is constructed from the environment.
    """
    config = config or Config()
    service_now: ServiceNow = ServiceNow(config)
    llm: LLM = LLM(config)
    ticket = service_now.get_ticket(ticket_number)
    resp: LLMResponse = _process_ticket(ticket, dry_run, config, service_now, llm, reply_mode)
    display_short_ticket(ticket)
    display_response(resp)

def display_full_ticket(ticket: FullTicket, *, console: Console | None = None) -> None:
    """Display the full details of a ticket using Rich formatting.

    Parameters
    ----------
    ticket : FullTicket
        The ticket to display.
    console : Console | None
        Rich Console to write to. Defaults to the module-level console (stdout).
    """
    con = console or _console

    con.print(_header_panel(ticket))

    ## Request details
    con.print(
        _section(
            "Request details",
            [
                ("Service", ticket.which_service),
                ("Subscription type", ticket.subscription_type),
                ("New / existing", ticket.new_or_existing_allocation),
                ("Research area", ticket.research_area_programme or ticket.if_other_please_specify),
                ("Data sensitivity", ticket.data_sensitivity),
                ("Start date", ticket.start_date),
                ("End date", ticket.end_date),
            ],
        )
    )

    ## Project
    con.print(
        _section(
            "Project",
            [
                ("Project title", ticket.project_title),
                ("PI / Supervisor", ticket.pi_supervisor_name),
                ("PI email", ticket.pi_supervisor_email),
            ],
        )
    )

    ## Finance
    con.print(
        _section(
            "Finance",
            [
                ("Credits requested", ticket.credits_requested),
                ("Finance code", ticket.which_finance_code),
                ("PMU contact email", ticket.pmu_contact_email),
                ("Subscription / Azure ID", ticket.azure_subscription_id_or_hpc_group_project_id),
                ("Cost breakdown", ticket.cost_compute_time_breakdown),
            ],
        )
    )

    ## Technical requirements
    con.print(
        _section(
            "Technical requirements",
            [
                ("CPU hours", ticket.cpu_hours_required),
                ("GPU hours", ticket.gpu_hours_required),
                ("Facility", ticket.which_facility or ticket.if_other_please_specify_facility),
                ("Computational requirements", ticket.computational_requirements),
                ("Platform justification", ticket.platform_justification),
                ("Research justification", ticket.research_justification),
            ],
        )
    )

    ## Access
    if ticket.users_who_require_access_names_and_emails:
        con.print(
            _section(
                "Users requiring access",
                [
                    ("Users", ticket.users_who_require_access_names_and_emails),
                ],
            )
        )

    ## Work notes and comments, merged chronologically
    notes = ticket.get_combined_notes()
    if notes:
        con.print(_notes_panel(notes))

def display_multi_tickets(tickets: list[Ticket], *, console: Console | None = None) -> None:
    """Display a table of ticket summaries.

    Each section represents tickets with common values of "Category" and "Description". The Description field is used for the section title.

    Each row represents a single ticket. It should have the columns
    - Number
    - Opened date/time
    - Requested by
    - Status (new / assigned / on-hold / etc)
    """
    con = console or _console

    if not tickets:
        con.print("[dim]No tickets found.[/dim]")
        return

    ## Group tickets by (u_category, short_description)
    groups: dict[tuple[str, str], list[Ticket]] = {}
    for ticket in tickets:
        key = (ticket.u_category, ticket.short_description)
        groups.setdefault(key, []).append(ticket)

    for (category, description), group in groups.items():
        table = Table(show_header=True, header_style="bold cyan", box=None, padding=(0, 2))
        table.add_column("Number", no_wrap=True)
        table.add_column("Opened", no_wrap=False)
        table.add_column("Requested by")
        table.add_column("Status")
        table.add_column("Assigned To")

        for ticket in group:
            table.add_row(
                ticket.number,
                ticket.opened_at,
                ticket.requested_for,
                ticket.state,
                ticket.assigned_to if ticket.assigned_to else "[dim]UNASSIGNED[/dim]",
            )

        title = f"[bold]{description} / {category}[/bold]"
        con.print(Panel(table, title=title, title_align="left", border_style="bright_blue"))

def display_response(response: LLMResponse, *, console: Console | None = None) -> None:
    """Display an LLM response.

    Parameters
    ----------
    response : LLMResponse
        The response to display.
    console : Console | None
        Rich Console to write to. Defaults to the module-level console (stdout).
    """
    con = console or _console

    title = "[bold]LLM Response[/bold]"
    if response.ticket_number:
        title += f"  [dim]{response.ticket_number}[/dim]"
    if response.llm_model:
        title += f"  [dim italic]{response.llm_model}[/dim italic]"

    ## Main response text
    response_text = response.response_text if response.response_text else "NO RESPONSE RECEIVED!"
    con.print(
        Panel(
            Text(response_text, overflow="fold"),
            title=title,
            title_align="left",
            border_style="bright_blue",
        )
    )

    ## Reasoning (only present for chain-of-thought models)
    if response.reasoning:
        con.print(
            Panel(
                Text(response.reasoning, overflow="fold"),
                title="[bold]Reasoning[/bold]",
                title_align="left",
                border_style="dim",
            )
        )

    ## Planned tool call
    if response.planned_tool_call:
        tool_name = response.planned_tool_call.get("function", {}).get("name", "unknown")
        args = response.planned_tool_call.get("function", {}).get("arguments", {})
        rows = list(args.items()) if isinstance(args, dict) else [("arguments", str(args))]
        con.print(
            Panel(
                _kv_table(rows),
                title=f"[bold]Tool call:[/bold] [cyan]{tool_name}[/cyan]",
                title_align="left",
                border_style="yellow",
            )
        )

def display_short_ticket(ticket: Ticket, *, console: Console | None = None) -> None:
    """Display a high-level ticket summary.

    Parameters
    ----------
    ticket : Ticket
        The ticket to display.
    console : Console | None
        Rich Console to write to. Defaults to the module-level console (stdout).
    """
    con = console or _console
    con.print(_header_panel(ticket))

def extract_key_facts(filename: Path) -> dict:
    """Extract ticket fields from a ServiceNow HTML export file.

    Parameters
    ----------
    filename : Path
        Path to the HTML file.

    Returns
    -------
    dict[str, str | pd.DataFrame | None]
        Extracted fields. String values are ``None`` when the corresponding
        HTML element is absent. ``"activities"`` is always a ``pd.DataFrame``.
    """
    soup = BeautifulSoup(filename.read_text(), "html.parser")

    activities = _extract_activities_as_table(soup)

    comments = _extract_comments(activities)
    work_notes = "\n\n".join(comments["text"].dropna())

    def _q(question: str, starting_point: str | None = None) -> str:
        return _extract_question_answer_pair_table(soup, question, starting_point) or ""

    return {
        ## Internal keys (not FullTicket fields)
        "sys_id": _extract_sys_id(soup),
        "ticket_number": _extract_ticket_number(soup),
        "assigned_to": _extract_assigned_to(soup),
        "state": _extract_state(soup),
        "category": _extract_category(soup),
        "sub_category": _extract_sub_category(soup),
        "activity_count": str(len(activities)),
        "comment_count": str(_extract_comment_count(activities)),
        "activities": activities,
        ## FullTicket fields — keys match FullTicket field names exactly
        "work_notes": work_notes,
        "comments": "",
        "which_service": _extract_platform_choice(soup) or "",
        "requested_for": _q("requested for", "Variables"),
        "project_title": _q("project title"),
        "research_area_programme": _extract_research_area_or_programme(soup) or "",
        "pi_supervisor_name": _q("PI/Supervisor name"),
        "pi_supervisor_email": _q("PI/Supervisor email"),
        "subscription_type": _q("Subscription type"),
        "which_finance_code": _q("Which finance code"),
        "pmu_contact_email": _q("PMU Contact email"),
        "credits_requested": _q("Credits requested"),
        "which_facility": _q("Which facility"),
        "if_other_please_specify": "",  ## handled by fallback logic in _extract_research_area_or_programme
        "if_other_please_specify_facility": "",
        "cpu_hours_required": _q("CPU hours required?"),
        "gpu_hours_required": _q("GPU hours required?"),
        "new_or_existing_allocation": _q("New or existing allocation"),
        "azure_subscription_id_or_hpc_group_project_id": _q("Azure subscription ID or HPC Group/Project ID"),
        "start_date": _q("Start date"),
        "end_date": _q("End date"),
        "data_sensitivity": _q("Data sensitivity"),
        "platform_justification": _q("Platform justification"),
        "research_justification": _q("Research justification"),
        "computational_requirements": _q("Computational requirements"),
        "users_who_require_access_names_and_emails": _q("Users who require access (names and emails)"),
        "cost_compute_time_breakdown": _q("Cost/compute time breakdown"),
    }