kosmos
12ca06ba0d
sys.stdin.read(1) on the buffered TextIOWrapper pulls the whole escape sequence into its Python-level buffer, so the follow-up select() sees an empty fd and every arrow/Home/End press degrades to bare Esc; the unread '[B' tail then leaks into the next keypress. Read raw bytes via os.read() and select() on the file descriptor instead, keeping the same key tokens and classifications. Add pty-driven regression tests over a real cbreak terminal covering CSI sequences, single-byte keys, bare Esc, and tail-leak.
471 lines
17 KiB
Python
471 lines
17 KiB
Python
"""Interactive onboarding wizard behind `weekly-activity config`.
|
|
|
|
A small menu-driven editor for the account-list config: it shows the
|
|
current sources (secrets masked), then offers Add a new source /
|
|
Remove a source / Exit, supporting several accounts per provider.
|
|
|
|
Adding a non-anonymous Launchpad account runs launchpadlib's interactive
|
|
OAuth flow right here (the browser opens once and the token lands in the
|
|
system keyring) so `weekly-activity report` never has to authenticate;
|
|
supplying a credentials file skips that flow. Token values are never
|
|
echoed back.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import getpass
|
|
import os
|
|
import sys
|
|
from collections.abc import Callable, Sequence
|
|
from pathlib import Path
|
|
|
|
from weekly_activity.aggregate import source_label
|
|
from weekly_activity.config import (
|
|
ActivityConfig,
|
|
BtsSettings,
|
|
ConfigError,
|
|
GithubSettings,
|
|
GitlabSettings,
|
|
LaunchpadSettings,
|
|
load_config,
|
|
save_config,
|
|
)
|
|
from weekly_activity.sources.launchpad import LaunchpadSource
|
|
|
|
_APPLICATION_NAME = "weekly-activity"
|
|
|
|
# Stable processing order, shared by menus and summaries: (config attr, menu text).
|
|
_KINDS: tuple[tuple[str, str], ...] = (
|
|
("launchpad", "Launchpad account"),
|
|
("github", "GitHub account"),
|
|
("gitlab", "GitLab instance/account"),
|
|
("bts", "Debian BTS email"),
|
|
)
|
|
|
|
|
|
def run_wizard(config_path: Path) -> None:
|
|
"""Edit the source configuration interactively; write when changed."""
|
|
try:
|
|
config = _load_existing(config_path)
|
|
original = _snapshot(config)
|
|
print()
|
|
print("Welcome! This wizard manages the sources aggregated by")
|
|
print("`weekly-activity report`. Enter a menu number, accept [defaults]")
|
|
print("with Enter, and pick Exit to save; Ctrl-C aborts without writing.")
|
|
try:
|
|
_menu_loop(config)
|
|
except EOFError:
|
|
print()
|
|
print("End of input; applying current selections.")
|
|
except KeyboardInterrupt:
|
|
raise SystemExit(130) from None # parity with secret-prompt interruption
|
|
_finish(config, original, config_path)
|
|
|
|
|
|
def _snapshot(config: ActivityConfig) -> ActivityConfig:
|
|
"""Copy the lists so edits can be compared against the starting point."""
|
|
return ActivityConfig(
|
|
launchpad=list(config.launchpad),
|
|
github=list(config.github),
|
|
gitlab=list(config.gitlab),
|
|
bts=list(config.bts),
|
|
)
|
|
|
|
|
|
def _menu_loop(config: ActivityConfig) -> None:
|
|
while True:
|
|
print()
|
|
print("Current sources:")
|
|
print(_describe(config))
|
|
action = _menu("What do you want to do?", ["Add a new source", "Remove a source", "Exit"])
|
|
if action == 0:
|
|
_add_source_flow(config)
|
|
elif action == 1:
|
|
_remove_source_flow(config)
|
|
else:
|
|
if not config.enabled_sources() and not _confirm(
|
|
"No sources are configured; write an empty configuration anyway?", default=False
|
|
):
|
|
continue
|
|
return
|
|
|
|
|
|
def _finish(config: ActivityConfig, original: ActivityConfig, config_path: Path) -> None:
|
|
if config == original:
|
|
print()
|
|
print("No changes made; configuration left untouched.")
|
|
return
|
|
save_config(config, config_path)
|
|
enabled = ", ".join(config.enabled_sources())
|
|
print()
|
|
print(f"Configuration written to {config_path}.")
|
|
print(f"Enabled sources: {enabled}.")
|
|
print(_describe(config, indent=""))
|
|
print("Tokens are never printed back. Run `weekly-activity report` to")
|
|
print("aggregate activity across all of them in one output.")
|
|
|
|
|
|
def _load_existing(path: Path) -> ActivityConfig:
|
|
try:
|
|
return load_config(path)
|
|
except FileNotFoundError:
|
|
return ActivityConfig()
|
|
except ConfigError as exc:
|
|
print(f"warning: ignoring unusable existing config: {exc}", file=sys.stderr)
|
|
print(f"warning: {path} will be replaced by a fresh configuration.", file=sys.stderr)
|
|
return ActivityConfig()
|
|
|
|
|
|
# --- Add / remove flows ---
|
|
|
|
|
|
def _add_source_flow(config: ActivityConfig) -> None:
|
|
chosen = _menu(
|
|
"Which source do you want to add?", [text for _, text in _KINDS], cancellable=True
|
|
)
|
|
if chosen is None:
|
|
return
|
|
kind = _KINDS[chosen][0]
|
|
account = _ADDERS[kind]()
|
|
if account is None:
|
|
print("Cancelled; nothing was added.")
|
|
return
|
|
accounts = getattr(config, kind)
|
|
accounts.append(account)
|
|
label = source_label(kind, _display_name(account), _identity(account), len(accounts) > 1)
|
|
print(f"Added: {label} ({_identity(account)}).")
|
|
|
|
|
|
def _remove_source_flow(config: ActivityConfig) -> None:
|
|
entries: list[tuple[str, int]] = []
|
|
options: list[str] = []
|
|
for kind, _ in _KINDS:
|
|
accounts = getattr(config, kind)
|
|
for index, account in enumerate(accounts):
|
|
entries.append((kind, index))
|
|
options.append(_account_line(kind, account, len(accounts) > 1))
|
|
if not entries:
|
|
print("There are no sources to remove yet.")
|
|
return
|
|
picked = _menu("Which account do you want to remove?", options, cancellable=True)
|
|
if picked is None or not _confirm(f"Remove {options[picked]}?", default=False):
|
|
return
|
|
kind, index = entries[picked]
|
|
del getattr(config, kind)[index]
|
|
print(f"Removed: {options[picked]}.")
|
|
|
|
|
|
def _add_github() -> GithubSettings | None:
|
|
print("GitHub identity is the profile @handle.")
|
|
username = _ask_required("GitHub username")
|
|
name = _ask("Display name (optional label)")
|
|
print("Leave both token fields empty to fall back to $GITHUB_TOKEN or `gh auth token`.")
|
|
token_env = _ask("Env var holding the token (optional, e.g. GITHUB_TOKEN)")
|
|
token = _ask_secret("GitHub token (input hidden)")
|
|
return GithubSettings(username=username, token=token, token_env=token_env, name=name)
|
|
|
|
|
|
def _add_gitlab() -> GitlabSettings | None:
|
|
print("GitLab needs the instance URL plus your username there.")
|
|
username = _ask_required("GitLab username")
|
|
url = _ask("Instance URL", "https://gitlab.com")
|
|
name = _ask("Display name (optional label)")
|
|
print("Leave both token fields empty to discover the token from the environment.")
|
|
token = _ask_secret("API token (input hidden)")
|
|
token_env = _ask("Env var holding the token (optional, e.g. SALSA_TOKEN)")
|
|
return GitlabSettings(username=username, url=url, token=token, token_env=token_env, name=name)
|
|
|
|
|
|
def _add_bts() -> BtsSettings | None:
|
|
print("The BTS identifies people by the email used on bugs (submitter/owner).")
|
|
email = _ask_required("Email address on Debian bugs")
|
|
name = _ask("Display name (optional label)")
|
|
return BtsSettings(email=email, name=name)
|
|
|
|
|
|
def _add_launchpad() -> LaunchpadSettings | None:
|
|
print("Launchpad identity is the ~name shown on your profile page.")
|
|
username = _ask_required("Launchpad username")
|
|
name = _ask("Display name (optional label)")
|
|
anonymous = _confirm("Anonymous mode (public data only, no OAuth)?", default=True)
|
|
if anonymous:
|
|
return LaunchpadSettings(username=username, mode="anonymous", name=name)
|
|
|
|
credentials_file = _ask("Credentials file path (empty to authenticate right now)")
|
|
if credentials_file:
|
|
expanded = os.path.expanduser(credentials_file)
|
|
if not os.path.isfile(expanded):
|
|
print(f"note: {expanded} does not exist yet; it will be read at report time.")
|
|
print(f"OAuth credentials will be read from {credentials_file} at report time.")
|
|
return LaunchpadSettings(
|
|
username=username, mode="credentials", credentials_file=credentials_file, name=name
|
|
)
|
|
|
|
# Real config-time OAuth: construct LaunchpadSource (this runs
|
|
# launchpadlib's browser flow) and keep only its side effect — the token
|
|
# in the system keyring keyed to the application name.
|
|
while True:
|
|
print("A browser window will open to authorize weekly-activity with Launchpad.")
|
|
try:
|
|
LaunchpadSource(
|
|
service="production",
|
|
anonymous=False,
|
|
application_name=_APPLICATION_NAME,
|
|
credentials_file=None,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - network/user errors here are recoverable
|
|
print(f"warning: Launchpad authentication failed: {exc}")
|
|
if not _confirm("Retry authentication?", default=False):
|
|
return None
|
|
else:
|
|
break
|
|
print("Launchpad authorization complete: the token is stored in your system keyring;")
|
|
print("`weekly-activity report` will reuse it automatically.")
|
|
return LaunchpadSettings(username=username, mode="credentials", name=name)
|
|
|
|
|
|
_ADDERS: dict[str, Callable[[], object | None]] = {
|
|
"launchpad": _add_launchpad,
|
|
"github": _add_github,
|
|
"gitlab": _add_gitlab,
|
|
"bts": _add_bts,
|
|
}
|
|
|
|
|
|
# --- Display helpers (secrets masked) ---
|
|
|
|
|
|
def _describe(config: ActivityConfig, indent: str = " ") -> str:
|
|
"""Human-readable summary masking every token value."""
|
|
lines: list[str] = []
|
|
for kind, _ in _KINDS:
|
|
accounts = getattr(config, kind)
|
|
for account in accounts:
|
|
lines.append(indent + _account_line(kind, account, len(accounts) > 1))
|
|
return "\n".join(lines) if lines else f"{indent}(no sources configured)"
|
|
|
|
|
|
def _account_line(kind: str, account: object, ambiguous: bool) -> str:
|
|
identity = _identity(account)
|
|
prefix = f"{source_label(kind, _display_name(account), identity, ambiguous)}: "
|
|
if kind == "launchpad" and isinstance(account, LaunchpadSettings):
|
|
auth = (
|
|
"anonymous (public data only)"
|
|
if account.mode == "anonymous"
|
|
else f"OAuth via {account.credentials_file or 'system keyring'}"
|
|
)
|
|
return f"{prefix}{identity} — {auth}"
|
|
if kind == "github" and isinstance(account, GithubSettings):
|
|
fallback = account.token_env or "GITHUB_TOKEN / gh"
|
|
token_desc = "stored" if account.token else f"not stored (uses {fallback})"
|
|
return f"{prefix}{identity} — token {token_desc}"
|
|
if kind == "gitlab" and isinstance(account, GitlabSettings):
|
|
fallback = account.token_env or "<HOST>_TOKEN / GITLAB_TOKEN env"
|
|
token_desc = "stored" if account.token else f"not stored ({fallback})"
|
|
return f"{prefix}{identity} @ {account.url} — token {token_desc}"
|
|
if kind == "bts" and isinstance(account, BtsSettings):
|
|
return f"{prefix}{identity}"
|
|
return f"{prefix}{identity}" # pragma: no cover - exhaustive over known kinds
|
|
|
|
|
|
def _identity(account: object) -> str:
|
|
if isinstance(account, BtsSettings):
|
|
return account.email
|
|
if isinstance(account, (GithubSettings, GitlabSettings, LaunchpadSettings)):
|
|
return account.username
|
|
return "" # pragma: no cover - exhaustive over known kinds
|
|
|
|
|
|
def _display_name(account: object) -> str:
|
|
if isinstance(account, (GithubSettings, GitlabSettings, LaunchpadSettings, BtsSettings)):
|
|
return account.name
|
|
return "" # pragma: no cover - exhaustive over known kinds
|
|
|
|
|
|
# --- Prompt helpers ---
|
|
|
|
|
|
def _menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None:
|
|
"""Pick an option: interactive arrows on a TTY, numbered prompt otherwise.
|
|
|
|
Returns the selected index, or None when backed out of a cancellable menu.
|
|
"""
|
|
if sys.stdin.isatty() and sys.stdout.isatty():
|
|
return _arrow_menu(title, options, cancellable=cancellable)
|
|
return _numbered_menu(title, options, cancellable=cancellable)
|
|
|
|
|
|
def _numbered_menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None:
|
|
"""Numbered-choice fallback for non-TTY stdin (piped input, tests, CI)."""
|
|
print(title)
|
|
for number, option in enumerate(options, 1):
|
|
print(f" {number}) {option}")
|
|
back_number = len(options) + 1 if cancellable else 0
|
|
if cancellable:
|
|
print(f" {back_number}) Back")
|
|
while True:
|
|
answer = input("Enter choice: ").strip()
|
|
if not answer and cancellable:
|
|
return None
|
|
if answer.isdigit():
|
|
number = int(answer)
|
|
if 1 <= number <= len(options):
|
|
return number - 1
|
|
if cancellable and number == back_number:
|
|
return None
|
|
valid = back_number or len(options)
|
|
print(f"Please enter a number between 1 and {valid}.")
|
|
|
|
|
|
# --- Arrow-key menu (interactive TTYs only) ---
|
|
|
|
|
|
_BOLD = "\x1b[1m"
|
|
_RESET = "\x1b[0m"
|
|
|
|
_ERASE_LINE = "\x1b[2K"
|
|
_CURSOR_DOWN = "\x1b[1B"
|
|
_CURSOR_UP_TEMPLATE = "\x1b[{}A"
|
|
|
|
# Key classifications shared by the reader and the menu loop.
|
|
_KEY_UP = "up"
|
|
_KEY_DOWN = "down"
|
|
_KEY_HOME = "home"
|
|
_KEY_END = "end"
|
|
_KEY_ENTER = "enter"
|
|
_KEY_ESCAPE = "escape"
|
|
_KEY_OTHER = "other"
|
|
|
|
_CSI_KEYS = {"[A": _KEY_UP, "[B": _KEY_DOWN, "[H": _KEY_HOME, "[F": _KEY_END}
|
|
|
|
|
|
def _read_keypress() -> str:
|
|
"""Read one keystroke from stdin (cbreak mode); classify it."""
|
|
import select
|
|
|
|
fd = sys.stdin.fileno()
|
|
# Read raw bytes from the fd: a buffered text read would pull the whole
|
|
# escape sequence into its own buffer, leaving nothing for select to see.
|
|
first = os.read(fd, 1)
|
|
if first == b"\x1b":
|
|
# A bare Esc is not followed by more bytes; CSI sequences are. Peek briefly.
|
|
readable, _, _ = select.select([fd], [], [], 0.05)
|
|
if not readable:
|
|
return _KEY_ESCAPE
|
|
tail = os.read(fd, 2)
|
|
return _CSI_KEYS.get(tail.decode("ascii", "replace"), _KEY_OTHER)
|
|
if first in (b"\r", b"\n"):
|
|
return _KEY_ENTER
|
|
if first == b"\x01": # Ctrl-A
|
|
return _KEY_HOME
|
|
if first == b"\x05": # Ctrl-E
|
|
return _KEY_END
|
|
if first == b"k":
|
|
return _KEY_UP
|
|
if first == b"j":
|
|
return _KEY_DOWN
|
|
return _KEY_OTHER
|
|
|
|
|
|
def _option_line(option: str, selected: bool) -> str:
|
|
"""One rendered row: bold with a '>' marker on the selected line."""
|
|
prefix = "> " if selected else " "
|
|
if selected:
|
|
return f"{_BOLD}{prefix}{option}{_RESET}"
|
|
return f"{prefix}{option}"
|
|
|
|
|
|
def _draw_menu_rows(options: Sequence[str], selected: int | None) -> None:
|
|
"""Redraw or clear the option rows.
|
|
|
|
On entry the cursor sits at column 0 of the blank row just below the
|
|
block; writes move up and repaint each row so the title above stays
|
|
untouched. With ``selected=None`` every row is erased instead,
|
|
clearing the menu before trailing prompts run.
|
|
"""
|
|
out = sys.stdout
|
|
out.write(_CURSOR_UP_TEMPLATE.format(len(options)))
|
|
for index, option in enumerate(options):
|
|
out.write(_ERASE_LINE + "\r")
|
|
if selected is not None:
|
|
out.write(_option_line(option, index == selected))
|
|
out.write(_CURSOR_DOWN)
|
|
out.flush()
|
|
|
|
|
|
def _arrow_menu(title: str, options: Sequence[str], *, cancellable: bool = False) -> int | None:
|
|
"""Inline re-rendered menu driven by raw-mode keypresses."""
|
|
import termios
|
|
import tty
|
|
|
|
print(title)
|
|
selected = 0
|
|
for index, option in enumerate(options): # fresh draw lands one row per option
|
|
print(_option_line(option, index == selected))
|
|
fd = sys.stdin.fileno()
|
|
saved_state = termios.tcgetattr(fd)
|
|
choice: int | None = None
|
|
try:
|
|
tty.setcbreak(fd)
|
|
while True:
|
|
key = _read_keypress()
|
|
if key == _KEY_ENTER:
|
|
choice = selected
|
|
break
|
|
if key == _KEY_UP:
|
|
selected = (selected - 1) % len(options)
|
|
elif key == _KEY_DOWN:
|
|
selected = (selected + 1) % len(options)
|
|
elif key == _KEY_HOME:
|
|
selected = 0
|
|
elif key == _KEY_END:
|
|
selected = len(options) - 1
|
|
elif cancellable and key in (_KEY_ESCAPE, "q"):
|
|
break
|
|
else: # Esc/q on fixed menus, or any unrecognised key: no-op.
|
|
continue
|
|
_draw_menu_rows(options, selected)
|
|
except KeyboardInterrupt:
|
|
# ISIG stays enabled under cbreak, so Ctrl-C raises SIGINT here.
|
|
if not cancellable:
|
|
raise
|
|
finally:
|
|
termios.tcsetattr(fd, termios.TCSADRAIN, saved_state)
|
|
_draw_menu_rows(options, None)
|
|
return choice
|
|
|
|
|
|
def _ask(prompt: str, default: str = "") -> str:
|
|
suffix = f" [{default}]" if default else ""
|
|
answer = input(f"{prompt}{suffix}: ").strip()
|
|
return answer or default
|
|
|
|
|
|
def _ask_required(prompt: str, default: str = "") -> str:
|
|
while True:
|
|
answer = _ask(prompt, default)
|
|
if answer:
|
|
return answer
|
|
print("A value is required here.")
|
|
|
|
|
|
def _ask_secret(prompt: str, default: str = "") -> str:
|
|
"""Read a secret without echo; keep the stored one when input is empty."""
|
|
try:
|
|
entered = getpass.getpass(f"{prompt}: ")
|
|
except (EOFError, KeyboardInterrupt):
|
|
raise SystemExit(130) from None
|
|
return entered.strip() or default
|
|
|
|
|
|
def _confirm(prompt: str, *, default: bool = False) -> bool:
|
|
hint = "[Y/n]" if default else "[y/N]"
|
|
while True:
|
|
answer = input(f"{prompt} {hint}: ").strip().lower()
|
|
if not answer:
|
|
return default
|
|
if answer in {"y", "yes"}:
|
|
return True
|
|
if answer in {"n", "no"}:
|
|
return False
|