Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 126 additions & 63 deletions phabfive/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,17 +6,43 @@
import os
import re
import sys
import webbrowser

from phabricator import Phabricator, APIError
from InquirerPy import inquirer as inq
from rich.console import Console
from rich.prompt import Prompt, Confirm

from phabfive.constants import VALIDATORS


log = logging.getLogger(__name__)


def _has_display() -> bool:
"""Check if we can reasonably auto-open a browser.

Returns False in environments where webbrowser.open() would either fail
or open a text-mode browser in the terminal:

- WSL2: Python's webbrowser module finds text browsers (lynx, w3m) instead
of the Windows browser. Detected via /proc/sys/kernel/osrelease containing
"microsoft".
- Docker containers: No display server available. Detected via the
/.dockerenv sentinel file.
- Podman / systemd-nspawn containers: No display server available. Detected
via the "container" environment variable set by these runtimes.
"""
try:
with open("/proc/sys/kernel/osrelease") as f:
if "microsoft" in f.read().lower():
return False
except OSError:
pass
if os.path.exists("/.dockerenv") or os.environ.get("container"):
return False
return True


class SetupWizard:
"""Interactive setup wizard for phabfive configuration."""

Expand All @@ -30,35 +56,58 @@ def __init__(self):
def _check_existing_config(self) -> bool:
"""Check if there's an existing working configuration.

If a working config exists, warn the user and ask for confirmation.
Reads ~/.arcrc directly to avoid triggering the interactive server
selector in Phabfive(). Tests each host silently and warns if any
connection works.

Returns:
bool: True to proceed with setup, False to abort
"""
try:
from phabfive.core import Phabfive

# Try to create a Phabfive instance - this validates config and connection
phabfive = Phabfive()
whoami = phabfive.phab.user.whoami()
username = whoami.get("userName", "unknown")

self.console.print(
"\n[yellow]Warning:[/yellow] A working configuration already exists."
)
self.console.print(
f"Currently connected to [bold]{phabfive.conf['PHAB_URL']}[/bold] "
f"as [bold]{username}[/bold].\n"
)

if not Confirm.ask("Do you want to reconfigure phabfive?", default=False):
self.console.print("Setup cancelled.\n")
return False

if not os.path.exists(self.CONFIG_PATH):
return True

with open(self.CONFIG_PATH, "r") as f:
arcrc = json.load(f)

hosts = arcrc.get("hosts", {})
if not hosts:
return True

# Try each host silently to find a working connection
for host_url, host_data in hosts.items():
token = host_data.get("token")
if not token:
continue
try:
phab = Phabricator(host=host_url, token=token)
phab.update_interfaces()
whoami = phab.user.whoami()
username = whoami.get("userName", "unknown")

self.console.print(
"\n[yellow]Warning:[/yellow] A working configuration already exists."
)
self.console.print(
f"Currently connected to [bold]{host_url}[/bold] "
f"as [bold]{username}[/bold].\n"
)

if not inq.confirm(
message="Do you want to reconfigure phabfive?", default=False
).execute():
self.console.print("Setup cancelled.\n")
return False

return True
except Exception:
continue

# No working host found - proceed with setup
return True

except Exception:
# No valid config or connection failed - proceed with setup
# Cannot read .arcrc or other error - proceed with setup
return True

def run(self) -> bool:
Expand Down Expand Up @@ -94,21 +143,21 @@ def run(self) -> bool:

def _print_header(self):
"""Print the setup wizard header."""
self.console.print("\n[bold]Phabfive Setup[/bold]")
self.console.print("=" * 40)
self.console.print()
self.console.rule("[bold]Phabfive Setup[/bold]")
self.console.print(
"\nThis wizard will configure phabfive to connect to your "
"Phabricator/Phorge instance.\n"
)

def _prompt_url(self) -> bool:
"""Prompt for and validate the Phabricator URL."""
self.console.print("[bold][1/3] Phabricator URL[/bold]")
self.console.rule("[bold][1/3] Phabricator URL[/bold]")

while True:
url = Prompt.ask(
"Enter your Phabricator URL (e.g., https://phorge.example.com)"
)
url = inq.text(
message="Enter your Phabricator URL (e.g., phorge.example.com):"
).execute()

if not url:
self.console.print("[red]URL cannot be empty[/red]")
Expand All @@ -126,30 +175,34 @@ def _prompt_url(self) -> bool:
continue

self.phab_url = normalized
self.console.print(f"[green]> Using API endpoint: {normalized}[/green]\n")
self.console.print(f" [green]✓[/green] API endpoint: {normalized}\n")
return True

def _prompt_token(self) -> bool:
"""Prompt for and validate the API token."""
self.console.print("[bold][2/3] API Token[/bold]")
self.console.rule("[bold][2/3] API Token[/bold]")

# Extract base URL for settings link
from urllib.parse import urlparse

parsed = urlparse(self.phab_url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
token_url = f"{base_url}/settings/panel/apitokens/"

self.console.print("To create an API token:")
self.console.print(
f" 1. Go to {base_url}/settings/user/YOUR_USERNAME/page/apitokens/"
)
self.console.print(' 2. Click "Generate API Token"')
self.console.print(" 3. Copy the token (starts with 'cli-')\n")

from InquirerPy import inquirer as inq
if _has_display():
self.console.print(
f"Opening [bold]{token_url}[/bold] in your browser to generate a token.\n"
)
webbrowser.open(token_url)
self.console.print(
"[dim]If the browser didn't open, visit the URL above manually.[/dim]\n"
)
else:
self.console.print(
f"Open this URL in your browser to generate a token:\n\n [bold]{token_url}[/bold]\n"
)

while True:
token = inq.secret(message="Enter your API token:").execute()
token = inq.secret(message="Paste your API token here:").execute()

if not token:
self.console.print("[red]Token cannot be empty[/red]")
Expand All @@ -169,33 +222,33 @@ def _prompt_token(self) -> bool:

def _verify_connection(self) -> bool:
"""Verify the connection to Phabricator."""
self.console.print("[bold][3/3] Verifying connection...[/bold]")
self.console.rule("[bold][3/3] Verify Connection[/bold]")

try:
phab = Phabricator(host=self.phab_url, token=self.phab_token)
phab.update_interfaces()
whoami = phab.user.whoami()
with self.console.status("Verifying connection..."):
phab = Phabricator(host=self.phab_url, token=self.phab_token)
phab.update_interfaces()
whoami = phab.user.whoami()

username = whoami.get("userName", "unknown")
realname = whoami.get("realName", "")

if realname:
self.console.print(
f"[green]> Connected successfully as: "
f"{username} ({realname})[/green]\n"
f" [green]✓[/green] Connected as: {username} ({realname})\n"
)
else:
self.console.print(
f"[green]> Connected successfully as: {username}[/green]\n"
)
self.console.print(f" [green]✓[/green] Connected as: {username}\n")

return True

except APIError as e:
self.console.print(f"[red]> Connection failed: {e}[/red]")
self.console.print("[red]Please check your URL and token.[/red]\n")

if Confirm.ask("Would you like to try again?", default=True):
if inq.confirm(
message="Would you like to try again?", default=True
).execute():
return (
self._prompt_url()
and self._prompt_token()
Expand Down Expand Up @@ -241,17 +294,23 @@ def _save_config(self) -> bool:

def _print_success(self):
"""Print success message."""
self.console.print(f"[green]Configuration saved to {self.CONFIG_PATH}[/green]")
self.console.print()
self.console.rule("[bold green]Setup Complete[/bold green]")
self.console.print(
f"\n [green]✓[/green] Configuration saved to {self.CONFIG_PATH}"
)
if os.name != "nt":
self.console.print(
"[green]File permissions set to 0600 (owner read/write only)[/green]"
" [green]✓[/green] File permissions set to 0600 (owner read/write only)"
)
self.console.print("\n[bold]Example commands:[/bold]")
self.console.print(" phabfive user whoami")
self.console.print(" phabfive maniphest search --assigned @me\n")

def _normalize_url(self, url: str) -> str:
"""Normalize URL to end with /api/."""
if not url.startswith(("http://", "https://")):
url = "https://" + url
url = url.rstrip("/")
if not url.endswith("/api"):
url += "/api"
Expand Down Expand Up @@ -312,16 +371,18 @@ def _setup_arcconfig(console) -> bool:
except (json.JSONDecodeError, IOError):
pass

if not Confirm.ask("Do you want to overwrite it?", default=False):
if not inq.confirm(
message="Do you want to overwrite it?", default=False
).execute():
return False

console.print("[bold]Create .arcconfig[/bold]")
console.rule("[bold]Create .arcconfig[/bold]")
console.print(f"This will create .arcconfig in: {git_root}\n")

while True:
url = Prompt.ask(
"Enter your Phabricator URL (e.g., https://phorge.example.com)"
)
url = inq.text(
message="Enter your Phabricator URL (e.g., https://phorge.example.com):"
).execute()

if not url:
console.print("[red]URL cannot be empty[/red]")
Expand All @@ -333,7 +394,7 @@ def _setup_arcconfig(console) -> bool:
if url.endswith("/api"):
url = url[:-4].rstrip("/")

console.print(f"[green]> Using URL: {url}[/green]\n")
console.print(f" [green]✓[/green] Using URL: {url}\n")
break

try:
Expand All @@ -342,7 +403,7 @@ def _setup_arcconfig(console) -> bool:
json.dump(arcconfig_data, f, indent=2)
f.write("\n")

console.print(f"[green].arcconfig created at {arcconfig_path}[/green]")
console.print(f" [green]✓[/green] .arcconfig created at {arcconfig_path}")
console.print("[dim]Remember to commit .arcconfig to your repository.[/dim]\n")
return True

Expand Down Expand Up @@ -376,13 +437,15 @@ def offer_setup_on_error(error_message: str) -> bool:
is_url_error = "PHAB_URL" in error_message and "PHAB_TOKEN" not in error_message

if is_url_error:
if Confirm.ask(
"Would you like to create .arcconfig for this repository?",
if inq.confirm(
message="Would you like to create .arcconfig for this repository?",
default=True,
):
).execute():
return _setup_arcconfig(console)
else:
if Confirm.ask("Would you like to run interactive setup now?", default=True):
if inq.confirm(
message="Would you like to run interactive setup now?", default=True
).execute():
wizard = SetupWizard()
return wizard.run()

Expand Down
Loading