diff --git a/requirements/project.md b/requirements/project.md index dfcac9c..73bbe47 100644 --- a/requirements/project.md +++ b/requirements/project.md @@ -18,7 +18,7 @@ - Inspect existing media files through `ffprobe` and compare discovered stream metadata with stored normalization rules. - Convert media files through `ffmpeg` into a normalized output layout, including video recoding, audio transcoding to Opus, metadata cleanup and rewrite, and controlled disposition flags. - Build output filenames from detected or configured show, season, and episode information, optionally enriched from TMDB and a configurable Jinja-style filename template. - - Support auxiliary file operations such as subtitle import, unmuxing, crop detection, and rename-only runs. + - Support auxiliary file operations such as subtitle import, unmuxing, crop detection, rename-only conversion runs, and direct in-place episode renaming. - Supported environments: - Local execution on a Python-capable workstation. - Best-supported on Linux-like systems because the implementation assumes `~/.local`, `/dev/null`, `nice`, and `cpulimit`. @@ -35,7 +35,7 @@ ## Functional Requirements -- The system shall provide a CLI entrypoint named `ffx` with commands for `convert`, `inspect`, `shows`, `unmux`, `cropdetect`, `setup`, `configure_workstation`, `upgrade`, `version`, and `help`. +- The system shall provide a CLI entrypoint named `ffx` with commands for `convert`, `inspect`, `shows`, `rename`, `unmux`, `cropdetect`, `setup`, `configure_workstation`, `upgrade`, `version`, and `help`. - The system shall support a two-step local installation and preparation flow: - `tools/setup.sh` is the bootstrap entrypoint for the first step and shall own bundle virtualenv creation, package installation, shell alias exposure, and optional Python test-package installation. - `tools/configure_workstation.sh` is the bootstrap entrypoint for the second step and shall own workstation dependency checks and installation plus local config and directory seeding. @@ -69,6 +69,7 @@ - `--cpu` shall accept either a positive absolute `cpulimit` value such as `200`, or a percentage suffixed with `%` such as `25%` to represent a share of present CPUs; omitting the option or using `0` shall disable CPU limiting. - When both limits are configured, the process wrapper shall execute the target command through `cpulimit` around a `nice -n ...` invocation so both limits apply to the launched media command. - The system shall support extracting streams into separate files via `unmux` and reporting suggested crop parameters via `cropdetect`. +- The system shall support in-place episode renaming via `rename`, requiring a `--prefix`, accepting optional `--season` and `--suffix` overrides, preserving the source extension, and supporting dry-run output without moving files. - Crop detection shall use a configurable sampling window, defaulting to a 60-second seek and a 180-second analysis duration, and repeated crop-detection requests for the same source plus sampling window shall reuse cached results within one process. - The system shall handle invalid input and system failures gracefully by logging warnings or raising `click` errors for missing files, invalid media, missing TMDB credentials, incompatible database versions, and ambiguous track dispositions when prompting is disabled. diff --git a/src/ffx/cli.py b/src/ffx/cli.py index 6703198..e3c3594 100755 --- a/src/ffx/cli.py +++ b/src/ffx/cli.py @@ -33,7 +33,7 @@ if TYPE_CHECKING: from ffx.media_descriptor import MediaDescriptor from ffx.track_descriptor import TrackDescriptor -LIGHTWEIGHT_COMMANDS = {None, 'version', 'help', 'setup', 'configure_workstation', 'upgrade'} +LIGHTWEIGHT_COMMANDS = {None, 'version', 'help', 'setup', 'configure_workstation', 'upgrade', 'rename'} CPU_OPTION_HELP = ( "Limit CPU for started processes. Use an absolute cpulimit value such as 200 " + "(about 2 cores), or use a percentage such as 25% for a share of present cores. " @@ -185,6 +185,33 @@ def resolveUnmuxOutputDirectory(context, outputDirectory, subtitlesOnly, label): return os.path.join(configuredSubtitlesBaseDirectory, resolvedLabel), True +def buildRenameTargetFilename(sourcePath, prefix, seasonOverride=None, suffix=''): + from ffx.file_properties import FileProperties + + sourceFilename = os.path.basename(sourcePath) + seasonEpisodeValues = FileProperties.extractSeasonEpisodeValues(sourceFilename) + if seasonEpisodeValues is None: + return None + + sourceSeason, sourceEpisode = seasonEpisodeValues + resolvedSeason = int(seasonOverride) if seasonOverride is not None else ( + int(sourceSeason) if sourceSeason is not None else 1 + ) + + _sourceBasename, sourceExtension = os.path.splitext(sourceFilename) + + targetFilenameTokens = [ + str(prefix).strip(), + f"s{resolvedSeason}e{int(sourceEpisode)}", + ] + + resolvedSuffix = str(suffix).strip() + if resolvedSuffix: + targetFilenameTokens.append(resolvedSuffix) + + return f"{'_'.join(targetFilenameTokens)}{sourceExtension}" + + @click.group() @click.pass_context @@ -242,7 +269,7 @@ def version(): def help(): click.echo(f"ffx {VERSION}\n") click.echo("Maintenance commands: setup, configure_workstation, upgrade") - click.echo("Media commands: shows, inspect, convert, unmux, cropdetect") + click.echo("Media commands: shows, inspect, convert, rename, unmux, cropdetect") click.echo("Use 'ffx --help' or 'ffx --help' for full command help.") @@ -408,6 +435,55 @@ def inspect(ctx, filename): app.run() +@ffx.command() +@click.pass_context +@click.argument('paths', nargs=-1) +@click.option('--prefix', type=str, required=True, help='Required target filename prefix') +@click.option('--season', type=int, default=None, help='Override target season index') +@click.option('--suffix', type=str, default='', help='Optional target filename suffix') +@click.option('--dry-run', is_flag=True, default=False, help='Only print planned renames') +def rename(ctx, paths, prefix, season, suffix, dry_run): + """Rename matching episode files in place.""" + + resolvedPrefix = str(prefix).strip() + resolvedSuffix = str(suffix).strip() + effectiveDryRun = bool(ctx.obj.get('dry_run', False) or dry_run) + + if not resolvedPrefix: + raise click.ClickException("Rename prefix must not be empty.") + + processedCount = 0 + + for sourcePath in paths: + if not os.path.isfile(sourcePath): + continue + + targetFilename = buildRenameTargetFilename( + sourcePath, + resolvedPrefix, + seasonOverride=season, + suffix=resolvedSuffix, + ) + if targetFilename is None: + continue + + sourceFilename = os.path.basename(sourcePath) + targetPath = os.path.join(os.path.dirname(sourcePath), targetFilename) + click.echo(f"{sourceFilename} -> {targetFilename}") + processedCount += 1 + + if effectiveDryRun or os.path.abspath(sourcePath) == os.path.abspath(targetPath): + continue + + if os.path.exists(targetPath): + raise click.ClickException(f"Target file already exists: {targetPath}") + + shutil.move(sourcePath, targetPath) + + if processedCount == 0: + click.echo("No matching files found.") + + def getUnmuxSequence(trackDescriptor: TrackDescriptor, sourcePath, targetPrefix, targetDirectory = ''): # executable and input file diff --git a/src/ffx/file_properties.py b/src/ffx/file_properties.py index 2f8d0af..20c5d94 100644 --- a/src/ffx/file_properties.py +++ b/src/ffx/file_properties.py @@ -30,6 +30,18 @@ class FileProperties(): DEFAULT_INDEX_DIGITS = 3 + @classmethod + def extractSeasonEpisodeValues(cls, sourceText: str) -> tuple[int | None, int] | None: + seasonEpisodeMatch = re.search(cls.SEASON_EPISODE_INDICATOR_MATCH, str(sourceText)) + if seasonEpisodeMatch is not None: + return int(seasonEpisodeMatch.group(1)), int(seasonEpisodeMatch.group(2)) + + episodeMatch = re.search(cls.EPISODE_INDICATOR_MATCH, str(sourceText)) + if episodeMatch is not None: + return None, int(episodeMatch.group(1)) + + return None + def __init__(self, context, sourcePath): self.context = context @@ -65,26 +77,19 @@ class FileProperties(): databaseMatchedGroups = matchResult['match'].groups() self.__logger.debug(f"FileProperties.__init__(): Matched groups: {databaseMatchedGroups}") - seIndicator = databaseMatchedGroups[0] - - se_match = re.search(FileProperties.SEASON_EPISODE_INDICATOR_MATCH, seIndicator) - e_match = re.search(FileProperties.EPISODE_INDICATOR_MATCH, seIndicator) - + indicatorSource = databaseMatchedGroups[0] else: self.__logger.debug(f"FileProperties.__init__(): Checking file name for indicator {self.__sourceFilename}") + indicatorSource = self.__sourceFilename - se_match = re.search(FileProperties.SEASON_EPISODE_INDICATOR_MATCH, self.__sourceFilename) - e_match = re.search(FileProperties.EPISODE_INDICATOR_MATCH, self.__sourceFilename) - - if se_match is not None: - self.__season = int(se_match.group(1)) - self.__episode = int(se_match.group(2)) - elif e_match is not None: - self.__season = -1 - self.__episode = int(e_match.group(1)) - else: + seasonEpisodeValues = self.extractSeasonEpisodeValues(indicatorSource) + if seasonEpisodeValues is None: self.__season = -1 self.__episode = -1 + else: + sourceSeason, sourceEpisode = seasonEpisodeValues + self.__season = -1 if sourceSeason is None else int(sourceSeason) + self.__episode = int(sourceEpisode) self.__ffprobeData = None diff --git a/tests/unit/test_cli_rename.py b/tests/unit/test_cli_rename.py new file mode 100644 index 0000000..5d0d57e --- /dev/null +++ b/tests/unit/test_cli_rename.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from pathlib import Path +import sys +import tempfile +import unittest + +from click.testing import CliRunner + + +SRC_ROOT = Path(__file__).resolve().parents[2] / "src" + +if str(SRC_ROOT) not in sys.path: + sys.path.insert(0, str(SRC_ROOT)) + + +from ffx import cli # noqa: E402 + + +class RenameCliTests(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.workspace = Path(self.tempdir.name) + + def tearDown(self): + self.tempdir.cleanup() + + def write_source(self, filename: str, payload: bytes = b"episode") -> Path: + source_path = self.workspace / filename + source_path.write_bytes(payload) + return source_path + + def invoke_rename(self, *args: str): + runner = CliRunner() + result = runner.invoke(cli.ffx, ["rename", *args]) + self.assertEqual(0, result.exit_code, result.output) + return result + + def test_rename_moves_matching_file_in_place(self): + source_path = self.write_source("demo_S02E03.mkv", b"season-episode") + + result = self.invoke_rename("--prefix", "dball", str(source_path)) + + target_path = self.workspace / "dball_s2e3.mkv" + self.assertIn("demo_S02E03.mkv -> dball_s2e3.mkv", result.output) + self.assertFalse(source_path.exists()) + self.assertTrue(target_path.exists()) + self.assertEqual(b"season-episode", target_path.read_bytes()) + + def test_rename_uses_default_season_and_suffix_for_episode_only_match(self): + source_path = self.write_source("demo_E07.mp4", b"episode-only") + + result = self.invoke_rename( + "--prefix", + "dball", + "--suffix", + "bonus", + str(source_path), + ) + + target_path = self.workspace / "dball_s1e7_bonus.mp4" + self.assertIn("demo_E07.mp4 -> dball_s1e7_bonus.mp4", result.output) + self.assertFalse(source_path.exists()) + self.assertTrue(target_path.exists()) + self.assertEqual(b"episode-only", target_path.read_bytes()) + + def test_rename_cli_season_overrides_source_season(self): + source_path = self.write_source("demo_s02e07.webm") + + result = self.invoke_rename( + "--prefix", + "dball", + "--season", + "5", + str(source_path), + ) + + target_path = self.workspace / "dball_s5e7.webm" + self.assertIn("demo_s02e07.webm -> dball_s5e7.webm", result.output) + self.assertFalse(source_path.exists()) + self.assertTrue(target_path.exists()) + + def test_rename_dry_run_prints_mapping_without_moving(self): + source_path = self.write_source("demo_E07.mkv") + + result = self.invoke_rename( + "--dry-run", + "--prefix", + "dball", + str(source_path), + ) + + target_path = self.workspace / "dball_s1e7.mkv" + self.assertIn("demo_E07.mkv -> dball_s1e7.mkv", result.output) + self.assertTrue(source_path.exists()) + self.assertFalse(target_path.exists()) + + def test_rename_skips_non_matching_filenames(self): + source_path = self.write_source("demo_finale.mkv") + + result = self.invoke_rename("--prefix", "dball", str(source_path)) + + self.assertIn("No matching files found.", result.output) + self.assertTrue(source_path.exists()) + + +if __name__ == "__main__": + unittest.main()