Video automations: Auto-Backup Database (video DB) — custom twin (phase 5)

Unlike the cleanup twins, backup can't share the music handler — it's a different
DB file. Extract the music backup body into _backup_db_at(db_path, ...) (music
behaviour byte-identical, now a thin wrapper over DATABASE_PATH) and add
auto_backup_video_database pointing at VIDEO_DATABASE_PATH (video_library.db).
New video_backup_database action (scope='video' block + registry), owned_by='video'
system automation on the music cadence (every 3 days).

Tests: a REAL backup behaviour test — music backup lands next to music_library.db,
video backup next to video_library.db, no cross-contamination (this is the whole
reason it can't be shared); scope isolation; single video-owned seed; own handler.
Existing music maintenance tests (22) still green — refactor is non-regressing.
EXPECTED_ACTION_NAMES updated.

core/automation/blocks.py

@@ -276,6 +276,9 @@ ACTIONS: list[dict] = [
      "description": "Clear completed downloads and empty directories", "available": True},
     {"type": "video_full_cleanup", "label": "Full Cleanup", "icon": "trash", "scope": "video",
      "description": "Clear quarantine, download queue, import folder, and search history in one sweep", "available": True},
+    # Custom (NOT a shared handler): backs up video_library.db, not the music DB.
+    {"type": "video_backup_database", "label": "Backup Database", "icon": "save", "scope": "video",
+     "description": "Create a timestamped backup of the video library database", "available": True},
 ]
 
 

core/automation/handlers/maintenance.py

@@ -105,11 +105,10 @@ def auto_update_discovery_pool(config: Dict[str, Any], deps: AutomationDeps) ->
 _MAX_BACKUPS = 5
 
 
-def auto_backup_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
-    """Create a hot SQLite backup, then prune old backups so only the
-    newest ``_MAX_BACKUPS`` remain."""
-    automation_id = config.get('_automation_id')
-    db_path = os.environ.get('DATABASE_PATH', 'database/music_library.db')
+def _backup_db_at(db_path: str, deps: AutomationDeps, automation_id) -> Dict[str, Any]:
+    """Create a hot SQLite backup of ``db_path``, then prune old backups so only
+    the newest ``_MAX_BACKUPS`` remain. Shared by the music and video backup
+    actions — only the DB file differs (they can't share one backup)."""
     if not os.path.exists(db_path):
         return {'status': 'error', 'reason': 'Database file not found'}
 
@@ -146,6 +145,20 @@ def auto_backup_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[s
     return {'status': 'completed', 'backup_path': backup_path, 'size_mb': str(size_mb)}
 
 
+def auto_backup_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Hot SQLite backup of the MUSIC database (``DATABASE_PATH``)."""
+    db_path = os.environ.get('DATABASE_PATH', 'database/music_library.db')
+    return _backup_db_at(db_path, deps, config.get('_automation_id'))
+
+
+def auto_backup_video_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
+    """Hot SQLite backup of the VIDEO database (``VIDEO_DATABASE_PATH`` /
+    video_library.db). Same logic as the music backup but a different DB file —
+    the two can't share one backup, so this is a video-specific action."""
+    db_path = os.environ.get('VIDEO_DATABASE_PATH', 'database/video_library.db')
+    return _backup_db_at(db_path, deps, config.get('_automation_id'))
+
+
 # ─── refresh_beatport_cache ──────────────────────────────────────────
 
 

core/automation/handlers/registration.py

@@ -26,6 +26,7 @@ from core.automation.handlers.maintenance import (
     auto_cleanup_wishlist,
     auto_update_discovery_pool,
     auto_backup_database,
+    auto_backup_video_database,
     auto_refresh_beatport_cache,
 )
 from core.automation.handlers.download_cleanup import (
@@ -168,6 +169,11 @@ def register_all(deps: AutomationDeps) -> None:
         'video_full_cleanup',
         lambda config: auto_full_cleanup(config, deps),
     )
+    # Video DB backup — its OWN handler (video_library.db, not the music DB).
+    engine.register_action_handler(
+        'video_backup_database',
+        lambda config: auto_backup_video_database(config, deps),
+    )
     engine.register_action_handler(
         'clean_completed_downloads',
         lambda config: auto_clean_completed_downloads(config, deps),

core/automation_engine.py

@@ -209,6 +209,14 @@ SYSTEM_AUTOMATIONS = [
         'initial_delay': 900,
         'owned_by': 'video',
     },
+    {
+        'name': 'Auto-Backup Database',
+        'trigger_type': 'schedule',
+        'trigger_config': {'interval': 3, 'unit': 'days'},
+        'action_type': 'video_backup_database',
+        'initial_delay': 600,
+        'owned_by': 'video',
+    },
     # Video twin of music's 'Auto-Deep Scan Library', split into TWO because Movies
     # and TV are independent libraries — a TV scan never pulls in new movies and
     # vice-versa. Fixed weekly deep scan (re-read + prune removed) at 02:00 server-

tests/automation/test_handler_registration.py

@@ -63,6 +63,7 @@ EXPECTED_ACTION_NAMES = frozenset({
     'video_clean_search_history',
     'video_clean_completed_downloads',
     'video_full_cleanup',
+    'video_backup_database',
 })
 
 # Action names that MUST register a guard (duplicate-run prevention).

tests/automation/test_video_maintenance_twins.py

@@ -113,3 +113,69 @@ def test_video_full_cleanup_reuses_the_music_handler():
     handlers = _registered_handlers()
     assert "video_full_cleanup" in handlers
     assert "full_cleanup" in handlers
+
+
+# ── Phase 5: Auto-Backup Database (video — CUSTOM, not a shared handler) ─────
+
+def test_video_backup_database_is_video_scoped_only():
+    assert "video_backup_database" in _action_types("video")
+    assert "video_backup_database" not in _action_types("music")
+    assert "backup_database" in _action_types("music")
+    assert "backup_database" not in _action_types("video")
+
+
+def test_video_backup_database_seeds_one_video_owned_system_row():
+    rows = _system_by_action("video_backup_database")
+    assert len(rows) == 1 and rows[0]["owned_by"] == "video"
+    assert rows[0]["trigger_config"] == {"interval": 3, "unit": "days"}
+
+
+def _mk_sqlite(path):
+    import sqlite3
+    con = sqlite3.connect(str(path))
+    con.execute("CREATE TABLE t (x)")
+    con.commit()
+    con.close()
+
+
+class _Deps:
+    class _L:
+        def error(self, *a, **k):
+            pass
+        def debug(self, *a, **k):
+            pass
+    logger = _L()
+    def update_progress(self, *a, **k):
+        pass
+
+
+def test_video_backup_targets_video_db_and_music_backup_targets_music_db(tmp_path, monkeypatch):
+    # The whole reason this one can't be shared: each backs up ITS OWN db file.
+    import glob
+    from core.automation.handlers.maintenance import (
+        auto_backup_database, auto_backup_video_database)
+    music_db = tmp_path / "music_library.db"
+    video_db = tmp_path / "video_library.db"
+    _mk_sqlite(music_db)
+    _mk_sqlite(video_db)
+    monkeypatch.setenv("DATABASE_PATH", str(music_db))
+    monkeypatch.setenv("VIDEO_DATABASE_PATH", str(video_db))
+    deps = _Deps()
+
+    r_music = auto_backup_database({"_automation_id": "m"}, deps)
+    r_video = auto_backup_video_database({"_automation_id": "v"}, deps)
+
+    assert r_music["status"] == "completed" and r_video["status"] == "completed"
+    # each backup sits next to its OWN db — no cross-contamination
+    assert r_music["backup_path"].startswith(str(music_db))
+    assert r_video["backup_path"].startswith(str(video_db))
+    assert glob.glob(str(music_db) + ".backup_*")
+    assert glob.glob(str(video_db) + ".backup_*")
+    # the video backup did NOT touch the music db's backups and vice-versa
+    assert not glob.glob(str(music_db) + ".backup_*" )[0].startswith(str(video_db))
+
+
+def test_video_backup_database_has_its_own_handler():
+    handlers = _registered_handlers()
+    assert "video_backup_database" in handlers
+    assert "backup_database" in handlers