fix: Restore console output in non-interactive mode (#226)

podcast_archiver/base.py

@@ -37,7 +37,7 @@ def __init__(self, settings: Settings | None = None, database: BaseDatabase | No
     def register_cleanup(self, ctx: click.RichContext) -> None:
         def _cleanup(signum: int, *args: Any) -> None:
             logger.debug("Signal %s received", signum)
-            rprint("[error]✘ Terminating.[/]")
+            rprint("✘ Terminating", style="error")
             self.processor.shutdown()
             progress_manager.stop()
             ctx.close()
@@ -67,5 +67,5 @@ def run(self) -> int:
             result = self.processor.process(url)
             failures += result.failures
 
-        rprint("\n[completed]Done.[/]\n")
+        rprint("✔ All done", style="completed")
         return failures

podcast_archiver/config.py

@@ -162,7 +162,8 @@ def validate_model(self) -> Settings:
             if getattr(self, name, field.default) == field.default:
                 continue
             rprint(
-                f":warning: Option '{opt_name}' / setting '{name}' is deprecated and {constants.DEPRECATION_MESSAGE}."
+                f"Option '{opt_name}' / setting '{name}' is deprecated and {constants.DEPRECATION_MESSAGE}.",
+                style="warning",
             )
         return self
 

podcast_archiver/console.py

@@ -6,10 +6,13 @@
 _theme = Theme(
     {
         "error": "bold dark_red",
-        "warning": "magenta",
-        "missing": "orange1 bold",
-        "completed": "bold dark_cyan",
+        "warning": "orange1 bold",
+        "warning_hint": "orange1 dim",
+        "completed": "dark_cyan bold",
         "success": "dark_cyan",
+        "present": "dark_cyan",
+        "missing": "orange1",
+        "title": "bright_magenta bold",
     }
 )
 console = Console(theme=_theme)

podcast_archiver/enums.py

@@ -1,6 +1,12 @@
 from __future__ import annotations
 
 from enum import Enum
+from typing import TYPE_CHECKING
+
+from rich.text import Text
+
+if TYPE_CHECKING:
+    from rich.console import RenderableType
 
 
 class StrEnum(str, Enum):
@@ -14,12 +20,38 @@ class QueueCompletionType(StrEnum):
     MAX_EPISODES = "✔ Maximum episode count reached"
     FAILED = "✘ Failed"
 
+    @property
+    def style(self) -> str:
+        if self in self.successful():
+            return "completed"
+        return "error"
+
+    @classmethod
+    def successful(cls) -> set[QueueCompletionType]:
+        return {
+            cls.COMPLETED,
+            cls.FOUND_EXISTING,
+            cls.MAX_EPISODES,
+        }
+
+    def __rich__(self) -> RenderableType:
+        return Text(str(self), style=self.style, end="")
+
 
 class DownloadResult(StrEnum):
-    ALREADY_EXISTS = "Present"
-    COMPLETED_SUCCESSFULLY = "Archived"
-    FAILED = " Failed"
-    ABORTED = "Aborted"
+    ALREADY_EXISTS = "✓ Present"
+    COMPLETED_SUCCESSFULLY = "✓ Archived"
+    MISSING = "✘ Missing"
+    FAILED = "✘ Failed"
+    ABORTED = "✘ Aborted"
+
+    @property
+    def style(self) -> str:
+        if self in self.successful():
+            return "success"
+        if self is self.MISSING:
+            return "missing"
+        return "error"
 
     @classmethod
     def successful(cls) -> set[DownloadResult]:
@@ -28,5 +60,12 @@ def successful(cls) -> set[DownloadResult]:
             cls.COMPLETED_SUCCESSFULLY,
         }
 
-    def __str__(self) -> str:
-        return self.value
+    @classmethod
+    def max_length(cls) -> int:
+        return max(len(v) for v in cls)
+
+    def render_padded(self, padding: str = "   ") -> RenderableType:
+        return Text(f"{self:{self.max_length()}s}{padding}", style=self.style, end="")
+
+    def __rich__(self) -> RenderableType:
+        return self.render_padded()

podcast_archiver/logging.py

@@ -6,8 +6,9 @@
 from os import environ
 from typing import TYPE_CHECKING, Any
 
+from rich.console import Console
+from rich.highlighter import NullHighlighter
 from rich.logging import RichHandler
-from rich.text import Text
 
 from podcast_archiver.console import console
 
@@ -16,18 +17,36 @@
 
 logger = logging.getLogger("podcast_archiver")
 
+plain_console = Console(
+    color_system=None,
+    no_color=True,
+    width=999999,
+    highlighter=NullHighlighter(),
+)
 
 REDIRECT_VIA_LOGGING: bool = False
 
 
-def rprint(*msg: RenderableType, **kwargs: Any) -> None:
+def _make_plain(msg: RenderableType) -> str:
+    with plain_console.capture() as capture:
+        plain_console.print(msg, no_wrap=True)
+    return capture.get().rstrip("\n")
+
+
+def rprint(msg: RenderableType, style: str | None = None, new_line_start: bool = True, **kwargs: Any) -> None:
     if not REDIRECT_VIA_LOGGING:
-        console.print(*msg, **kwargs)
+        console.print(msg, style=style, new_line_start=new_line_start, **kwargs)
         return
 
-    for m in msg:
-        if isinstance(m, Text):
-            logger.info(m.plain.strip())
+    log = logger.info
+    if style == "error":
+        log = logger.error
+    elif style == "warning":
+        log = logger.warning
+
+    for plain in _make_plain(msg).splitlines():
+        if plain:
+            log(plain)
 
 
 def is_interactive() -> bool:

podcast_archiver/models/episode.py

@@ -11,8 +11,7 @@
     field_validator,
     model_validator,
 )
-from rich.table import Table
-from rich.text import Text
+from rich.text import Span, Text
 
 from podcast_archiver.constants import DEFAULT_DATETIME_FORMAT, MAX_TITLE_LENGTH
 from podcast_archiver.exceptions import MissingDownloadUrl
@@ -49,15 +48,7 @@ def __str__(self) -> str:
         return f"{self.published_time.strftime(DEFAULT_DATETIME_FORMAT)} {self.title}"
 
     def __rich__(self) -> RenderableType:
-        """Makes the Progress class itself renderable."""
-        grid = Table.grid()
-        grid.add_column(style="dim")
-        grid.add_column()
-        grid.add_row(
-            Text(f"{self.published_time:%Y-%m-%d} "),
-            Text(self.title, overflow="ellipsis", no_wrap=True),
-        )
-        return grid
+        return Text(f"{self.published_time:%Y-%m-%d} {self.title}", spans=[Span(0, 10, "dim")], end="")
 
     @field_validator("title", mode="after")
     @classmethod

podcast_archiver/models/feed.py

@@ -108,8 +108,8 @@ def parse_feed(cls, source: str | bytes, alt_url: str | None) -> FeedPage:
         obj = cls.model_validate(feedobj)
         if obj.bozo and (exc := obj.bozo_exception) and isinstance(exc, SAXParseException):
             url = source if isinstance(source, str) and not alt_url else alt_url
-            rprint(f"[orange1]Feed content is not well-formed for {url}[/]")
-            rprint(f"[orange1 dim]Continuing processing but here be dragons ({exc.getMessage()})[/]")
+            rprint(f"Feed content is not well-formed for {url}", style="warning")
+            rprint(f"Continuing processing but here be dragons ({exc.getMessage()})", style="warning_hint")
         return obj
 
     @classmethod

podcast_archiver/processor.py

@@ -4,9 +4,6 @@
 from threading import Event
 from typing import TYPE_CHECKING
 
-from rich.console import Group
-from rich.text import Text
-
 from podcast_archiver import constants
 from podcast_archiver.config import Settings
 from podcast_archiver.database import get_database
@@ -56,7 +53,7 @@ def process(self, url: str) -> ProcessingResult:
             return ProcessingResult(feed=None, tombstone=QueueCompletionType.FAILED)
 
         result = self.process_feed(feed=feed)
-        rprint(result.tombstone, style="completed")
+        rprint(result, end="\n\n")
         return result
 
     def load_feed(self, url: str, known_feeds: dict[str, FeedInfo]) -> Feed | None:
@@ -104,7 +101,7 @@ def _does_already_exist(self, episode: BaseEpisode, *, target: Path) -> bool:
         return True
 
     def process_feed(self, feed: Feed) -> ProcessingResult:
-        rprint(f"\n[bold bright_magenta]Archiving: {feed}[/]\n")
+        rprint(f"→ Processing: {feed}", style="title")
         tombstone = QueueCompletionType.COMPLETED
         results: EpisodeResultsList = []
         with PrettyPrintEpisodeRange() as pretty_range:
@@ -153,14 +150,12 @@ def _handle_results(self, episode_results: EpisodeResultsList) -> tuple[int, int
                 continue
 
             if episode_result.result in DownloadResult.successful():
-                prefix = Text(f"✔ {episode_result.result} ", style="success", end=" ")
                 success += 1
                 self.database.add(episode_result.episode)
             else:
-                prefix = Text(f"✖ {episode_result.result} ", style="error", end=" ")
                 failures += 1
 
-            rprint(Group(prefix, episode_result.episode))
+            rprint(episode_result, new_line_start=False)
         return success, failures
 
     def shutdown(self) -> None:

podcast_archiver/types.py

@@ -4,7 +4,11 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Protocol, TypeAlias
 
+from rich.console import Group
+
 if TYPE_CHECKING:
+    from rich.console import RenderableType
+
     from podcast_archiver.enums import DownloadResult, QueueCompletionType
     from podcast_archiver.models.episode import BaseEpisode
     from podcast_archiver.models.feed import Feed
@@ -16,6 +20,9 @@ class EpisodeResult:
     result: DownloadResult
     is_eager: bool = False
 
+    def __rich__(self) -> RenderableType:
+        return Group(self.result, self.episode)
+
 
 @dataclass(slots=True, frozen=True)
 class ProcessingResult:
@@ -24,6 +31,9 @@ class ProcessingResult:
     success: int = 0
     failures: int = 0
 
+    def __rich__(self) -> RenderableType:
+        return self.tombstone
+
 
 class ProgressCallback(Protocol):
     def __call__(self, total: int | None = None, completed: int | None = None) -> None: ...

podcast_archiver/utils/__init__.py

@@ -3,6 +3,7 @@
 import os
 import re
 from contextlib import contextmanager
+from functools import partial
 from string import Formatter
 from typing import IO, TYPE_CHECKING, Any, Generator, Iterable, Iterator, Literal, TypedDict, overload
 
@@ -140,27 +141,28 @@ def atomic_write(target: Path, mode: Literal["w", "wb"] = "w") -> Iterator[IO[by
 
 @contextmanager
 def handle_feed_request(url: str) -> Generator[None, Any, None]:
+    printerr = partial(rprint, style="error")
     try:
         yield
     except HTTPError as exc:
         logger.debug("Failed to request feed url %s", url, exc_info=exc)
         if (response := getattr(exc, "response", None)) is None:
-            rprint(f"[error]Failed to retrieve feed {url}: {exc}[/]")
+            printerr(f"Failed to retrieve feed {url}: {exc}")
             return
 
-        rprint(f"[error]Received status code {response.status_code} from {url}[/]")
+        printerr(f"Received status code {response.status_code} from {url}")
 
     except ValidationError as exc:
         logger.debug("Feed validation failed for %s", url, exc_info=exc)
-        rprint(f"[error]Received invalid feed from {url}[/]")
+        printerr(f"Received invalid feed from {url}")
 
     except NotModified as exc:
         logger.debug("Skipping retrieval for %s", exc.info)
-        rprint(f"\n[bar.finished]⏲ Feed of {exc.info} is unchanged, skipping.[/]")
+        rprint(f"⏲ Feed of {exc.info} is unchanged, skipping.", style="success")
 
     except Exception as exc:
         logger.debug("Unexpected error for url %s", url, exc_info=exc)
-        rprint(f"[error]Failed to retrieve feed {url}: {exc}[/]")
+        printerr(f"Failed to retrieve feed {url}: {exc}")
 
 
 def get_field_titles() -> str: