Skip to content

Commit

Permalink
refactor: Simplify settings parsing
Browse files Browse the repository at this point in the history
  • Loading branch information
@janw
janw committed Mar 30, 2024
1 parent 26061bd commit a4e4b9a
Show file tree
Hide file tree
Showing 17 changed files with 331 additions and 349 deletions.
31 changes: 19 additions & 12 deletions podcast_archiver/base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,10 @@
from pathlib import Path
from typing import TYPE_CHECKING

from pydantic import AnyHttpUrl
from pydantic import AnyHttpUrl, ValidationError

from podcast_archiver.console import console
from podcast_archiver.exceptions import InvalidFeed
from podcast_archiver.logging import logger
from podcast_archiver.processor import FeedProcessor

Expand All @@ -22,27 +23,33 @@ class PodcastArchiver:

def __init__(self, settings: Settings):
self.settings = settings
self.processor = FeedProcessor(settings=self.settings)
self.processor = FeedProcessor(settings=settings)

logger.debug("Initializing with settings: %s", settings)

self.feeds = set()
for feed in self.settings.feeds:
self.add_feed(feed)
for opml in self.settings.opml_files:
self.add_from_opml(opml)
try:
self.feeds = set()
for feed in self.settings.feeds:
self.add_feed(feed)
for opml in self.settings.opml_files:
self.add_from_opml(opml)
except ValidationError as exc:
raise InvalidFeed(feed=exc.errors()[0]["input"]) from exc

def register_cleanup(self, ctx: click.RichContext) -> None:
@ctx.call_on_close
def _cleanup() -> None:
self.processor.shutdown()

def add_feed(self, feed: Path | AnyHttpUrl) -> None:
def add_feed(self, feed: Path | AnyHttpUrl | str) -> None:
if isinstance(feed, Path):
with open(feed, "r") as fp:
self.feeds.union(set(fp.read().strip().splitlines()))
else:
self.feeds.add(feed)
for f in fp.read().strip().splitlines():
self.add_feed(f)
return
if isinstance(feed, str):
feed = AnyHttpUrl(feed)
self.feeds.add(feed)

def add_from_opml(self, opml: Path) -> None:
with opml.open("r") as file:
Expand All @@ -51,7 +58,7 @@ def add_from_opml(self, opml: Path) -> None:
# TODO: Move parsing to pydantic
for elem in tree.findall(".//outline[@type='rss'][@xmlUrl!='']"):
if url := elem.get("xmlUrl"):
self.add_feed(AnyHttpUrl(url))
self.add_feed(url)

def run(self) -> int:
failures = 0
Expand Down
156 changes: 63 additions & 93 deletions podcast_archiver/cli.py
View file
Original file line number Diff line number Diff line change
@@ -1,17 +1,22 @@
import pathlib
from os import PathLike, getenv
from typing import Any, cast
from typing import Any

import rich_click as click
from click.core import Context, Parameter

from podcast_archiver import __version__ as version
from podcast_archiver import constants
from podcast_archiver.base import PodcastArchiver
from podcast_archiver.config import DEFAULT_SETTINGS, Settings
from podcast_archiver.config import (
ConfigPath,
Settings,
get_default_config_path,
print_default_config,
)
from podcast_archiver.console import console
from podcast_archiver.constants import ENVVAR_PREFIX, PROG_NAME
from podcast_archiver.exceptions import InvalidSettings
from podcast_archiver.exceptions import InvalidFeed, InvalidSettings
from podcast_archiver.logging import configure_logging
from podcast_archiver.models import ALL_FIELD_TITLES_STR

click.rich_click.USE_RICH_MARKUP = True
click.rich_click.USE_MARKDOWN = True
Expand Down Expand Up @@ -46,64 +51,6 @@
}


class ConfigPath(click.Path):
def __init__(self) -> None:
return super().__init__(
exists=True,
readable=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
path_type=pathlib.Path,
)

def convert( # type: ignore[override]
self, value: str | PathLike[str], param: Parameter | None, ctx: Context | None
) -> str | bytes | PathLike[str] | None:
if value is None:
return None
if (
ctx
and param
and isinstance(value, pathlib.Path)
and value == param.get_default(ctx, call=True)
and not value.exists()
):
try:
value.parent.mkdir(exist_ok=True, parents=True)
with value.open("w") as fp:
Settings.generate_default_config(file=fp)
except (OSError, FileNotFoundError):
return None

filepath = cast(pathlib.Path, super().convert(value, param, ctx))
if not ctx or ctx.resilient_parsing:
return filepath

try:
ctx.default_map = ctx.default_map or {}
settings = Settings.load_from_yaml(filepath)
ctx.default_map.update(settings.model_dump(exclude_unset=True, exclude_none=True, by_alias=True))
except InvalidSettings as exc:
self.fail(f"{self.name.title()} {click.format_filename(filepath)!r} is invalid: {exc}", param, ctx)

return filepath


def get_default_config_path() -> pathlib.Path | None:
if getenv("TESTING", "0").lower() in ("1", "true"):
return None
return pathlib.Path(click.get_app_dir(PROG_NAME)) / "config.yaml" # pragma: no cover


def generate_default_config(ctx: click.Context, param: click.Parameter, value: bool) -> None:
if not value or ctx.resilient_parsing:
return

Settings.generate_default_config()
ctx.exit()


@click.command(
context_settings={
"auto_envvar_prefix": ENVVAR_PREFIX,
Expand All @@ -114,25 +61,37 @@ def generate_default_config(ctx: click.Context, param: click.Parameter, value: b
@click.option(
"-f",
"--feed",
"feeds",
default=[],
multiple=True,
show_envvar=True,
help=Settings.model_fields["feeds"].description + " Use repeatedly for multiple feeds.", # type: ignore[operator]
help="Feed URLs to archive. Use repeatedly for multiple feeds.",
)
@click.option(
"-o",
"--opml",
"opml_files",
type=click.Path(
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
path_type=pathlib.Path,
),
default=[],
multiple=True,
show_envvar=True,
help=(
Settings.model_fields["opml_files"].description # type: ignore[operator]
+ " Use repeatedly for multiple files."
"OPML files containing feed URLs to archive. OPML files can be exported from a variety of podcatchers."
"Use repeatedly for multiple files."
),
)
@click.option(
"-d",
"--dir",
"archive_directory",
type=click.Path(
exists=False,
exists=True,
writable=True,
file_okay=False,
dir_okay=True,
Expand All @@ -141,85 +100,96 @@ def generate_default_config(ctx: click.Context, param: click.Parameter, value: b
),
show_default=True,
required=False,
default=DEFAULT_SETTINGS.archive_directory,
default=pathlib.Path("."),
show_envvar=True,
help=Settings.model_fields["archive_directory"].description,
help=(
"Directory to which to download the podcast archive. "
"By default, the archive will be created in the current working directory ('.')."
),
)
@click.option(
"-F",
"--filename-template",
type=str,
show_default=True,
required=False,
default=DEFAULT_SETTINGS.filename_template,
default=constants.DEFAULT_FILENAME_TEMPLATE,
show_envvar=True,
help=Settings.model_fields["filename_template"].description,
help=(
"Template to be used when generating filenames. Available template variables are: "
f"{ALL_FIELD_TITLES_STR}, and 'ext' (the filename extension)."
),
)
@click.option(
"-u",
"--update",
"update_archive",
type=bool,
default=DEFAULT_SETTINGS.update_archive,
is_flag=True,
show_envvar=True,
help=Settings.model_fields["update_archive"].description,
help=(
"Update the feeds with newly added episodes only. "
"Adding episodes ends with the first episode already present in the download directory."
),
)
@click.option(
"--write-info-json",
type=bool,
default=DEFAULT_SETTINGS.write_info_json,
is_flag=True,
show_envvar=True,
help=Settings.model_fields["write_info_json"].description,
help="Write episode metadata to a .info.json file next to the media file itself.",
)
@click.option(
"-q",
"--quiet",
type=bool,
default=DEFAULT_SETTINGS.quiet,
is_flag=True,
show_envvar=True,
help=Settings.model_fields["quiet"].description,
help="Print only minimal progress information. Errors will always be emitted.",
)
@click.option(
"-C",
"--concurrency",
type=int,
default=DEFAULT_SETTINGS.concurrency,
default=constants.DEFAULT_CONCURRENCY,
show_envvar=True,
help=Settings.model_fields["concurrency"].description,
help="Maximum number of simultaneous downloads.",
)
@click.option(
"--debug-partial",
type=bool,
default=DEFAULT_SETTINGS.debug_partial,
is_flag=True,
show_envvar=True,
help=Settings.model_fields["debug_partial"].description,
help=f"Download only the first {constants.DEBUG_PARTIAL_SIZE} bytes of episodes for debugging purposes.",
)
@click.option(
"-v",
"--verbose",
count=True,
show_envvar=True,
default=DEFAULT_SETTINGS.verbose,
help=Settings.model_fields["verbose"].description,
is_eager=True,
callback=configure_logging,
help="Increase the level of verbosity while downloading.",
)
@click.option(
"-S",
"--slugify",
"slugify_paths",
type=bool,
default=DEFAULT_SETTINGS.slugify_paths,
is_flag=True,
show_envvar=True,
help=Settings.model_fields["slugify_paths"].description,
help="Format filenames in the most compatible way, replacing all special characters.",
)
@click.option(
"-m",
"--max-episodes",
"maximum_episode_count",
type=int,
default=DEFAULT_SETTINGS.maximum_episode_count,
help=Settings.model_fields["maximum_episode_count"].description,
default=0,
help=(
"Only download the given number of episodes per podcast feed. "
"Useful if you don't really need the entire backlog."
),
)
@click.version_option(
version,
Expand All @@ -233,7 +203,7 @@ def generate_default_config(ctx: click.Context, param: click.Parameter, value: b
expose_value=False,
is_flag=True,
is_eager=True,
callback=generate_default_config,
callback=print_default_config,
help="Emit an example YAML config file to stdout and exit.",
)
@click.option(
Expand All @@ -248,12 +218,10 @@ def generate_default_config(ctx: click.Context, param: click.Parameter, value: b
help="Path to a config file. Command line arguments will take precedence.",
)
@click.pass_context
def main(ctx: click.RichContext, /, **kwargs: Any) -> int:
configure_logging(kwargs["verbose"])
console.quiet = kwargs["quiet"] or kwargs["verbose"] > 1
def main(ctx: click.RichContext, **kwargs: Any) -> int:
settings = Settings(**kwargs)
console.quiet = settings.quiet or settings.verbose > 1
try:
settings = Settings.load_from_dict(kwargs)

# Replicate click's `no_args_is_help` behavior but only when config file does not contain feeds/OPMLs
if not (settings.feeds or settings.opml_files):
click.echo(ctx.command.get_help(ctx))
Expand All @@ -262,6 +230,8 @@ def main(ctx: click.RichContext, /, **kwargs: Any) -> int:
pa = PodcastArchiver(settings=settings)
pa.register_cleanup(ctx)
pa.run()
except InvalidFeed as exc:
raise click.BadParameter(f"Cannot parse feed '{exc.feed}'") from exc
except InvalidSettings as exc:
raise click.BadParameter(f"Invalid settings: {exc}") from exc
except KeyboardInterrupt as exc:
Expand Down
Loading

0 comments on commit a4e4b9a

Please sign in to comment.