Untitled

import obspython as obs
from shutil import move
import os
import time
import logging

# Output paths and search strings
output_paths = {
    "anythingelse": "D:/vods/anything-else",
    "bridges": "D:/vods/bridges",
    "destinystudio": "D:/vods/destiny-stream",
    "saturdaymorning": "D:/vods/saturday-morning"
}

# Set up logging to a file
log_file = "obs_recording_move.log"
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
file_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(file_handler)

# Description of the script for OBS
def script_description():
    """Return the plugin description."""
    return "This plugin moves the recorded file to a specified folder after recording stops, based on keywords in the filename."

# Hook to connect recording stop signal
def script_load(settings):
    """Hook stop recording signal on plugin load."""
    signal_handler = obs.obs_output_get_signal_handler(
        obs.obs_frontend_get_recording_output()
    )
    obs.signal_handler_connect(signal_handler, "stop", signal_handler_function)

# Update function to set output paths and search strings dynamically
def script_update(settings):
    """Called when script settings are updated."""
    global output_paths
    output_paths = {
        "anythingelse": obs.obs_data_get_string(settings, "search_string1"),
        "bridges": obs.obs_data_get_string(settings, "search_string2"),
        "destinystudio": obs.obs_data_get_string(settings, "search_string3"),
        "saturdaymorning": obs.obs_data_get_string(settings, "search_string4")
    }

# Property panel for OBS interface configuration
def script_properties():
    """Define properties for OBS interface configuration."""
    props = obs.obs_properties_create()

    obs.obs_properties_add_text(props, "search_string1", "Keyword for Output 1", obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_path(props, "output_path1", "Output Folder 1", obs.OBS_PATH_DIRECTORY, "", "")

    obs.obs_properties_add_text(props, "search_string2", "Keyword for Output 2", obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_path(props, "output_path2", "Output Folder 2", obs.OBS_PATH_DIRECTORY, "", "")

    obs.obs_properties_add_text(props, "search_string3", "Keyword for Output 3", obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_path(props, "output_path3", "Output Folder 3", obs.OBS_PATH_DIRECTORY, "", "")

    obs.obs_properties_add_text(props, "search_string4", "Keyword for Output 4", obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_path(props, "output_path4", "Output Folder 4", obs.OBS_PATH_DIRECTORY, "", "")

    return props

# Signal handler for recording stop
def signal_handler_function(calldata):
    """Handle the recording stop signal and move the file accordingly."""
    try:
        last_recording = obs.obs_frontend_get_last_recording()

        # Wait for the file to be fully available
        time.sleep(5)

        if not last_recording or not os.path.exists(last_recording):
            logger.error(f"Recording file not found: {last_recording}")
            return

        # Move the file based on matching keyword
        for keyword, output_path in output_paths.items():
            if keyword.lower() in last_recording.lower():
                if os.path.exists(output_path):
                    move(last_recording, output_path)
                    logger.info(f"Moved file to: {output_path}")
                else:
                    logger.error(f"Output path does not exist: {output_path}")
                break
        else:
            logger.warning(f"No keyword match found for recording: {last_recording}")

    except Exception as e:
        logger.error(f"Error occurred during file move: {e}")
import obspython as obs
from shutil import move
import os
import time
import logging
from typing import Dict, Optional
import re

class VODMover:
    def __init__(self):
        # Output paths and search strings with default values
        self.output_paths: Dict[str, str] = {
            "anythingelse": "D:/vods/anything-else",
            "bridges": "D:/vods/bridges",
            "destinystudio": "D:/vods/destiny-stream",
            "saturdaymorning": "D:/vods/saturday-morning"
        }

        # Initialize logging
        self.setup_logging()

    def setup_logging(self) -> None:
        """Set up logging configuration"""
        log_file = "obs_recording_move.log"
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(levelname)s - %(message)s",
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()  # Also log to console
            ]
        )
        self.logger = logging.getLogger(__name__)

    def move_recording(self, recording_path: str) -> None:
        """
        Move recording to appropriate folder based on filename patterns
        """
        if not recording_path or not os.path.exists(recording_path):
            self.logger.error(f"Invalid recording path: {recording_path}")
            return

        try:
            filename = os.path.basename(recording_path).lower()
            destination = self._get_destination_folder(filename)

            if destination:
                self._perform_move(recording_path, destination)
            else:
                self.logger.warning(f"No matching folder found for: {filename}")
                # Move to anything-else folder as fallback
                self._perform_move(recording_path, self.output_paths["anythingelse"])

        except Exception as e:
            self.logger.error(f"Error processing recording {recording_path}: {str(e)}")

    def _get_destination_folder(self, filename: str) -> Optional[str]:
        """
        Determine destination folder based on filename patterns
        """
        patterns = {
            "bridges": r"bridge|bridges",
            "destinystudio": r"destiny|studio",
            "saturdaymorning": r"saturday|morning|weekend"
        }

        for key, pattern in patterns.items():
            if re.search(pattern, filename, re.IGNORECASE):
                return self.output_paths[key]

        return None

    def _perform_move(self, source: str, destination: str) -> None:
        """
        Perform the actual file move operation
        """
        if not os.path.exists(destination):
            os.makedirs(destination, exist_ok=True)
            self.logger.info(f"Created destination directory: {destination}")

        try:
            # Wait for file to be fully written
            time.sleep(5)

            # Get destination filepath
            dest_path = os.path.join(destination, os.path.basename(source))

            # Handle existing files
            if os.path.exists(dest_path):
                base, ext = os.path.splitext(dest_path)
                counter = 1
                while os.path.exists(f"{base}_{counter}{ext}"):
                    counter += 1
                dest_path = f"{base}_{counter}{ext}"

            move(source, dest_path)
            self.logger.info(f"Successfully moved {source} to {dest_path}")

        except Exception as e:
            self.logger.error(f"Failed to move file {source}: {str(e)}")

# Global instance of VODMover
vod_mover = VODMover()

def script_description():
    """Return the script description for OBS"""
    return "Automatically moves VOD recordings to categorized folders based on filename patterns"

def script_load(settings):
    """Hook stop recording signal on script load"""
    signal_handler = obs.obs_output_get_signal_handler(
        obs.obs_frontend_get_recording_output()
    )
    obs.signal_handler_connect(signal_handler, "stop", on_recording_stop)

def script_update(settings):
    """Update paths when settings are changed"""
    global vod_mover

    # Update paths from OBS settings
    vod_mover.output_paths = {
        "anythingelse": obs.obs_data_get_string(settings, "anything_else_path"),
        "bridges": obs.obs_data_get_string(settings, "bridges_path"),
        "destinystudio": obs.obs_data_get_string(settings, "destiny_path"),
        "saturdaymorning": obs.obs_data_get_string(settings, "saturday_path")
    }

def script_properties():
    """Define properties for OBS interface"""
    props = obs.obs_properties_create()

    # Add path properties for each category
    obs.obs_properties_add_path(
        props, "anything_else_path",
        "Anything Else Path",
        obs.OBS_PATH_DIRECTORY,
        "", "D:/vods/anything-else"
    )
    obs.obs_properties_add_path(
        props, "bridges_path",
        "Bridges Path",
        obs.OBS_PATH_DIRECTORY,
        "", "D:/vods/bridges"
    )
    obs.obs_properties_add_path(
        props, "destiny_path",
        "Destiny Studio Path",
        obs.OBS_PATH_DIRECTORY,
        "", "D:/vods/destiny-stream"
    )
    obs.obs_properties_add_path(
        props, "saturday_path",
        "Saturday Morning Path",
        obs.OBS_PATH_DIRECTORY,
        "", "D:/vods/saturday-morning"
    )

    return props

def on_recording_stop(calldata):
    """Handle recording stop event"""
    try:
        recording_path = obs.obs_frontend_get_last_recording()
        vod_mover.move_recording(recording_path)
    except Exception as e:
        logging.error(f"Error in recording stop handler: {str(e)}")
import obspython as obs
from shutil import move
import os
import time
import logging

# Change the log file location to a user-writable directory
log_file = os.path.join(os.path.expanduser("~"), "obs_recording_move.log")

# Output paths and search strings (set dynamically)
output_paths = {}

# Set up logging to a file
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Ensure logging to a file as well
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
logger.addHandler(file_handler)

# Description of the script for OBS
def script_description():
    """Return the plugin description."""
    return "This plugin moves the recorded file to a specified folder after recording stops, based on keywords in the filename."

# Hook to connect recording stop signal
def script_load(settings):
    """Hook the recording stop signal when the plugin loads."""
    signal_handler = obs.obs_output_get_signal_handler(
        obs.obs_frontend_get_recording_output()
    )
    obs.signal_handler_connect(signal_handler, "stop", signal_handler_function)

# Update function to set output paths and search strings dynamically
def script_update(settings):
    """Called when script settings are updated."""
    global output_paths
    output_paths.clear()

    # Retrieve settings from the OBS properties panel
    for i in range(1, 5):
        keyword = obs.obs_data_get_string(settings, f"search_string{i}")
        output_path = obs.obs_data_get_string(settings, f"output_path{i}")

        if keyword and output_path:
            output_paths[keyword] = output_path

# Property panel for OBS interface configuration
def script_properties():
    """Define properties for OBS interface configuration."""
    props = obs.obs_properties_create()

    # Add fields for up to 4 keyword-folder pairs
    for i in range(1, 5):
        obs.obs_properties_add_text(
            props, f"search_string{i}", f"Keyword for Output {i}", obs.OBS_TEXT_DEFAULT
        )
        obs.obs_properties_add_path(
            props, f"output_path{i}", f"Output Folder {i}", obs.OBS_PATH_DIRECTORY, "", ""
        )

    return props

# Signal handler for recording stop
def signal_handler_function(calldata):
    """Handle the recording stop signal and move the file accordingly."""
    try:
        last_recording = obs.obs_frontend_get_last_recording()

        # Wait a bit to ensure the file is available
        time.sleep(5)

        if not last_recording or not os.path.exists(last_recording):
            logger.error(f"Recording file not found: {last_recording}")
            return

        # Move the file based on keyword matching
        for keyword, output_path in output_paths.items():
            if keyword.lower() in os.path.basename(last_recording).lower():
                # Ensure the output path exists
                os.makedirs(output_path, exist_ok=True)

                destination = os.path.join(output_path, os.path.basename(last_recording))
                move(last_recording, destination)

                logger.info(f"Moved file to: {destination}")
                return  # Exit once the file is moved successfully

        # If no matching keyword was found
        logger.warning(f"No keyword match found for recording: {last_recording}")

    except Exception as e:
        logger.error(f"Error occurred during file move: {e}")

#!/usr/bin/env python3
# Original: <https://pastebin.com/xsUhLTtH>
"""
Moves a recording file when filename matches match_keyword X to output_folder Y

Created by `Sateviss` :chatting:
Refactored and cleaned by `JohnSmith`
"""

import logging
from pathlib import Path
from time import sleep
from typing import NamedTuple

import obspython as obs

# Configuration constants
NUM_OF_MATCHES = 4
"""Number of match strings to be configurable in OBS"""

NUM_OF_POLL_TRIES = 15
"""Num of polls to try between OBS signal and moving the file."""
POLL_WAIT_TIME = 1
"""Seconds to wait between each file poll."""

LOG_FORMAT = "%(asctime)s | %(name)s | %(levelname)s | %(message)s"
"""Logging format for the script."""
LOG_LEVEL = logging.INFO
"""Logging level for stdout."""
LOG_FILEPATH = Path.home() / "obs_recording_migration_script.log"
"""Logging file path for the script."""


# Data classes
class RecordingMatchCase(NamedTuple):
    """Entry type to encode a match case for recording migration."""

    match_keyword: str
    """Keyword to match against the recording filename."""

    output_folder: Path
    """Output folder to move the recording file to."""


# Initialisation functions
def init_logger() -> logging.Logger:
    """Initialises a logger object for the script."""

    print("Initialising logger object...")

    # Setup logging to stdout
    logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT)
    logger = logging.getLogger(__name__)

    # Setup logging to logfile
    file_handler = logging.FileHandler(LOG_FILEPATH)
    file_handler.setFormatter(logging.Formatter(LOG_FORMAT))

    # Join logging handlers together
    logger.addHandler(file_handler)

    logger.info(f"Logger initialized! {LOG_FILEPATH=}")
    return logger


# Global mutable variables
script_logger = init_logger()
"""Logger object for the script."""
recording_migrate_mappings: set[RecordingMatchCase] = set()
"""Mapping of match strings to output paths."""


# ================================
# OBS invocation hooks
# ================================
def script_description() -> str:
    """Return the plugin description string."""
    # Just re-using the script docstring; see top of file
    return __doc__ if __doc__ is not None else ""


def script_load(settings) -> None:
    """Initialises hook for stop recording signal during plugin load."""

    script_logger.debug("Attaching handler...")

    try:
        obs_frontend = obs.obs_frontend_get_recording_output()
        signal_handler = obs.obs_output_get_signal_handler(obs_frontend)

        obs.signal_handler_connect(signal_handler, "stop", signal_handler_function)

    except Exception as err:
        script_logger.error(f"Error occurred during handler attachment: {err}")
        raise err

    script_logger.info("Script handler hook was successfully attached!")


def script_properties():
    """Hook to setup initial OBS properties for the script."""

    script_logger.info("Setting up OBS properties...")

    # global NUM_OF_MATCHES
    props = obs.obs_properties_create()

    for i in range(NUM_OF_MATCHES):

        obs.obs_properties_add_text(
            props, f"match{i}", f"Match {i + 1}", obs.OBS_TEXT_DEFAULT
        )
        obs.obs_properties_add_path(
            props, f"path{i}", f"Path {i + 1}", obs.OBS_PATH_DIRECTORY, None, None
        )

    return props


def script_update(settings) -> None:
    """Hook to update recording patterns"""

    # global NUM_OF_MATCHES
    global recording_migrate_mappings

    for i in range(NUM_OF_MATCHES):

        new_match_entry = RecordingMatchCase(
            match_keyword=obs.obs_data_get_string(settings, f"match{i}"),
            output_folder=Path(obs.obs_data_get_string(settings, f"path{i}")),
        )

        recording_migrate_mappings.add(new_match_entry)
        script_logger.debug(f"Loaded matcher {i}:", new_match_entry)


def signal_handler_function(calldata) -> None:
    """Handles the recording stop signal and moves the file accordingly."""
    # global recording_migrate_mappings

    script_logger.info("Handler triggered! Running recording migration hook...")

    try:

        # Get the last recording file path
        orig_recording_raw: str = obs.obs_frontend_get_last_recording()
        if not orig_recording_raw:
            script_logger.error(
                f"No recording file path was returned by OBS: {orig_recording_raw=}",
            )
            return
        orig_recording: Path = Path(orig_recording_raw)
        script_logger.debug(f"Recording file path: {orig_recording=}")

        # Wait for the file to be fully available
        for poll_i in range(NUM_OF_POLL_TRIES):
            sleep(POLL_WAIT_TIME)
            if orig_recording.exists():
                break
            script_logger.debug(f"Recording file not found yet: {poll_i=}")
        else:
            script_logger.error(f"Recording file cannot be found: {orig_recording=}")
            # TODO: turn these returns into error raises?
            return

        script_logger.info(f"Valid initial recording file given! {orig_recording=}")
        script_logger.debug("Trying to match file...")

        # Find matching file migration mapping
        for mapping in recording_migrate_mappings:

            # Checking if the current match is valid...
            if mapping.match_keyword.casefold() not in orig_recording.name.casefold():
                script_logger.debug(f"Not a match! {mapping=}")
                continue

            script_logger.info(f"Suitable match found: matcher={mapping}")

            # Checking if the matching output folder is currently valid
            # NOTE: if this gets race conditioned somehow, just throw everything below into a retry loop
            if not mapping.output_folder.exists():
                script_logger.error(
                    f"Output path does not exist: {mapping.output_folder=}"
                )
                break

            if not mapping.output_folder.is_dir():
                script_logger.error(
                    f"Output Path is not actually a directory: {mapping.output_folder=}"
                )
                break

            # Constructing the new recording path
            new_recording_path = mapping.output_folder / orig_recording.name
            script_logger.debug(
                f"Valid resulting recording path constructed: {new_recording_path}"
            )

            script_logger.debug("Attempting file migration...")
            orig_recording.rename(new_recording_path)
            script_logger.info(
                f"Recording file has been migrated to: {new_recording_path}"
            )
            break

        else:
            script_logger.error(
                f"No keyword match found for recording: {orig_recording}"
            )

    except Exception as err:
        script_logger.error(f"Error occurred during file migration hook: {err}")
        # TODO: should error messages be chained up to OBS itself?
        # raise err