Replace click.echo with rich print
MODIFY
plextraktsync/commands/cache.py
plextraktsync/commands/cache.py
from functools import partial
from typing import TYPE_CHECKING
-import click
-
from plextraktsync.factory import factory
if TYPE_CHECKING:
@@ -91,6 +89,7 @@ def cache_status(cache):
def cache(sort: str, limit: int, reverse: bool, expire: bool, url: str):
session = factory.session
+ print = factory.print
if expire:
expire_url(session, url)
@@ -100,9 +99,9 @@ def cache(sort: str, limit: int, reverse: bool, expire: bool, url: str):
inspect_url(session, url)
return
- click.echo(f"Cache status:\n{cache_status(session.cache)}\n")
+ print(f"Cache status:\n{cache_status(session.cache)}\n")
- click.echo("URLs:")
+ print("URLs:")
sorted = get_sorted_cache(session, sort, reverse)
for i, r in limit_iterator(sorted, limit):
- click.echo(f"- {i + 1:3d}. {r}")
+ print(f"- {i + 1:3d}. {r}")

MODIFY
plextraktsync/commands/clear_collections.py
plextraktsync/commands/clear_collections.py
-import click
+
from plextraktsync.factory import factory, logger
def clear_collections(confirm, dry_run):
+ print = factory.print
+
if not confirm and not dry_run:
- click.echo("You need to pass --confirm or --dry-run option to proceed")
+ print("You need to pass --confirm or --dry-run option to proceed")
return
trakt = factory.trakt_api

MODIFY
plextraktsync/commands/login.py
plextraktsync/commands/login.py
-import click
+
from plextraktsync.commands.plex_login import (has_plex_token,
plex_login_autoconfig)
from plextraktsync.commands.trakt_login import (has_trakt_token,
trakt_login_autoconfig)
+from plextraktsync.factory import factory
from plextraktsync.style import highlight, success
@@ -16,10 +17,12 @@ def ensure_login():
def login():
- click.echo(highlight("Checking Plex and Trakt login credentials existence"))
- click.echo("")
- click.echo("It will not test if the credentials are valid, only that they are present.")
- click.echo('If you need to re-login use "plex-login" or "trakt-login" commands respectively.')
- click.echo("")
+ print = factory.print
+
+ print(highlight("Checking Plex and Trakt login credentials existence"))
+ print("")
+ print("It will not test if the credentials are valid, only that they are present.")
+ print('If you need to re-login use "plex-login" or "trakt-login" commands respectively.')
+ print("")
ensure_login()
- click.echo(success("Done!"))
+ print(success("Done!"))

MODIFY
plextraktsync/commands/plex_login.py
plextraktsync/commands/plex_login.py
}
)
+print = factory.print
+
@flatten_list
def server_urls(server: MyPlexResource):
@@ -65,7 +67,7 @@ def server_urls(server: MyPlexResource):
def myplex_login(username, password):
while True:
username = click.prompt(PROMPT_PLEX_USERNAME, type=str, default=username)
- click.echo(NOTICE_2FA_PASSWORD)
+ print(NOTICE_2FA_PASSWORD)
password = click.prompt(
PROMPT_PLEX_PASSWORD,
type=str,
@@ -76,9 +78,9 @@ def myplex_login(username, password):
try:
return MyPlexAccount(username, password)
except Unauthorized as e:
- click.echo(error(f"Log in to Plex failed: {e}, Try again."))
+ print(error(f"Log in to Plex failed: '{e}', Try again."))
except BadRequest as e:
- click.echo(error(f"Log in to Plex failed: {e}"))
+ print(error(f"Log in to Plex failed: '{e}'"))
exit(1)
@@ -87,7 +89,7 @@ def choose_managed_user(account: MyPlexAccount):
if not users:
return None
- click.echo(success("Managed user(s) found:"))
+ print(success("Managed user(s) found:"))
users = sorted(users)
users.insert(0, account.username)
user = inquirer.select(
@@ -121,12 +123,12 @@ def fmt_server(s):
product = decorator(f"{s.product}/{s.productVersion}")
platform = decorator(f"{s.device}: {s.platform}/{s.platformVersion}")
- click.echo(
+ print(
f"- {highlight(s.name)}: [Last seen: {decorator(str(s.lastSeenAt))}, Server: {product} on {platform}]"
)
c: ResourceConnection
for c in s.connections:
- click.echo(f" {c.uri}")
+ print(f" {c.uri}")
owned_servers = [s for s in servers if s.owned]
unowned_servers = [s for s in servers if not s.owned]
@@ -134,12 +136,12 @@ def fmt_server(s):
server_names = []
if owned_servers:
- click.echo(success(f"{len(owned_servers)} owned servers found:"))
+ print(success(f"{len(owned_servers)} owned servers found:"))
for s in sorter(owned_servers):
fmt_server(s)
server_names.append(s.name)
if unowned_servers:
- click.echo(success(f"{len(unowned_servers)} unowned servers found:"))
+ print(success(f"{len(unowned_servers)} unowned servers found:"))
for s in sorter(unowned_servers):
fmt_server(s)
server_names.append(s.name)
@@ -180,14 +182,14 @@ def choose_server(account: MyPlexAccount):
raise ClickException("Unable to find server from Plex account")
# Connect to obtain baseUrl
- click.echo(
+ print(
title(
f"Attempting to connect to {server.name}. This may take time and print some errors."
)
)
- click.echo(title("Server connections:"))
+ print(title("Server connections:"))
for c in server.connections:
- click.echo(f" {c.uri}")
+ print(f" {c.uri}")
plex = server.connect()
return [server, plex]
except NotFound as e:
@@ -214,10 +216,10 @@ def login(username: str, password: str):
return
account = myplex_login(username, password)
- click.echo(success("Login to MyPlex was successful!"))
+ print(success("Login to MyPlex was successful!"))
[server, plex] = choose_server(account)
- click.echo(success(f"Connection to {plex.friendlyName} established successfully!"))
+ print(success(f"Connection to {plex.friendlyName} established successfully!"))
token = server.accessToken
user = account.username
@@ -245,4 +247,4 @@ def login(username: str, password: str):
CONFIG["PLEX_SERVER"] = server.name
CONFIG.save()
- click.echo(SUCCESS_MESSAGE)
+ print(SUCCESS_MESSAGE)

MODIFY
plextraktsync/commands/self_update.py
plextraktsync/commands/self_update.py
from typing import TYPE_CHECKING
-import click
-
+from plextraktsync.factory import factory
from plextraktsync.util.execp import execp
if TYPE_CHECKING:
@@ -35,21 +34,23 @@ def pr_number() -> Optional[int]:
def self_update(pr: int):
+ print = factory.print
+
if not pr:
pr = pr_number()
if pr:
- click.echo(f"Installed as pr #{pr}, enabling pr mode")
+ print(f"Installed as pr #{pr}, enabling pr mode")
if pr:
if has_previous_pr(pr):
# Uninstall because pipx doesn't update otherwise:
# - https://github.com/pypa/pipx/issues/902
- click.echo(f"Uninstalling previous plextraktsync@{pr}")
+ print(f"Uninstalling previous plextraktsync@{pr}")
execp(f"pipx uninstall plextraktsync@{pr}")
- click.echo(f"Updating PlexTraktSync to the pull request #{pr} version using pipx")
+ print(f"Updating PlexTraktSync to the pull request #{pr} version using pipx")
execp(f"pipx install --suffix=@{pr} --force git+https://github.com/Taxel/PlexTraktSync@refs/pull/{pr}/head")
return
- click.echo("Updating PlexTraktSync to the latest version using pipx")
+ print("Updating PlexTraktSync to the latest version using pipx")
execp("pipx upgrade PlexTraktSync")

MODIFY
plextraktsync/commands/sync.py
plextraktsync/commands/sync.py
from typing import TYPE_CHECKING
-import click
-
from plextraktsync.commands.login import ensure_login
from plextraktsync.decorators.measure_time import measure_time
from plextraktsync.factory import factory, logger
@@ -64,7 +62,7 @@ def sync(
wc.add_movie(movie)
if not wc.is_valid():
- click.echo("Nothing to sync, this is likely due conflicting options given.")
+ print("Nothing to sync, this is likely due conflicting options given.")
return
with measure_time("Completed full sync"):

MODIFY
plextraktsync/commands/trakt_login.py
plextraktsync/commands/trakt_login.py
def trakt_authenticate(api: TraktApi):
- click.echo(title("Sign in to Trakt"))
+ print = factory.print
- click.echo("If you do not have a Trakt client ID and secret:")
- click.echo(" 1 - Open http://trakt.tv/oauth/applications on any computer")
- click.echo(" 2 - Login to your Trakt account")
- click.echo(" 3 - Press the NEW APPLICATION button")
- click.echo(" 4 - Set the NAME field = plex")
- click.echo(" 5 - Set the REDIRECT URL field = urn:ietf:wg:oauth:2.0:oob")
- click.echo(" 6 - Press the SAVE APP button")
- click.echo("")
+ print(title("Sign in to Trakt"))
+
+ print("If you do not have a Trakt client ID and secret:")
+ print(" 1 - Open http://trakt.tv/oauth/applications on any computer")
+ print(" 2 - Login to your Trakt account")
+ print(" 3 - Press the NEW APPLICATION button")
+ print(" 4 - Set the NAME field = plex")
+ print(" 5 - Set the REDIRECT URL field = urn:ietf:wg:oauth:2.0:oob")
+ print(" 6 - Press the SAVE APP button")
+ print("")
while True:
client_id = click.prompt(PROMPT_TRAKT_CLIENT_ID, type=str)
client_secret = click.prompt(PROMPT_TRAKT_CLIENT_SECRET, type=str)
- click.echo("Attempting to authenticate with Trakt")
+ print("Attempting to authenticate with Trakt")
try:
return api.device_auth(client_id=client_id, client_secret=client_secret)
except (ForbiddenException, JSONDecodeError) as e:
- click.echo(error(f"Log in to Trakt failed: {e}, Try again."))
+ print(error(f"Log in to Trakt failed: {e}, Try again."))
def has_trakt_token():
@@ -65,6 +67,7 @@ def trakt_login():
def login():
+ print = factory.print
api = factory.trakt_api
trakt_authenticate(api)
user = api.me.username
@@ -73,4 +76,4 @@ def login():
CONFIG["TRAKT_USERNAME"] = user
CONFIG.save()
- click.echo(TRAKT_LOGIN_SUCCESS)
+ print(TRAKT_LOGIN_SUCCESS)

MODIFY
plextraktsync/commands/unmatched.py
plextraktsync/commands/unmatched.py
-import click
+
from plextraktsync.commands.login import ensure_login
from plextraktsync.factory import factory
@@ -13,7 +13,7 @@ def unmatched(no_progress_bar: bool, local: bool):
walker = factory.walker
if not wc.is_valid():
- click.echo("Nothing to scan, this is likely due conflicting options given.")
+ print("Nothing to scan, this is likely due conflicting options given.")
return
failed = []

Replace click.echo with rich print