source/reprocess_file/plugin.py - master - kushpel/plugins

source/reprocess_file/plugin.py
305 строк13 KB
yajrendrag
add regex path_map option
10 апр 2026, 03:56
10 апр 2026, 03:56c937a1a
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
    Written by:               yajrendrag <yajdude@gmail.com>
    Date:                     16 November 2025, (11:00 AM)

    Copyright:
        Copyright (C) 2025 Jay Gardner

        This program is free software: you can redistribute it and/or modify it under the terms of the GNU General
        Public License as published by the Free Software Foundation, version 3.

        This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the
        implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
        for more details.

        You should have received a copy of the GNU General Public License along with this program.
        If not, see <https://www.gnu.org/licenses/>.

"""

import logging
import re
import threading
import time
import subprocess
import json
import os
from pathlib import Path

from unmanic.libs.unplugins.settings import PluginSettings

logger = logging.getLogger("Unmanic.Plugin.reprocess_file")


def delayed_create_task(abspath, library_id, max_retries=10, delay=1):
    """
    Create a task with retry logic, waiting for any existing task to complete.

    When this plugin runs during post-processing, the current task still exists
    in the database (it gets deleted after the plugin returns). This function
    runs in a background thread and retries task creation until the original
    task is deleted and the new task can be created.

    :param abspath: Absolute path to the file to add to the task queue
    :param library_id: Target library ID for the new task
    :param max_retries: Maximum number of retry attempts
    :param delay: Delay in seconds between retry attempts
    :return: True if task was created, False otherwise
    """
    from unmanic.libs.task import Task

    for attempt in range(max_retries):
        time.sleep(delay)
        task = Task()
        if task.create_task_by_absolute_path(abspath, task_type='local', library_id=library_id):
            logger.info(f"Successfully created task for '{abspath}' in library {library_id}")
            return True
        logger.debug(f"Attempt {attempt + 1}/{max_retries}: Task creation failed for '{abspath}', retrying...")

    logger.error(f"Failed to create task for '{abspath}' in library {library_id} after {max_retries} attempts")
    return False

class Settings(PluginSettings):
    settings = {
        "target_library": "Select",
        "reprocess_based_on_task_status": True,
        "status_that_adds_file_to_queue": "Failed",
        "diagnostic_script": "",
        "change_suffix": True,
        "new_suffix": "",
        "modify_path": True,
        "path_map": "",
        "use_regex_path_map": False,
    }

    def __init__(self, *args, **kwargs):
        super(Settings, self).__init__(*args, **kwargs)
        self.form_settings = {
            "target_library": self.__set_target_library_form_settings(),
            "reprocess_based_on_task_status":  {
                "label": "Check this option to only add the file back to the queue if it's completion status matches your configured result.  Otherwise the file is always added back to the task queue",
            },
            "status_that_adds_file_to_queue": self.__set_status_that_adds_file_to_queue_form_settings(),
            "diagnostic_script": self.__set_diagnostic_script_form_settings(),
            "change_suffix":  {
                "label": "Check this option to process a file with the same name, but a different suffix - after checking this you'll be able to enter a new suffix",
            },
            "new_suffix": self.__set_new_suffix_form_settings(),
            "modify_path":  {
                "label": "Check this option to replace a component in the file path - after checking this you'll enter a mapping of old leading path:new leading path",
            },
            "path_map": self.__set_path_map_form_settings(),
            "use_regex_path_map": self.__set_use_regex_path_map_form_settings(),
        }

    def __set_target_library_form_settings(self):
        values = {
            "description": "Select the library to send the source file to for additional processing",
            "label": "Target Library",
            "input_type": "select",
            "select_options": self.get_library_options(),
        }
        return values

    def __set_status_that_adds_file_to_queue_form_settings(self):
        values = {
            "description": "Specify which task status - success or failure - is cause for reprocessing the file",
            "label": "Task Status",
            "input_type": "select",
            "select_options": [
                {
                    "value": "success",
                    "label": "Success",
                },
                {
                    "value": "failed",
                    "label": "Failed",
                },
            ],
        }
        if not self.get_setting('reprocess_based_on_task_status'):
            values["display"] = 'hidden'
        return values

    def __set_diagnostic_script_form_settings(self):
        values = {
            "label": "Enter the path to a shell script to run that returns true or false, e.g., /config/reprocess_script.sh",
            "description": "If the script returns true, the file will be reprocessed.  if left empty, this test will be ignored.  This option is only available when reprocessing by task status",
            "input_type": "textarea",
        }
        if not self.get_setting('reprocess_based_on_task_status'):
            values["display"] = 'hidden'
        return values

    def __set_new_suffix_form_settings(self):
        values = {
            "label": "Specify a new suffix for the file that is to be reprocessed",
            "description": "multi-suffix formats ok, e.g., .language.srt",
            "input_type": "textarea",
        }
        if not self.get_setting('change_suffix'):
            values["display"] = 'hidden'
        return values

    def __set_path_map_form_settings(self):
        use_regex = self.get_setting('use_regex_path_map')
        if use_regex:
            description = (
                "Enter a regex pattern and replacement separated by a colon. "
                "The pattern is matched against the full file path using re.sub(). "
                "Use capture groups and backreferences (\\1, \\2) in the replacement. "
                "For example, to change 'video_file' to 'video_file2' anywhere in the path, "
                "enter: (video_file):\\g<1>2 — or to swap a season folder, "
                "enter: (Season )\\d+:\\g<1>99"
            )
        else:
            description = (
                "for example, if you want to map /library/TVShows in a path like "
                "/library/TVShows/Showname/Season X/Showname - SXEY - Episode Title.mp4 "
                "to /Moved Shows, enter the map as /library/TVShows:/Moved Shows here"
            )
        values = {
            "label": "Specify a path map translation to be made",
            "description": description,
            "input_type": "textarea",
        }
        if not self.get_setting('modify_path'):
            values["display"] = 'hidden'
        return values

    def __set_use_regex_path_map_form_settings(self):
        values = {
            "label": "Use regex patterns in path map",
            "description": "When enabled, the path map uses regex: the left side of the colon is a regex pattern and the right side is the replacement (supports backreferences like \\1, \\g<1>)",
        }
        if not self.get_setting('modify_path'):
            values["display"] = 'hidden'
        return values


    def get_library_options(self):
        from unmanic.libs.library import Library

        options = []
        libraries = Library.get_all_libraries()

        for lib in libraries:
            options.append({
                'value': lib['id'],
                'label': f"{lib['name']} (ID: {lib['id']})"
            })

        return options


def on_postprocessor_task_results(data):
    """
    Add the original source file to another library's pending tasks

    The 'data' object argument includes:
        library_id                      - The library that the current task is associated with.
        task_id                         - Integer, unique identifier of the task.
        task_type                       - String, "local" or "remote".
        final_cache_path                - The path to the final cache file that was then used as the source for all destination files.
        task_processing_success         - Boolean, did all task processes complete successfully.
        file_move_processes_success     - Boolean, did all postprocessor movement tasks complete successfully.
        destination_files               - List containing all file paths created by postprocessor file movements.
        source_data                     - Dictionary containing data pertaining to the original source file.
        start_time                      - Float, UNIX timestamp when the task began.
        finish_time                     - Float, UNIX timestamp when the task completed.

    :param data:
    :return:
    """
    # Configure settings object (maintain compatibility with v1 plugins)
    if data.get('library_id'):
        settings = Settings(library_id=data.get('library_id'))
    else:
        settings = Settings()

    reprocess_all = not settings.get_setting("reprocess_based_on_task_status")
    change_suffix = settings.get_setting("change_suffix")
    if change_suffix:
        new_suffix = settings.get_setting("new_suffix")

    modify_path = settings.get_setting("modify_path")
    if modify_path:
        path_map = settings.get_setting("path_map")

    logger.debug(f"reprocess_all: {reprocess_all}")

    if not reprocess_all:
        reprocess_which = settings.get_setting("status_that_adds_file_to_queue")
        script_path = settings.get_setting('diagnostic_script')

    if reprocess_all or (reprocess_which == "failed" and not data.get("task_processing_success")) or (reprocess_which == "success" and data.get("task_processing_success")):

        # Get library
        target_library_id = settings.get_setting('target_library')

        # Get the original source file
        abspath = Path(data.get('source_data').get('abspath'))
        if change_suffix:
            abspath = abspath.with_suffix(new_suffix)

        # create new path to use for the reprocessed file by substituting new path parts for the old path parts in abspath
        if modify_path:
            use_regex = settings.get_setting("use_regex_path_map")
            if use_regex:
                # Split on the first colon: left is regex pattern, right is replacement
                colon_idx = path_map.index(':')
                pattern = path_map[:colon_idx]
                replacement = path_map[colon_idx + 1:]
                try:
                    new_path = re.sub(pattern, replacement, str(abspath))
                    abspath = Path(new_path)
                    logger.debug(f"Regex path map applied: pattern='{pattern}', replacement='{replacement}', result='{abspath}'")
                except re.error as e:
                    logger.error(f"Invalid regex pattern in path_map: {e}")
            else:
                abspath_parts_list = list(abspath.parts)
                old_path_parts = list(Path(path_map.split(':')[0]).parts)
                new_path_parts = list(Path(path_map.split(':')[1]).parts)
                new_parts_list = new_path_parts + abspath_parts_list[len(old_path_parts):]
                abspath = Path(*new_parts_list)

        # if not reprocessing all files and if a script path is provided run the script_path and get the results
        # if not reprocessing all files and no script path is provided, set the script_results to 0 (success) so 
        # the reprocess still happens (this handles the case of the user simply wanting to reprocess on success or failure
        # without further testing.
        # input=json.dumps(data) passes the entire data object to the script for use by the script.

        if not reprocess_all and script_path and os.path.isfile(script_path) and abspath:
            result = subprocess.run(
                [script_path],input=json.dumps(data),
                capture_output=True, text=True, timeout=30
            )
            script_results = result.returncode
            logger.debug(f"script_results: {script_results}")
        else:
            script_results = 0

        logger.debug(f"abspath: {abspath}")

        if target_library_id and abspath and script_results == 0:

            # Add source file to the target library's queue using a background thread.
            # This is necessary because the current task still exists in the database
            # when this plugin runs (it's deleted after the plugin returns). The background
            # thread waits for the current task to be deleted before creating the new one.
            thread = threading.Thread(
                target=delayed_create_task,
                args=(str(abspath), int(target_library_id)),
                daemon=True
            )
            thread.start()
            logger.info(f"Started background task to add '{abspath}' to library {target_library_id}")

        else:

            logger.error(f"Unable to create task for file '{abspath}' in library {target_library_id}")

    return data