diff --git a/README.md b/README.md index 5c77f08..18b0967 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,8 @@ After loading the script, configure these fields: - `Enabled`: turns automation on or off - `Watch List`: one game per line using `exe_name|archive_subfolder` - `Archive Root`: destination root for archived recordings -- `Auto Delete Original Recording`: deletes the source recording after a verified archive copy succeeds +- `Disable Auto Archive`: leaves recordings in OBS's normal recording location without copying them to `Archive Root`, and turns off automatic deletion of the source recording +- `Auto Delete Original Recording`: deletes the source recording after a verified archive copy succeeds; this is disabled when auto archive is turned off - `Poll Interval Ms`: how often the script scans running processes - `Exit Grace Period Sec`: delay before stop after the last game exits - `Copy Timeout Sec`: max wait for OBS to finish writing, release, verify, and delete the recording file @@ -171,6 +172,7 @@ cs2.exe|Counter-Strike 2 ``` - `Archive Root`: `D:\GameArchive` +- `Disable Auto Archive`: `false` - `Auto Delete Original Recording`: `true` - `Poll Interval Ms`: `1000` - `Exit Grace Period Sec`: `10` diff --git a/obs_scripts/auto_record_games.py b/obs_scripts/auto_record_games.py index 50eb76c..2c66419 100644 --- a/obs_scripts/auto_record_games.py +++ b/obs_scripts/auto_record_games.py @@ -58,6 +58,7 @@ def script_defaults(settings) -> None: obs.obs_data_set_default_bool(settings, "enabled", True) obs.obs_data_set_default_string(settings, "watch_list", "") obs.obs_data_set_default_string(settings, "archive_root", "") + obs.obs_data_set_default_bool(settings, "disable_archive", False) obs.obs_data_set_default_bool(settings, "auto_delete_recordings", True) obs.obs_data_set_default_int(settings, "poll_interval_ms", 1000) obs.obs_data_set_default_int(settings, "exit_grace_period_sec", 10) @@ -77,6 +78,8 @@ def script_properties(): "", None, ) + disable_archive = obs.obs_properties_add_bool(properties, "disable_archive", "Disable Auto Archive") + obs.obs_property_set_modified_callback(disable_archive, _on_disable_archive_modified) obs.obs_properties_add_bool(properties, "auto_delete_recordings", "Auto Delete Original Recording") obs.obs_properties_add_int(properties, "poll_interval_ms", "Poll Interval Ms", 250, 60_000, 250) obs.obs_properties_add_int(properties, "exit_grace_period_sec", "Exit Grace Period Sec", 0, 600, 1) @@ -120,14 +123,16 @@ def _load_settings(obs_settings) -> ScriptSettings: _warn(message) archive_root = obs.obs_data_get_string(obs_settings, "archive_root").strip() - if not archive_root: + disable_archive = obs.obs_data_get_bool(obs_settings, "disable_archive") + if not archive_root and not disable_archive: _warn("Archive Root is empty; completed recordings will not be copied until it is configured.") return ScriptSettings( enabled=obs.obs_data_get_bool(obs_settings, "enabled"), watch_entries=watch_entries, archive_root=archive_root, - auto_delete_recordings=obs.obs_data_get_bool(obs_settings, "auto_delete_recordings"), + disable_archive=disable_archive, + auto_delete_recordings=obs.obs_data_get_bool(obs_settings, "auto_delete_recordings") and not disable_archive, poll_interval_ms=max(250, obs.obs_data_get_int(obs_settings, "poll_interval_ms")), exit_grace_period_sec=max(0, obs.obs_data_get_int(obs_settings, "exit_grace_period_sec")), copy_timeout_sec=max(1, obs.obs_data_get_int(obs_settings, "copy_timeout_sec")), @@ -135,6 +140,16 @@ def _load_settings(obs_settings) -> ScriptSettings: ) +def _on_disable_archive_modified(props, _property, settings) -> bool: + disable_archive = obs.obs_data_get_bool(settings, "disable_archive") + auto_delete = obs.obs_properties_get(props, "auto_delete_recordings") + if auto_delete is not None: + obs.obs_property_set_enabled(auto_delete, not disable_archive) + if disable_archive: + obs.obs_data_set_bool(settings, "auto_delete_recordings", False) + return True + + def _register_timer(interval_ms: int) -> None: if STATE is None: return @@ -215,7 +230,7 @@ def _on_frontend_event(event: int) -> None: if event == obs.OBS_FRONTEND_EVENT_RECORDING_STOPPED: copy_request = STATE.engine.on_recording_stopped(now) _log("OBS recording stopped.") - if copy_request is not None: + if copy_request is not None and not STATE.settings.disable_archive: _submit_copy(copy_request) diff --git a/src/obs_auto_record/settings.py b/src/obs_auto_record/settings.py index 39baafc..1b81d6e 100644 --- a/src/obs_auto_record/settings.py +++ b/src/obs_auto_record/settings.py @@ -28,6 +28,7 @@ class ScriptSettings: enabled: bool = True watch_entries: tuple[WatchEntry, ...] = () archive_root: str = "" + disable_archive: bool = False auto_delete_recordings: bool = True poll_interval_ms: int = 1000 exit_grace_period_sec: int = 10 diff --git a/tests/test_auto_record_games.py b/tests/test_auto_record_games.py new file mode 100644 index 0000000..8a44c07 --- /dev/null +++ b/tests/test_auto_record_games.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path +import sys +import types +import unittest +import ctypes +from unittest.mock import Mock, patch + +ROOT = Path(__file__).resolve().parent.parent +SCRIPT_PATH = ROOT / "obs_scripts" / "auto_record_games.py" + + +def _load_script_module(): + logs: list[tuple[int, str]] = [] + obs_module = types.ModuleType("obspython") + obs_module.OBS_FRONTEND_EVENT_RECORDING_STARTED = 1 + obs_module.OBS_FRONTEND_EVENT_RECORDING_STOPPED = 2 + obs_module.OBS_TEXT_MULTILINE = 0 + obs_module.OBS_PATH_DIRECTORY = 0 + obs_module.LOG_INFO = 10 + obs_module.LOG_WARNING = 20 + + def obs_data_get_string(settings, key: str) -> str: + return str(settings.get(key, "")) + + def obs_data_get_bool(settings, key: str) -> bool: + return bool(settings.get(key, False)) + + def obs_data_get_int(settings, key: str) -> int: + return int(settings.get(key, 0)) + + def obs_data_set_bool(settings, key: str, value: bool) -> None: + settings[key] = value + + def script_log(level: int, message: str) -> None: + logs.append((level, message)) + + def obs_properties_create(): + return {} + + def _add_property(props, kind: str, key: str, label: str): + prop = {"type": kind, "key": key, "label": label, "enabled": True, "modified_callback": None} + props[key] = prop + return prop + + def obs_properties_add_bool(props, key: str, label: str): + return _add_property(props, "bool", key, label) + + def obs_properties_add_text(props, key: str, label: str, _kind: int): + return _add_property(props, "text", key, label) + + def obs_properties_add_path(props, key: str, label: str, _kind: int, _filter: str, _default): + return _add_property(props, "path", key, label) + + def obs_properties_add_int(props, key: str, label: str, _low: int, _high: int, _step: int): + return _add_property(props, "int", key, label) + + def obs_property_set_modified_callback(prop, callback) -> None: + prop["modified_callback"] = callback + + def obs_properties_get(props, key: str): + return props.get(key) + + def obs_property_set_enabled(prop, enabled: bool) -> None: + prop["enabled"] = enabled + + obs_module.obs_data_get_string = obs_data_get_string + obs_module.obs_data_get_bool = obs_data_get_bool + obs_module.obs_data_get_int = obs_data_get_int + obs_module.obs_data_set_bool = obs_data_set_bool + obs_module.obs_properties_create = obs_properties_create + obs_module.obs_properties_add_bool = obs_properties_add_bool + obs_module.obs_properties_add_text = obs_properties_add_text + obs_module.obs_properties_add_path = obs_properties_add_path + obs_module.obs_properties_add_int = obs_properties_add_int + obs_module.obs_property_set_modified_callback = obs_property_set_modified_callback + obs_module.obs_properties_get = obs_properties_get + obs_module.obs_property_set_enabled = obs_property_set_enabled + obs_module.script_log = script_log + obs_module.logs = logs + + class _FakeWinFunc: + def __init__(self, return_value: int) -> None: + self.return_value = return_value + self.argtypes = None + self.restype = None + + def __call__(self, *args, **kwargs) -> int: + return self.return_value + + class _FakeKernel32: + def __init__(self) -> None: + self.CreateToolhelp32Snapshot = _FakeWinFunc(0) + self.Process32FirstW = _FakeWinFunc(0) + self.Process32NextW = _FakeWinFunc(0) + self.CloseHandle = _FakeWinFunc(1) + + module_name = "test_auto_record_games_module" + spec = importlib.util.spec_from_file_location(module_name, SCRIPT_PATH) + if spec is None or spec.loader is None: + raise RuntimeError("Failed to load auto_record_games.py") + + module = importlib.util.module_from_spec(spec) + with patch.dict(sys.modules, {"obspython": obs_module}, clear=False), patch.object( + ctypes, + "WinDLL", + return_value=_FakeKernel32(), + create=True, + ): + sys.modules.pop(module_name, None) + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module, obs_module + + +class AutoRecordGamesTests(unittest.TestCase): + def test_load_settings_reads_disable_archive_without_warning_for_empty_archive_root(self) -> None: + module, obs_module = _load_script_module() + + settings = module._load_settings( + { + "enabled": True, + "watch_list": "eldenring.exe|Elden Ring", + "archive_root": " ", + "disable_archive": True, + "auto_delete_recordings": True, + "poll_interval_ms": 1000, + "exit_grace_period_sec": 10, + "copy_timeout_sec": 120, + "verbose_logging": False, + } + ) + + self.assertTrue(settings.disable_archive) + self.assertFalse(settings.auto_delete_recordings) + self.assertEqual(obs_module.logs, []) + + def test_recording_stop_skips_archive_submission_when_disabled(self) -> None: + module, _obs_module = _load_script_module() + copy_request = module.CopyRequest(trigger_exe_name="eldenring.exe", archive_subfolder="Elden Ring") + engine = Mock() + engine.on_recording_stopped.return_value = copy_request + module.STATE = types.SimpleNamespace( + settings=module.ScriptSettings(disable_archive=True), + engine=engine, + ) + + with patch.object(module, "_submit_copy") as submit_copy: + module._on_frontend_event(module.obs.OBS_FRONTEND_EVENT_RECORDING_STOPPED) + + submit_copy.assert_not_called() + + def test_disable_archive_callback_turns_off_and_disables_auto_delete(self) -> None: + module, _obs_module = _load_script_module() + properties = module.script_properties() + settings = { + "disable_archive": True, + "auto_delete_recordings": True, + } + + disable_archive = properties["disable_archive"] + callback = disable_archive["modified_callback"] + self.assertIsNotNone(callback) + + callback(properties, disable_archive, settings) + + self.assertFalse(settings["auto_delete_recordings"]) + self.assertFalse(properties["auto_delete_recordings"]["enabled"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_settings.py b/tests/test_settings.py index 32ebf12..37c785f 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -9,7 +9,7 @@ if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) -from obs_auto_record.settings import parse_watch_list, sanitize_archive_subfolder +from obs_auto_record.settings import ScriptSettings, parse_watch_list, sanitize_archive_subfolder class SettingsTests(unittest.TestCase): @@ -36,6 +36,9 @@ def test_archive_subfolder_is_sanitized(self) -> None: self.assertEqual(sanitize_archive_subfolder(' Elden:Ring? '), "Elden_Ring_") self.assertEqual(sanitize_archive_subfolder("con"), "_con") + def test_script_settings_can_disable_archive(self) -> None: + self.assertTrue(ScriptSettings(disable_archive=True).disable_archive) + if __name__ == "__main__": unittest.main()