Build Fix Learn Guide
Complete ARM Setup Guide
This is the complete Build Fix Learn procedure for recreating the Automated Ripping Machine and Jellyfin server. It covers the hardware baseline, Ubuntu, RAID1 storage, permissions, Jellyfin, GPU transcoding, MakeMKV, HandBrakeCLI, movie mapping, TV episode mapping, disc fingerprints, the ARM pipeline, automatic Jellyfin library movement, testing, troubleshooting, recovery, and the full sanitized production source.
Reading this guide
You can read the entire build procedure directly on this page, including the current sanitized production scripts, configuration example, systemd service, database initialization code, and arm-status dashboard. The downloadable .md file contains the same material for offline use.
Offline copy
Download the complete guide
The Markdown download contains the same complete procedure and source shown below. The source ZIP is also available if you want the sanitized production files already separated into their installable directory structure.
Guide Contents
This online guide is the complete V12 build guide, not a shortened article companion. It includes the full installation procedure and the sanitized production source needed to recreate the ARM software.
Build and installation: hardware baseline, Ubuntu, RAID1, permissions, Jellyfin, NVIDIA/NVENC, MakeMKV, HandBrakeCLI, ARM account, directories, configuration, SQLite database, and systemd.
Automation logic: disc inspection, movie classification and main-feature mapping, TV classification and episode mapping, independent parallel ripping, the single transcode queue, automatic Jellyfin library movement, generic-disc fingerprinting, cleanup, restart recovery, correction tools, and the terminal dashboard.
Full source appendix: arm.conf.example, arm.service, arm-status, arm_config.py, arm_logger.py, init_db.py, arm_orchestrator.py, disc_inspector.py, title_selector.py, ripper.py, transcoder.py, library_mover.py, cleanup_rips.py, recovery_check.py, and job_corrector.py.
The public source is sanitized. Personal information, usernames, private network addresses, API credentials, machine-specific identifiers, production databases, old backup revisions, and compiled cache files are not published.
ARM / Jellyfin Server Build Procedure
Known-Good Two-Drive Parallel Pipeline
Version: 1.2 (Build Fix Learn V12)
Original known-good baseline: 2026-08-16
V12 source audit: 2026-09-07
Purpose: Recreate the current Automated Ripping Machine (ARM) and Jellyfin server from a fresh Ubuntu installation. This is a living document and should be updated whenever hardware, storage, Blu-ray support, pipeline logic, transcoding, or recovery procedures change.
1. Final Architecture
The server is designed to:
- Detect inserted DVD discs automatically.
- Inspect multiple optical drives independently.
- Select the appropriate movie or TV titles.
- Rip independently from each optical drive.
- Allow multiple optical drives to rip simultaneously.
- Queue completed rips for transcoding.
- Run only one HandBrake transcode worker at a time.
- Move completed media into the Jellyfin library.
- Clean temporary working files only when safe.
- Recover orphaned RIPPING or TRANSCODING jobs after interrupted ARM restarts.
Current optical drives:
/dev/sr0— hp DVD A DH16AAL/dev/sr1— ATAPI iHAS124 C
The former slow external /dev/sr2 drive was removed from production.
Future plan: add a Blu-ray optical drive and extend the pipeline as needed.
2. Hardware Baseline
The build began with a reused gaming desktop. The Intel Core i7-4790K, 16 GB of RAM, case/motherboard, and DVD drives were already available. The main hardware added for the ARM/Jellyfin conversion was the 1 TB SSD, two 2 TB hard drives, NVIDIA Quadro P400, and a Wi-Fi card. The finished server is intended to run headless during normal use, so it does not need a dedicated monitor, keyboard, or mouse.
- CPU: Intel Core i7-4790K, reused
- RAM: 16 GB, reused
- NVIDIA Quadro P400: purchased for NVENC transcoding
- OS/rip SSD: 1 TB SSD, purchased
- Media RAID: 2 × 2 TB hard drives in RAID1, purchased
- Wi-Fi card: network connection for headless operation where Ethernet is inconvenient
- Optical drives: two internal SATA DVD drives, reused
Current disk layout:
/dev/sda1 ┐
├── /dev/md0 RAID1 ext4 label=JELLYFIN -> /mnt/jellyfin
/dev/sdb1 ┘
/dev/sdc1 FAT32 EFI -> /boot/efi
/dev/sdc2 ext4 root -> /
/rips is a directory on the root SSD.Always verify device identities before destructive work:
lsblk -o NAME,SIZE,FSTYPE,LABEL,MODEL,TRAN,MOUNTPOINTS3. Install Ubuntu
Install Ubuntu Server 26.04 in UEFI mode on the SSD.
Recommended layout:
- EFI: about 1 GB FAT32
- Root: remaining SSD space, ext4
After installation:
sudo apt update
sudo apt full-upgradeVerify:
lsblk -f
df -hT / /boot/efi4. Build the Jellyfin RAID1
WARNING: Verify the correct media disks first. These operations can destroy data.
lsblk -o NAME,SIZE,MODEL,SERIALCreate RAID1 using the two media partitions:
sudo mdadm --create /dev/md0 \
--level=1 \
--raid-devices=2 \
/dev/sda1 /dev/sdb1Format:
sudo mkfs.ext4 -L JELLYFIN /dev/md0Create mount point:
sudo mkdir -p /mnt/jellyfinFind UUID:
sudo blkid /dev/md0Add the filesystem UUID to /etc/fstab:
UUID=<JELLYFIN-FILESYSTEM-UUID> /mnt/jellyfin ext4 defaults 0 2Test:
sudo mount -a
findmnt /mnt/jellyfin
cat /proc/mdstat
sudo mdadm --detail /dev/md05. Create Media Directories and Permissions
sudo mkdir -p \
/mnt/jellyfin/Movies \
"/mnt/jellyfin/TV Shows"ARM needs write access; Jellyfin needs read access.
Example ACLs:
sudo setfacl -m u:arm:rwx /mnt/jellyfin
sudo setfacl -m u:jellyfin:rx /mnt/jellyfin
sudo setfacl -d -m u:arm:rwx /mnt/jellyfin
sudo setfacl -d -m u:jellyfin:rx /mnt/jellyfinTest ARM writes:
sudo -u arm touch /mnt/jellyfin/.arm-write-test
sudo rm /mnt/jellyfin/.arm-write-test6. Install Jellyfin
Install Jellyfin from the official Jellyfin repository.
Then:
sudo systemctl enable jellyfin
sudo systemctl start jellyfin
sudo systemctl status jellyfin --no-pagerVerify user/group access:
id jellyfinConfigure Jellyfin libraries:
Movies: /mnt/jellyfin/Movies
TV Shows: /mnt/jellyfin/TV Shows7. Configure NVIDIA Transcoding
Install the appropriate NVIDIA driver for the Quadro P400.
Verify:
nvidia-smi
ls -l /dev/nvidia*
ls -l /dev/dri/The known-good HandBrake configuration uses NVENC H.264 with quality 15, automatic frame rate, no crop, preserved display aspect, all audio, all subtitles, optimized MP4 output.
8. Install MakeMKV and HandBrakeCLI
Required commands:
makemkvcon
HandBrakeCLIVerify:
command -v makemkvcon
command -v HandBrakeCLITest optical drives:
makemkvcon -r info disc:0
makemkvcon -r info disc:19. Create the ARM Service Account
Create a dedicated arm account with a nologin shell.
It must have access to optical drives through the cdrom group.
Verify:
id armIf needed:
sudo usermod -aG cdrom arm10. ARM Directory Layout
Create:
sudo mkdir -p \
/opt/arm/backups \
/opt/arm/config \
/opt/arm/docs \
/opt/arm/logs \
/opt/arm/pipelines \
/opt/arm/scripts \
/opt/arm/stateImportant paths:
/opt/arm/config/arm.conf
/opt/arm/state/arm.db
/opt/arm/scripts/arm_orchestrator.pyCore scripts:
arm_config.py
arm_orchestrator.py
cleanup_rips.py
disc_inspector.py
library_mover.py
ripper.py
title_selector.py
transcoder.py11. Create the Rip Workspace
/rips is on the OS SSD.
sudo mkdir -p \
/rips/incoming \
/rips/ripping \
/rips/transcoding/output \
/rips/encoding \
/rips/completed \
/rips/failed \
/rips/logsSet ownership:
sudo chown -R arm:arm /ripsTest:
sudo -u arm touch /rips/.arm-test
sudo rm /rips/.arm-test12. ARM Configuration
Current production optical section:
[optical]
drive_0 = /dev/sr0
drive_1 = /dev/sr1The removed external drive must not be present.
Important production setting:
auto_library_move = trueJellyfin media root:
[jellyfin]
media_root = /mnt/jellyfinAutomation should enable ripping, transcoding, library movement, and safe cleanup.
13. ARM Database
SQLite database:
/opt/arm/state/arm.dbImportant tables:
jobs
job_files
job_titles
events
system_stateIntegrity and recent jobs:
sudo python3 - <<'PY'
import sqlite3
conn = sqlite3.connect("/opt/arm/state/arm.db")
print(conn.execute("PRAGMA integrity_check").fetchone())
for row in conn.execute("SELECT id, drive, disc_label, status FROM jobs ORDER BY id DESC LIMIT 10"):
print(row)
PYExpected integrity result:
('ok',)14. Disc Inspection Behavior
Each configured optical drive may have one active disc_inspector.py process.
While an inspector owns a drive, the orchestrator must not run another MakeMKV detection probe against that same drive. This prevents competing MakeMKV information scans.
DVD inspection currently uses a 600-second timeout.
15. Movie and TV Classification
disc_inspector.py classifies discs as movie or TV content.
TV recognition uses season-style labels such as:
SEASON
S01
S1Known successful TV example:
GAME_OF_THRONES_S1_DISC1Movie discs use feature-title scoring.
16. Title Selection
Jobs progress:
INSPECTED -> TITLE_SELECTEDMovie flow normally selects one primary feature. TV discs may select multiple episode titles.
Selections are stored in job_titles.selected.
16A. Movie Mapping and Main-Feature Selection
Movie and TV title selection are separate paths in the production selector. A movie job selects exactly one main feature.
For every inspected title, title_selector.py calculates a feature score from:
- runtime
- chapter count
- file size
The current duration bonuses are strongest at 90 minutes or longer, then 60 minutes or longer, then 30 minutes or longer. Runtime also contributes directly, chapter count adds weight, and file size adds a smaller capped weight. Candidates are sorted by score, then duration and size. The highest-scoring title is marked selected = 1, every other title for that movie job is cleared, and the job advances to TITLE_SELECTED.
Conceptually:
MOVIE DISC
-> inspect titles
-> score runtime + chapters + size
-> select one main feature
-> MakeMKV rip
-> HandBrake NVENC transcode
-> library mover
-> /mnt/jellyfin/Movies/<MOVIE>/<MOVIE>.mp416B. TV Mapping and Episode Selection
TV jobs may select multiple titles. The production selector first filters episode candidates by duration, file size, and chapter structure, then applies grouping logic so it does not blindly accept every title on the disc. Selected episodes are kept in disc/title order and assigned season and episode metadata that follows the file through ripping, transcoding, correction, and library movement.
TV classification can use season markers and configured overrides. Ambiguous publisher volume labels can be held for metadata confirmation instead of treating a volume number as a season number.
Conceptually:
TV DISC
-> classify / confirm show and season
-> inspect titles
-> choose coherent episode candidates
-> assign SxxExx mapping
-> rip each selected title
-> transcode each episode
-> /mnt/jellyfin/TV Shows/<SHOW>/Season XX/17. Independent Parallel Ripping
The orchestrator tracks active ripper workers per drive.
Conceptually:
ACTIVE_RIPPERS = {
"/dev/sr0": process,
"/dev/sr1": process
}Only one ripper worker is allowed per drive.
The orchestrator does not wait for all rippers to finish before continuing its main loop.
Valid concurrent behavior:
SR0: RIPPING
SR1: INSPECTING / TITLE_SELECTED / RIPPINGAt peak, both may run:
makemkvcon -r mkv disc:0 ...
makemkvcon -r mkv disc:1 ...18. Single Transcode Queue
Completed jobs become:
RIPPEDThe orchestrator permits exactly one active transcoder.py worker at a time. The oldest eligible ripped job enters the transcode worker.
Correct overlap:
Job A: TRANSCODING
Job B: RIPPINGVerify:
sudo ps -eo pid,ppid,etime,args | \
grep -E '[m]akemkvcon|[H]andBrakeCLI|[t]ranscoder.py'19. Library Movement
Movies:
/mnt/jellyfin/Movies/<MOVIE_NAME>/<MOVIE_NAME>.mp4TV:
/mnt/jellyfin/TV Shows/<SHOW>/Season 01/<episode>.mp4Jobs are marked LIBRARY_MOVED only after destination verification.
Existing destinations are protected. ARM should fail with LIBRARY_DESTINATION_EXISTS rather than overwrite existing media.
19A. Generic Disc Fingerprints
Some DVDs identify themselves with generic labels such as DVD_VIDEO. The production disc_inspector.py therefore builds a stable SHA-256 fingerprint from normalized MakeMKV title metadata instead of relying only on the physical disc label.
For each title, the fingerprint includes:
- MakeMKV title ID, falling back to ARM title number
- duration in seconds
- size rounded to MiB
- chapter count
It intentionally excludes drive path, drive model, timestamp, and physical disc label so the same disc should produce the same fingerprint in either optical drive. For a generic label, ARM can derive a temporary human-facing title using the first 12 characters of the fingerprint and can supersede a repeat inspection when the same fingerprint has already reached LIBRARY_MOVED.
20. Cleanup
cleanup_rips.py deletes working data only when safe.
During active processing, BLOCKED is expected.
A clean system should eventually show:
Eligible jobs: 0
Safe to remove: 0.00 GB
Blocked working data: 0.00 GBFailed jobs may remain in database history after useless partial files are removed.
21. Restart Recovery
ARM includes recovery for orphaned jobs.
On startup, orphaned states may be returned to retryable states:
RIPPING -> TITLE_SELECTED
TRANSCODING -> RIPPEDThis protects against interrupted service restarts or reboots.
22. systemd Service
ARM service:
arm.serviceUseful commands:
sudo systemctl enable arm
sudo systemctl start arm
sudo systemctl status arm --no-pager
sudo journalctl -u arm -f23. Known-Good Functional Test
Test A — One DVD
Expected event sequence:
DISC_INSERTED
DISC_INSPECTED
TITLE_SELECTED
RIP_STARTED
RIP_COMPLETED
TRANSCODE_COMPLETED
LIBRARY_MOVE_COMPLETEDTest B — Two DVDs
Insert one disc in each optical drive and verify both can rip simultaneously.
Test C — Pipeline overlap
Allow one rip to finish while the other continues.
Verify simultaneous:
Job A = TRANSCODING
Job B = RIPPINGThis is the defining known-good scheduler test.
24. Dashboard
The ARM dashboard displays:
- ARM service status
- current inspection
- ripping
- transcoding
- drive states
- recent jobs
- library information
- active MakeMKV / HandBrake processes
- rip output growth
- recent ARM activity
- load, RAM, and
/ripsusage
Historical job state may occasionally appear for an idle drive; database/events are authoritative during troubleshooting.
25. Useful Troubleshooting Commands
Recent jobs:
sudo python3 - <<'PY'
import sqlite3
conn = sqlite3.connect("/opt/arm/state/arm.db")
for row in conn.execute("SELECT id, drive, disc_label, status, error_code, error_message FROM jobs ORDER BY id DESC LIMIT 20"):
print(row)
PYWorkers:
sudo ps -eo pid,ppid,etime,args | \
grep -E '[d]isc_inspector|[m]akemkvcon|[H]andBrakeCLI|[t]ranscoder.py'ARM journal:
sudo journalctl -u arm -n 100 --no-pager
sudo journalctl -u arm -fRAID:
cat /proc/mdstat
sudo mdadm --detail /dev/md0Space:
df -hT / /rips /mnt/jellyfin25A. Install the V12 Production Source
The public V12 package contains the current production scripts rather than reconstructed snippets. Copy the files only after reviewing the paths and configuration for your own machine.
Create the application directories if they do not already exist:
sudo mkdir -p /opt/arm/scripts /opt/arm/config /opt/arm/logs /opt/arm/state /opt/arm/backupsCopy the Python files into /opt/arm/scripts/, install the sanitized configuration template as /opt/arm/config/arm.conf, and replace REPLACE_WITH_YOUR_JELLYFIN_API_KEY if you want the correction utility to request a Jellyfin library refresh.
Install the systemd unit as:
/etc/systemd/system/arm.serviceInstall the terminal dashboard as:
/usr/local/bin/arm-statusThen set executable permissions and initialize a fresh database:
sudo chmod 755 /opt/arm/scripts/*.py
sudo chmod 755 /usr/local/bin/arm-status
sudo python3 /opt/arm/scripts/init_db.py
sudo systemctl daemon-reload
sudo systemctl enable arm
sudo systemctl start armDo not copy another person’s production arm.db. init_db.py creates the schema for a fresh installation.
26. ARM Application Snapshot
Known-good ARM application snapshot:
/opt/arm/backups/known-good-2026-08-16-parallel-pipelineContents:
arm.db
arm.service
config/
scripts/27. Whole-System Disaster Recovery
The dedicated external recovery HDD is ext4 and labeled:
ARM_RECOVERYIt contains:
ARM-Recovery/
├── README-RESTORE.txt
├── arm/
├── metadata/
├── scripts/
└── system/The recovery drive contains the known-good root filesystem snapshot, EFI contents, ARM application snapshot, partition layout, UUID information, RAID configuration, package inventory, enabled services, EFI boot information, complete SSD replacement instructions, and a reusable backup script.
Normally keep this drive unplugged and stored safely.
28. Future System Backups
Mount the recovery drive at:
/mnt/arm-recoveryRun:
sudo /mnt/arm-recovery/ARM-Recovery/scripts/backup-system.shThe script stops ARM and Jellyfin, verifies workers are gone, captures metadata, snapshots the root filesystem and EFI partition, excludes /rips and /mnt/jellyfin, verifies critical files, then restores prior service state.
29. Future Work
Planned additions:
- Blu-ray drive installation
- Blu-ray MakeMKV workflow
- Blu-ray title-selection rules
- Blu-ray transcoding policy
- snapshot retention policy
- dashboard live-state improvements
- per-job recovery-process matching
- storage expansion
- future Jellyfin hardware-acceleration changes
30. Known-Good Baseline Summary
As of 2026-08-16:
Ubuntu Server
|
+-- Jellyfin
| |
| +-- /mnt/jellyfin RAID1
|
+-- ARM Orchestrator
|
+-- /dev/sr0 -> independent inspection/rip worker
|
+-- /dev/sr1 -> independent inspection/rip worker
|
+-- RIPPED queue
|
+-- one HandBrake NVENC worker
|
+-- library mover
|
+-- cleanupValidated:
- two-drive automatic disc detection
- simultaneous disc inspection
- simultaneous DVD ripping
- ripping while another job transcodes
- one transcode worker
- automatic Jellyfin library movement
- duplicate-destination protection
- automated cleanup
- interrupted-job recovery
- ARM-level known-good backup
- whole-system external-HDD disaster-recovery snapshot
Update this procedure whenever production architecture changes.
Appendix: Full Sanitized V12 Production Source
The following source is included so this Markdown file is usable even without the separate ZIP package. The production API key and personal or host-specific information are not included.
config/arm.conf.example
[system]
name = ARM
version = 0.1
[user]
run_as = arm
[optical]
drive_0 = /dev/sr0
drive_1 = /dev/sr1
[ripping]
tool = makemkvcon
rip_root = /rips
rip_work = /rips/ripping
[transcoding]
tool = HandBrakeCLI
[jellyfin]
media_root = /mnt/jellyfin
url = http://127.0.0.1:8096
api_key = REPLACE_WITH_YOUR_JELLYFIN_API_KEY
[state]
root = /opt/arm/state
logs = /opt/arm/logs
backups = /opt/arm/backups
[automation]
auto_rip = true
auto_transcode = true
auto_library_move = true
auto_cleanup_rips = true
wait_for_next_disc = true
poll_seconds = 15
auto_eject = true
[tv_overrides]
# Optional format:
# DISC_LABEL_PREFIX = Jellyfin Show Name|SeasonNumber
# EXAMPLE_SHOW_S1 = Example Show|1systemd/arm.service
[Unit]
Description=ARM Automated Ripping Machine
After=local-fs.target
Wants=local-fs.target
[Service]
Type=simple
User=arm
Group=arm
ExecStart=/usr/bin/python3 /opt/arm/scripts/arm_orchestrator.py
Restart=always
RestartSec=10
WorkingDirectory=/opt/arm
StandardOutput=journal
StandardError=journal
NoNewPrivileges=true
[Install]
WantedBy=multi-user.targetdashboard/arm-status
#!/usr/bin/env python3
import configparser
import os
import select
import sqlite3
import subprocess
from pathlib import Path
import sys
import termios
import time
import tty
from datetime import datetime, timezone
DB = "/opt/arm/state/arm.db"
CONFIG = "/opt/arm/config/arm.conf"
RIP_ROOT = "/rips/ripping"
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
RED = "\033[31m"
CYAN = "\033[36m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------
def run(cmd, timeout=3):
try:
return subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=timeout,
check=False,
).stdout.strip()
except Exception:
return ""
def clear():
print("\033[2J\033[H", end="")
def status_text(status):
if status in ("RIPPING", "TRANSCODING", "TITLE_SELECTED"):
return f"{CYAN}{status}{RESET}"
if status in (
"RIP_FAILED",
"TRANSCODE_FAILED",
"SELECTION_FAILED",
):
return f"{RED}{status}{RESET}"
if status == "LIBRARY_MOVED":
return f"{GREEN}{status}{RESET}"
return f"{YELLOW}{status}{RESET}"
def parse_time(value):
if not value:
return None
try:
return datetime.fromisoformat(value)
except Exception:
return None
def elapsed(value):
dt = parse_time(value)
if not dt:
return ""
seconds = max(
0,
int(
(
datetime.now(timezone.utc) - dt
).total_seconds()
),
)
hours, remainder = divmod(seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02}:{minutes:02}:{seconds:02}"
def bytes_text(value):
value = float(value)
for unit in ("B", "KB", "MB", "GB", "TB"):
if value < 1024:
return f"{value:.1f} {unit}"
value /= 1024
return f"{value:.1f} PB"
def directory_size(path):
total = 0
if not os.path.exists(path):
return 0
for root, dirs, files in os.walk(path):
for filename in files:
try:
total += os.path.getsize(
os.path.join(root, filename)
)
except OSError:
pass
return total
# ---------------------------------------------------------------------------
# Library helpers
# ---------------------------------------------------------------------------
def library_status(conn):
"""
Return compact TV and movie library information for the dashboard.
TV is grouped by show/season and only displays episodes that have
actually reached the Jellyfin library.
Movies are intentionally limited to the most recent relevant movie
job so the normal dashboard stays within one terminal screen.
"""
tv = {}
rows = conn.execute("""
SELECT
jt.name,
jt.season_number,
jt.episode_number,
jf.status AS file_status,
jf.library_file,
j.destination
FROM job_titles jt
JOIN jobs j
ON j.id = jt.job_id
LEFT JOIN job_files jf
ON jf.job_title_id = jt.id
WHERE j.media_type = 'TV'
AND jt.season_number IS NOT NULL
AND jt.episode_number IS NOT NULL
ORDER BY
jt.season_number,
jt.episode_number
""").fetchall()
for row in rows:
show = row["name"]
# Existing ARM TV title rows may not contain the show name.
# The library path is authoritative once the file has reached
# Jellyfin, for example:
# /mnt/jellyfin/TV Shows/Game of Thrones/Season 01/...
if not show and row["library_file"]:
parts = Path(row["library_file"]).parts
try:
shows_index = parts.index("TV Shows")
if shows_index + 1 < len(parts):
show = parts[shows_index + 1]
except ValueError:
pass
if not show and row["destination"]:
parts = Path(row["destination"]).parts
try:
shows_index = parts.index("TV Shows")
if shows_index + 1 < len(parts):
show = parts[shows_index + 1]
except ValueError:
pass
if not show:
continue
season = int(row["season_number"])
episode = int(row["episode_number"])
key = (show, season)
if key not in tv:
tv[key] = {}
status = row["file_status"]
if status == "LIBRARY_MOVED":
tv[key][episode] = "✓"
elif status in (
"TRANSCODING",
"RIPPING",
"MOVING",
):
tv[key][episode] = "●"
elif status in (
"RIPPED",
"TRANSCODED",
):
# Completed the current stage and waiting for the next
# pipeline stage.
tv[key][episode] = "○"
elif status:
tv[key][episode] = "✗"
else:
# The title exists and is expected, but no job_file has
# reached an active/failed/library state yet.
tv[key][episode] = "○"
# Prefer the newest movie that is actively moving through the
# pipeline. This keeps the compact dashboard focused on live work.
movie = conn.execute("""
SELECT
id,
title,
disc_label,
status,
destination,
updated_at
FROM jobs
WHERE COALESCE(media_type, '') != 'TV'
AND status IN (
'RIPPING',
'TRANSCODING',
'MOVING',
'TITLE_SELECTED'
)
ORDER BY id DESC
LIMIT 1
""").fetchone()
# If nothing is actively processing, show the newest movie waiting
# between completed pipeline stages.
if movie is None:
movie = conn.execute("""
SELECT
id,
title,
disc_label,
status,
destination,
updated_at
FROM jobs
WHERE COALESCE(media_type, '') != 'TV'
AND status IN (
'RIPPED',
'TRANSCODED'
)
ORDER BY id DESC
LIMIT 1
""").fetchone()
# Finally, fall back to the most recently completed library move.
if movie is None:
movie = conn.execute("""
SELECT
id,
title,
disc_label,
status,
destination,
updated_at
FROM jobs
WHERE COALESCE(media_type, '') != 'TV'
AND status = 'LIBRARY_MOVED'
ORDER BY id DESC
LIMIT 1
""").fetchone()
return tv, movie
# ---------------------------------------------------------------------------
# Process monitoring
# ---------------------------------------------------------------------------
def process_lines():
output = run([
"ps",
"-eo",
"pid,etime,%cpu,%mem,cmd",
"--sort=-%cpu",
])
wanted = (
"makemkvcon",
"HandBrakeCLI",
"ripper.py",
"transcoder.py",
)
results = []
seen = set()
for line in output.splitlines():
lower = line.lower()
if not any(
name.lower() in lower
for name in wanted
):
continue
if "grep" in lower:
continue
parts = line.split(None, 1)
if not parts:
continue
pid = parts[0]
if pid in seen:
continue
seen.add(pid)
results.append(line)
return results
def process_summary():
processes = process_lines()
makemkv = []
handbrake = []
for line in processes:
lower = line.lower()
if "makemkvcon" in lower:
makemkv.append(line)
elif "handbrakecli" in lower:
handbrake.append(line)
return makemkv, handbrake, processes
# ---------------------------------------------------------------------------
# Drive helpers
# ---------------------------------------------------------------------------
def configured_drives():
config = configparser.ConfigParser()
config.read(CONFIG)
drives = []
if "optical" in config:
for key, value in config["optical"].items():
if key.startswith("drive_"):
drives.append(value)
return drives
def drive_jobs(jobs):
result = {
device: []
for device in configured_drives()
}
for row in jobs:
if row["drive"] in result:
result[row["drive"]].append(row)
return result
def drive_media_state(device):
"""
Determine whether the optical drive currently has media.
We use lsblk rather than ARM's database because the database
describes jobs, while this checks the physical drive.
"""
output = run([
"lsblk",
"-n",
"-o",
"TYPE",
device,
])
if output:
return "MEDIA"
return "EMPTY"
def eject_drive(device):
try:
result = subprocess.run(
["eject", device],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
except Exception as exc:
return False, str(exc)
if result.returncode == 0:
return True, f"Ejected {device}"
message = (
result.stderr.strip()
or result.stdout.strip()
or f"eject returned {result.returncode}"
)
return False, message
# ---------------------------------------------------------------------------
# Keyboard handling
# ---------------------------------------------------------------------------
def read_key(timeout=2):
"""
Read exactly one key without requiring ENTER.
Returns:
None - no key pressed
str - single key
"""
ready, _, _ = select.select(
[sys.stdin],
[],
[],
timeout,
)
if not ready:
return None
return sys.stdin.read(1)
def eject_menu():
"""
Interactive single-key eject menu.
e -> enter this menu
0 -> eject sr0
1 -> eject sr1
q/ESC -> cancel
"""
clear()
print(
f"{CYAN}{BOLD}"
"╔══════════════════════════════════════════════════════════════════════════════╗"
f"{RESET}"
)
print(
f"{CYAN}{BOLD}"
"║ EJECT DRIVE ║"
f"{RESET}"
)
print(
f"{CYAN}"
"╚══════════════════════════════════════════════════════════════════════════════╝"
f"{RESET}"
)
print()
print("Press:")
print()
print(f" {BOLD}0{RESET} Eject /dev/sr0")
print(f" {BOLD}1{RESET} Eject /dev/sr1")
print(f" {BOLD}ESC{RESET} Cancel")
print()
key = read_key(timeout=10)
if key is None:
return
if key == "\x1b":
return
if key.lower() == "q":
return
if key not in ("0", "1"):
return
device = f"/dev/sr{key}"
clear()
print()
print(
f"{YELLOW}{BOLD}"
f"Eject {device}? "
f"{RESET}"
)
print()
print("Press Y to confirm, or any other key to cancel.")
confirm = read_key(timeout=10)
if confirm is None:
return
if confirm.lower() != "y":
return
clear()
print()
print(f"Ejecting {device}...")
print()
ok, message = eject_drive(device)
if ok:
print(f"{GREEN}✓ {message}{RESET}")
else:
print(f"{RED}✗ {message}{RESET}")
print()
print("Returning to dashboard...")
time.sleep(1.5)
# ---------------------------------------------------------------------------
# Job correction / deletion
# ---------------------------------------------------------------------------
JOB_CORRECTOR = "/opt/arm/scripts/job_corrector.py"
def job_action_menu(action, old_settings):
"""
Ask for a Job ID and launch the standalone correction tool.
arm-status normally uses cbreak keyboard input, while the
corrector uses normal line input. Restore the terminal while
the corrector is active, then return to cbreak mode.
"""
termios.tcsetattr(
sys.stdin,
termios.TCSADRAIN,
old_settings,
)
try:
clear()
if action == "correct":
print(f"{BOLD}CORRECT JOB{RESET}")
else:
print(f"{BOLD}DELETE JOB{RESET}")
print()
print(
"Enter the ARM Job ID. "
"Use arm-status --all to find older jobs."
)
print()
value = input(
"Job ID (blank to cancel): "
).strip()
if not value:
return
try:
job_id = int(value)
except ValueError:
print()
print(f"{RED}Invalid Job ID.{RESET}")
input("Press ENTER to return...")
return
if action == "delete":
command = [
"sudo",
JOB_CORRECTOR,
"--delete",
str(job_id),
]
else:
command = [
"sudo",
JOB_CORRECTOR,
str(job_id),
]
clear()
subprocess.run(
command,
check=False,
)
print()
input(
"Press ENTER to return to ARM status..."
)
finally:
tty.setcbreak(sys.stdin.fileno())
def auto_eject_enabled():
"""
Return the persistent [automation] auto_eject setting.
"""
config = configparser.ConfigParser()
config.read(CONFIG)
if not config.has_section("automation"):
return False
try:
return config["automation"].getboolean(
"auto_eject",
fallback=False,
)
except ValueError:
return False
def toggle_auto_eject():
"""
Toggle [automation] auto_eject and persist it to arm.conf.
Returns the new boolean state.
"""
config = configparser.ConfigParser()
config.optionxform = str
config.read(CONFIG)
if not config.has_section("automation"):
config.add_section("automation")
try:
current = config["automation"].getboolean(
"auto_eject",
fallback=False,
)
except ValueError:
current = False
new_value = not current
config["automation"]["auto_eject"] = (
"true" if new_value else "false"
)
with open(CONFIG, "w") as handle:
config.write(handle)
return new_value
# ---------------------------------------------------------------------------
# Compact dashboard helpers
# ---------------------------------------------------------------------------
COMPACT_WIDTH = 100
def compact_rule():
print("-" * COMPACT_WIDTH)
def progress_color(percent):
"""
Progress color intentionally does NOT use red.
Red is reserved exclusively for failures/problems.
"""
if percent is None:
return DIM
if percent >= 100:
return GREEN
if percent >= 75:
return MAGENTA
if percent >= 50:
return BLUE
if percent >= 25:
return CYAN
return DIM
def rip_progress(conn, job_id):
"""
Estimate whole-job ripping progress from:
bytes currently present in /rips/ripping/job-N
divided by
total expected size of selected MakeMKV titles.
"""
row = conn.execute(
"""
SELECT COALESCE(SUM(size_bytes), 0)
FROM job_titles
WHERE job_id = ?
AND selected = 1
""",
(job_id,),
).fetchone()
expected = int(row[0] or 0)
rip_dir = Path(RIP_ROOT) / f"job-{job_id}"
actual = directory_size(str(rip_dir))
if expected <= 0:
return None, actual, expected
percent = int((actual / expected) * 100)
# While actively ripping, avoid claiming completion before
# the ripper has changed database state.
percent = max(0, min(percent, 100))
return percent, actual, expected
def progress_text(percent, growing=False):
if percent is None:
return f"{DIM} --%{RESET}"
color = progress_color(percent)
arrow = " ↑" if growing else ""
return (
f"{color}"
f"{percent:3d}%"
f"{arrow:<2}"
f"{RESET}"
)
def display_name(value):
"""
Pretty dashboard-only display name.
This does not alter the database or library naming.
"""
if not value:
return "UNKNOWN"
name = value.strip()
# Remove common leading/trailing junk from physical disc labels.
name = name.strip("_ ")
# Remove trailing #1234-style mastering/catalog suffixes.
import re
name = re.sub(
r"[_\s]+#\d+\s*$",
"",
name,
)
name = name.replace("_", " ")
name = re.sub(
r"\s+",
" ",
name,
).strip()
# Leave explicit job titles alone if already human-readable;
# otherwise title-case raw disc labels.
if name.isupper():
name = name.title()
return name
def compact_job_status(status):
if status == "LIBRARY_MOVED":
return f"{GREEN}✓ MOVED{RESET}"
if status in (
"RIP_FAILED",
"TRANSCODE_FAILED",
"SELECTION_FAILED",
"LIBRARY_MOVE_FAILED",
):
return f"{RED}✗ {status}{RESET}"
if status in (
"RIPPING",
"TRANSCODING",
"TITLE_SELECTED",
"INSPECTED",
):
return f"{CYAN}● {status}{RESET}"
if status in ("RIPPED", "TRANSCODED"):
return f"{YELLOW}○ {status}{RESET}"
if status == "SUPERSEDED":
return f"{DIM}○ SUPERSEDED{RESET}"
return f"{YELLOW}○ {status}{RESET}"
def gpu_status():
"""
Return a short GPU status string.
NVIDIA failure/absence is non-fatal to the dashboard.
"""
output = run(
[
"nvidia-smi",
"--query-gpu="
"utilization.gpu,"
"utilization.encoder,"
"temperature.gpu,"
"memory.used,"
"memory.total",
"--format=csv,noheader,nounits",
],
timeout=2,
)
if not output:
return "NVENC -- GPU -- TEMP --"
first = output.splitlines()[0]
parts = [
part.strip()
for part in first.split(",")
]
if len(parts) < 5:
return "NVENC -- GPU -- TEMP --"
gpu, enc, temp, mem_used, mem_total = parts[:5]
return (
f"NVENC {enc}% "
f"GPU {gpu}% "
f"{temp}°C "
f"VRAM {mem_used}/{mem_total} MiB"
)
def transcode_queue_depth(conn):
"""
Count jobs that are actually eligible to enter the single
transcode worker.
This deliberately mirrors the scheduler's job-level RIPPED state.
A TV episode that finishes ripping while the rest of its disc is
still ripping is not yet considered queued.
"""
row = conn.execute(
"""
SELECT COUNT(*)
FROM jobs
WHERE status = 'RIPPED'
"""
).fetchone()
return int(row[0] or 0)
def transcode_progress(job_id):
"""
Read the newest HandBrake transcode log for this job and return
the latest reported encoding percentage.
Works for movie and TV log naming because we search by job ID.
"""
import glob
import re
patterns = [
f"/opt/arm/logs/job-{job_id}*transcode*.log",
f"/opt/arm/logs/job-{job_id}*.log",
]
candidates = []
for pattern in patterns:
candidates.extend(glob.glob(pattern))
candidates = list(set(candidates))
if not candidates:
return None
candidates.sort(
key=lambda filename: Path(filename).stat().st_mtime,
reverse=True,
)
log = Path(candidates[0])
try:
# HandBrake progress uses carriage returns in some builds,
# so reading the complete text is more reliable than tailing
# by newline.
content = log.read_text(
errors="replace",
)
except Exception:
return None
matches = re.findall(
r"Encoding:\s+task\s+\d+\s+of\s+\d+,\s+"
r"([0-9]+(?:\.[0-9]+)?)\s*%",
content,
)
if not matches:
return None
try:
return float(matches[-1])
except ValueError:
return None
def active_transcode_text(conn, previous_sizes):
job = conn.execute(
"""
SELECT
id,
disc_label,
title,
media_type,
transcode_started_at
FROM jobs
WHERE status = 'TRANSCODING'
ORDER BY id
LIMIT 1
"""
).fetchone()
if not job:
return f"{DIM}○ IDLE{RESET}"
label = display_name(
job["title"]
or job["disc_label"]
or "UNKNOWN"
)
episode_text = ""
if job["media_type"] == "TV":
episode = conn.execute(
"""
SELECT season_number, episode_number
FROM job_files
WHERE job_id = ?
AND status = 'TRANSCODING'
ORDER BY episode_number
LIMIT 1
""",
(job["id"],),
).fetchone()
if episode:
episode_text = (
f" S{int(episode['season_number']):02d}"
f"E{int(episode['episode_number']):02d}"
)
percent = transcode_progress(job["id"])
progress = ""
if percent is not None:
key = f"transcode-{job['id']}"
previous = previous_sizes.get(key)
growing = (
previous is not None
and percent > previous
)
previous_sizes[key] = percent
# Keep one decimal internally but display a clean integer.
display_percent = int(percent)
progress = (
progress_text(
display_percent,
growing,
)
+ " "
)
age = elapsed(job["transcode_started_at"])
return (
f"{CYAN}● ACTIVE{RESET} "
f"{progress}"
f"{label[:30]}"
f"{episode_text} "
f"[{age}]"
)
def episode_ranges(numbers):
numbers = sorted(set(numbers))
if not numbers:
return ""
ranges = []
start = previous = numbers[0]
for number in numbers[1:]:
if number == previous + 1:
previous = number
continue
if start == previous:
ranges.append(f"E{start:02d}")
else:
ranges.append(
f"E{start:02d}-E{previous:02d}"
)
start = previous = number
if start == previous:
ranges.append(f"E{start:02d}")
else:
ranges.append(
f"E{start:02d}-E{previous:02d}"
)
return ",".join(ranges)
def compact_tv_job_progress(conn, job_id):
"""
Return episode-level progress for one TV job.
Example:
E01-E05 ✓TX E06 ●TX E07-E08 ○RIP
"""
rows = conn.execute(
"""
SELECT
jt.episode_number,
COALESCE(
(
SELECT jf.status
FROM job_files jf
WHERE jf.job_id = jt.job_id
AND (
jf.job_title_id = jt.id
OR (
jf.job_title_id IS NULL
AND jf.season_number = jt.season_number
AND jf.episode_number = jt.episode_number
)
)
ORDER BY jf.id DESC
LIMIT 1
),
'WAITING'
) AS file_status
FROM job_titles jt
WHERE jt.job_id = ?
AND jt.selected = 1
AND jt.episode_number IS NOT NULL
ORDER BY jt.episode_number
""",
(job_id,),
).fetchall()
if not rows:
return ""
labels = {
"LIBRARY_MOVED": "✓LIB",
"TRANSCODED": "✓TX",
"TRANSCODING": "●TX",
"RIPPED": "○RIP",
"RIPPING": "●RIP",
"PENDING": "WAIT",
"WAITING": "WAIT",
"LIBRARY_MOVE_FAILED": "✗MOVE",
"TRANSCODE_FAILED": "✗TX",
"RIP_FAILED": "✗RIP",
}
groups = []
current_status = None
current_numbers = []
for row in rows:
ep = int(row["episode_number"])
status = row["file_status"] or "WAITING"
if (
current_status is None
or status == current_status
):
current_status = status
current_numbers.append(ep)
continue
groups.append(
(current_status, current_numbers)
)
current_status = status
current_numbers = [ep]
if current_numbers:
groups.append(
(current_status, current_numbers)
)
parts = []
for status, numbers in groups:
ranges = episode_ranges(numbers)
if ranges:
parts.append(
f"{ranges} "
f"{labels.get(status, status)}"
)
return " ".join(parts)
def compact_tv_library(conn):
tv, _ = library_status(conn)
lines = []
for (show, season), episodes in sorted(
tv.items(),
key=lambda item: (
item[0][0].lower(),
item[0][1],
),
):
episode_numbers = sorted(episodes)
if not episode_numbers:
continue
failure = any(
value == "✗"
for value in episodes.values()
)
active = any(
value == "●"
for value in episodes.values()
)
# Detect internal gaps only. We cannot know whether episodes
# before the first or after the last should exist.
expected = set(
range(
episode_numbers[0],
episode_numbers[-1] + 1,
)
)
missing = expected - set(episode_numbers)
if failure:
flag = f"{RED}✗{RESET}"
elif missing:
flag = f"{YELLOW}⚠{RESET}"
elif active:
flag = f"{CYAN}●{RESET}"
else:
flag = " "
summary = episode_ranges(episode_numbers)
lines.append(
(
show,
season,
flag,
summary,
)
)
return lines
def recent_movies(conn, limit=3):
return conn.execute(
"""
SELECT
COALESCE(NULLIF(title, ''), disc_label) AS name
FROM jobs
WHERE COALESCE(media_type, '') != 'TV'
AND status = 'LIBRARY_MOVED'
ORDER BY id DESC
LIMIT ?
""",
(limit,),
).fetchall()
def raid_space_text():
try:
stats = os.statvfs("/mnt/jellyfin")
total = stats.f_blocks * stats.f_frsize
free = stats.f_bavail * stats.f_frsize
used = total - free
percent = (
int((used / total) * 100)
if total
else 0
)
return (
f"RAID {percent}% used "
f"{bytes_text(free)} free"
)
except Exception:
return "RAID unavailable"
def active_inspections():
"""
Return optical drives currently owned by disc_inspector.py.
Using the inspector process itself is more reliable than trying
to infer inspection from generic makemkvcon command lines.
"""
output = run([
"ps",
"-eo",
"args",
])
drives = []
for line in output.splitlines():
if "disc_inspector.py" not in line:
continue
# Do not accidentally count this dashboard/process search.
if "grep" in line:
continue
for device in configured_drives():
if device in line and device not in drives:
drives.append(device)
return drives
def render_compact(
conn,
service_indicator,
now,
previous_sizes,
):
# ------------------------------------------------------------
# Banner
# ------------------------------------------------------------
print(
f"{CYAN}{BOLD}"
"╔══════════════════════════════════════════════════════════════════════════════════════════════════╗"
f"{RESET}"
)
print(
f"{CYAN}{BOLD}"
"║ ARM • AUTOMATED RIPPING MACHINE ║"
f"{RESET}"
)
print(
f"{CYAN}"
"╚══════════════════════════════════════════════════════════════════════════════════════════════════╝"
f"{RESET}"
)
print(
f"{service_indicator} "
f"{now.strftime('%H:%M:%S %Z')}"
)
# ------------------------------------------------------------
# Drives / activity
# ------------------------------------------------------------
print()
print(f"{BOLD}ACTIVITY{RESET}")
compact_rule()
active_statuses = (
"INSPECTED",
"TITLE_SELECTED",
"RIPPING",
"RIPPED",
"TRANSCODING",
"TRANSCODED",
)
drive_parts = []
inspections = active_inspections()
for index, device in enumerate(configured_drives()):
# Inspection owns the physical drive before a job necessarily
# exists in the database.
if device in inspections:
part = (
f"D{index} {CYAN}● INSPECTING{RESET} "
f"{DIM}MakeMKV scan{RESET}"
)
drive_parts.append(part)
continue
# Only these states still require the physical optical disc.
#
# Once the job reaches RIPPED, the data is safely on disk and
# the user may replace the disc while transcoding/library work
# continues independently.
current = conn.execute(
"""
SELECT
id,
disc_label,
title,
status
FROM jobs
WHERE drive = ?
AND status IN (
'INSPECTED',
'TITLE_SELECTED',
'RIPPING'
)
ORDER BY id DESC
LIMIT 1
""",
(device,),
).fetchone()
if current and current["status"] == "RIPPING":
percent, actual, expected = rip_progress(
conn,
current["id"],
)
key = f"job-{current['id']}"
old_size = previous_sizes.get(key)
growing = (
old_size is not None
and actual > old_size
)
previous_sizes[key] = actual
label = display_name(
current["title"]
or current["disc_label"]
or "UNKNOWN"
)
part = (
f"D{index} {CYAN}● RIP{RESET} "
f"{progress_text(percent, growing)} "
f"{label[:26]}"
)
elif current:
label = display_name(
current["title"]
or current["disc_label"]
or "UNKNOWN"
)
part = (
f"D{index} "
f"{compact_job_status(current['status'])} "
f"{label[:24]}"
)
else:
media = drive_media_state(device)
if media == "MEDIA":
part = (
f"D{index} {GREEN}✓ REPLACE{RESET} "
f"{DIM}[E] eject{RESET}"
)
else:
part = (
f"D{index} {DIM}○ EMPTY{RESET}"
)
drive_parts.append(part)
if len(drive_parts) >= 2:
print(
f"{drive_parts[0]:<55}"
f"{drive_parts[1]}"
)
else:
for part in drive_parts:
print(part)
queue_depth = transcode_queue_depth(conn)
print(
"TX "
+ active_transcode_text(
conn,
previous_sizes,
)
+ f" Q {queue_depth} waiting "
+ gpu_status()
)
inspections = active_inspections()
if inspections:
labels = []
for device in inspections:
if device.startswith("/dev/sr"):
labels.append(
"D" + device.rsplit("sr", 1)[1]
)
else:
labels.append(device)
print(
f"INSPECT {CYAN}●{RESET} "
+ ", ".join(labels)
+ " identifying new disc"
)
else:
print(
f"INSPECT {DIM}○ IDLE{RESET}"
)
# ------------------------------------------------------------
# Recent jobs
# ------------------------------------------------------------
print()
print(f"{BOLD}RECENT JOBS{RESET}")
compact_rule()
recent = conn.execute(
"""
SELECT
id,
drive,
disc_label,
title,
media_type,
status
FROM jobs
ORDER BY id DESC
LIMIT 6
"""
).fetchall()
for row in recent:
name = display_name(
row["title"]
or row["disc_label"]
or "UNKNOWN"
)
drive = (
row["drive"]
.replace("/dev/", "")
if row["drive"]
else "-"
)
print(
f"{row['id']:<4} "
f"{drive:<4} "
f"{name[:38]:<39} "
f"{compact_job_status(row['status'])}"
)
if row["media_type"] == "TV":
progress = compact_tv_job_progress(
conn,
row["id"],
)
if progress:
print(
f" {DIM}"
f"{progress}"
f"{RESET}"
)
# ------------------------------------------------------------
# Library
# ------------------------------------------------------------
print()
print(f"{BOLD}LIBRARY{RESET}")
compact_rule()
tv_lines = compact_tv_library(conn)
if tv_lines:
for show, season, flag, summary in tv_lines[:4]:
print(
f"TV {show[:28]:<29} "
f"S{season:02d} {flag} "
f"{summary}"
)
else:
print(f"{DIM}TV No library entries yet.{RESET}")
movies = recent_movies(conn, 3)
movie_names = [
display_name(row["name"])
for row in movies
if row["name"]
]
if movie_names:
print(
"MOV "
+ " | ".join(
name[:25]
for name in movie_names
)
)
# ------------------------------------------------------------
# Storage / controls
# ------------------------------------------------------------
print()
compact_rule()
print(
f"{raid_space_text()} "
f"| {BOLD}[E]{RESET} Eject "
f"{BOLD}[A]{RESET} Auto-eject "
f"{GREEN + 'ON' + RESET if auto_eject_enabled() else DIM + 'OFF' + RESET} "
f"{BOLD}[C]{RESET} Correct "
f"{BOLD}[D]{RESET} Delete "
f"{BOLD}[Q]{RESET} Quit "
f"| {DIM}arm-status --all for diagnostics{RESET}"
)
# ---------------------------------------------------------------------------
# Main dashboard
# ---------------------------------------------------------------------------
def job_episode_summary(conn, job_id):
"""
Return selected TV episode ranges grouped by season.
Example: S01 E01-E05, E07
"""
rows = conn.execute(
"""
SELECT
season_number,
episode_number
FROM job_titles
WHERE job_id = ?
AND selected = 1
AND season_number IS NOT NULL
AND episode_number IS NOT NULL
ORDER BY season_number, episode_number
""",
(job_id,),
).fetchall()
if not rows:
return None
by_season = {}
for row in rows:
season = int(row["season_number"])
episode = int(row["episode_number"])
by_season.setdefault(
season,
[],
).append(episode)
parts = []
for season in sorted(by_season):
numbers = sorted(
set(by_season[season])
)
ranges = []
start = previous = numbers[0]
for number in numbers[1:]:
if number == previous + 1:
previous = number
continue
if start == previous:
ranges.append(
f"E{start:02d}"
)
else:
ranges.append(
f"E{start:02d}-E{previous:02d}"
)
start = previous = number
if start == previous:
ranges.append(
f"E{start:02d}"
)
else:
ranges.append(
f"E{start:02d}-E{previous:02d}"
)
parts.append(
f"S{season:02d} "
+ ",".join(ranges)
)
return " | ".join(parts)
def render_jobs_history():
"""
Print the complete ARM job history once and return.
"""
with sqlite3.connect(DB) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"""
SELECT
id,
drive,
disc_label,
title,
media_type,
status,
destination,
error_code
FROM jobs
ORDER BY id
"""
).fetchall()
print()
print(
f"{BOLD}ARM JOB HISTORY{RESET}"
)
print("-" * 118)
print(
f"{'ID':<6}"
f"{'TYPE':<8}"
f"{'STATUS':<24}"
f"{'DRIVE':<8}"
f"{'LABEL':<34}"
f"TITLE"
)
print("-" * 118)
if not rows:
print("No ARM jobs found.")
return
for row in rows:
drive = (
row["drive"].replace("/dev/", "")
if row["drive"]
else "-"
)
label = (
row["disc_label"]
or "-"
)
title = (
row["title"]
or "-"
)
print(
f"{row['id']:<6}"
f"{str(row['media_type'] or '-'):<8}"
f"{str(row['status'] or '-'):<24}"
f"{drive:<8}"
f"{label[:33]:<34}"
f"{title}"
)
if row["media_type"] == "TV":
episodes = job_episode_summary(
conn,
row["id"],
)
if episodes:
print(
f" Episodes: {episodes}"
)
print(
" Destination: "
f"{row['destination'] or '-'}"
)
if row["error_code"]:
print(
" Error: "
f"{row['error_code']}"
)
print()
print("-" * 118)
print(
f"{len(rows)} job(s)"
)
def main():
args = {
arg.lower()
for arg in sys.argv[1:]
}
show_jobs = (
"--jobs" in args
or "-jobs" in args
)
show_all = (
"--all" in args
or "-all" in args
)
if show_jobs:
render_jobs_history()
return 0
previous_sizes = {}
# Put terminal into cbreak mode so single keys work.
old_settings = termios.tcgetattr(sys.stdin)
try:
tty.setcbreak(sys.stdin.fileno())
while True:
clear()
now = datetime.now().astimezone()
service = run([
"systemctl",
"is-active",
"arm.service",
])
service_indicator = (
f"{GREEN}● ACTIVE{RESET}"
if service == "active"
else f"{RED}● "
f"{service.upper() or 'UNKNOWN'}"
f"{RESET}"
)
if show_all:
print(
f"{CYAN}{BOLD}"
"╔══════════════════════════════════════════════════════════════════════════════╗"
f"{RESET}"
)
print(
f"{CYAN}{BOLD}"
"║ ARM MACHINE STATUS ║"
f"{RESET}"
)
print(
f"{CYAN}"
"╚══════════════════════════════════════════════════════════════════════════════╝"
f"{RESET}"
)
print()
print(
f"ARM service: {service_indicator}"
)
print(
"Last refresh:",
now.strftime(
"%Y-%m-%d %H:%M:%S %Z"
),
)
try:
with sqlite3.connect(DB) as conn:
conn.row_factory = sqlite3.Row
jobs = conn.execute("""
SELECT
id,
drive,
disc_label,
status,
rip_started_at,
rip_completed_at,
transcode_started_at,
transcode_completed_at,
destination,
error_code,
error_message
FROM jobs
ORDER BY id DESC
LIMIT 5
""").fetchall()
drive_map = drive_jobs(jobs)
if not show_all:
render_compact(
conn,
service_indicator,
now,
previous_sizes,
)
key = read_key(timeout=2)
if key:
if key.lower() == "q":
return 0
if key.lower() == "e":
eject_menu()
if key.lower() == "a":
toggle_auto_eject()
if key.lower() == "c":
job_action_menu(
"correct",
old_settings,
)
if key.lower() == "d":
job_action_menu(
"delete",
old_settings,
)
continue
# -------------------------------------------------------
# CURRENT ACTIVITY
# -------------------------------------------------------
print()
print(
f"{BOLD}CURRENT ACTIVITY{RESET}"
)
print("-" * 120)
ripping = [
row
for row in jobs
if row["status"] == "RIPPING"
]
transcoding = [
row
for row in jobs
if row["status"] == "TRANSCODING"
]
selecting = [
row
for row in jobs
if row["status"] == "TITLE_SELECTED"
]
processes = process_lines()
inspecting = [
line
for line in processes
if "makemkvcon" in line.lower()
and " info " in (
" " + line.lower() + " "
)
]
if inspecting and not ripping and not transcoding:
print(
f"{YELLOW}● INSPECTING{RESET} "
"disc identification / title scan"
)
else:
print(
f"{DIM}"
"○ INSPECTING none"
f"{RESET}"
)
if ripping:
for row in ripping:
print(
f"{GREEN}● RIPPING{RESET} "
f"{row['disc_label']} "
f"({row['drive']}) "
f"[{elapsed(row['rip_started_at'])}]"
)
else:
print(
f"{DIM}"
"○ RIPPING none"
f"{RESET}"
)
if transcoding:
for row in transcoding:
print(
f"{GREEN}● TRANSCODING{RESET} "
f"{row['disc_label']} "
f"({row['drive']}) "
f"[{elapsed(row['transcode_started_at'])}]"
)
else:
print(
f"{DIM}"
"○ TRANSCODING none"
f"{RESET}"
)
if selecting:
for row in selecting:
print(
f"{YELLOW}● TITLE SELECT{RESET} "
f"{row['disc_label']} "
f"({row['drive']})"
)
# -------------------------------------------------------
# DRIVES
# -------------------------------------------------------
print()
print(
f"{BOLD}DRIVES{RESET}"
)
print("-" * 120)
for device in configured_drives():
drive_number = device[-1]
rows = drive_map[device]
current = next(
(
row
for row in rows
if row["status"]
not in ("LIBRARY_MOVED",)
),
None,
)
media = drive_media_state(
device
)
if current:
print(
f"Drive {drive_number}: "
f"{GREEN}● "
f"{current['status']:<12}"
f"{RESET} "
f"{device:<10} "
f"{current['disc_label'] or 'UNKNOWN':<28}"
)
elif media == "MEDIA":
print(
f"Drive {drive_number}: "
f"{YELLOW}● DISC READY"
f"{RESET} "
f"{device:<10} "
f"[{GREEN}E{RESET}] eject"
)
else:
print(
f"Drive {drive_number}: "
f"{DIM}○ EMPTY{RESET} "
f"{device}"
)
# -------------------------------------------------------
# PIPELINE
# -------------------------------------------------------
print()
print(
f"{BOLD}PIPELINE{RESET}"
)
print("-" * 120)
active_statuses = {
"RIPPING",
"TRANSCODING",
"TITLE_SELECTED",
"MOVING",
}
drive0_active = any(
r["drive"] == "/dev/sr0"
and r["status"] in active_statuses
for r in jobs
)
drive1_active = any(
r["drive"] == "/dev/sr1"
and r["status"] in active_statuses
for r in jobs
)
print(
"Drive 0: "
+ (
f"{GREEN}● ACTIVE{RESET}"
if drive0_active
else f"{DIM}○ IDLE{RESET}"
)
)
print(
"Drive 1: "
+ (
f"{GREEN}● ACTIVE{RESET}"
if drive1_active
else f"{DIM}○ IDLE{RESET}"
)
)
print(
"Transcode: "
+ (
f"{GREEN}● ACTIVE{RESET}"
if transcoding
else f"{DIM}○ IDLE{RESET}"
)
)
# -------------------------------------------------------
# LAST 5 JOBS
# -------------------------------------------------------
print()
print(
f"{BOLD}LAST 5 JOBS{RESET}"
)
print("-" * 120)
print(
f"{'ID':<4} "
f"{'DRIVE':<11} "
f"{'DISC':<28} "
f"{'STATUS':<22} "
f"{'AGE':<10}"
)
print("-" * 120)
for row in jobs:
if row["status"] == "RIPPING":
start = row["rip_started_at"]
elif row["status"] == "TRANSCODING":
start = row[
"transcode_started_at"
]
else:
start = row[
"rip_started_at"
]
label = (
row["disc_label"] or ""
)[:27]
print(
f"{row['id']:<4} "
f"{row['drive']:<11} "
f"{label:<28} "
f"{status_text(row['status']):<31} "
f"{elapsed(start):<10}"
)
# -------------------------------------------------------
# TV / MOVIE LIBRARY
# -------------------------------------------------------
print()
print(
f"{BOLD}TV LIBRARY{RESET}"
)
print("-" * 120)
tv_library, movie_library = library_status(conn)
if tv_library:
for (show, season), episodes in sorted(
tv_library.items(),
key=lambda item: (
item[0][0].lower(),
item[0][1],
),
):
# Put the status indicators before the season
# and episode text so the dashboard can be
# scanned visually from left to right.
parts = []
for episode in sorted(episodes):
indicator = episodes[episode]
if indicator == "✓":
symbol = f"{GREEN}✓{RESET}"
elif indicator == "●":
symbol = f"{YELLOW}●{RESET}"
elif indicator == "○":
symbol = f"{DIM}○{RESET}"
else:
symbol = f"{RED}✗{RESET}"
parts.append(
f"{symbol} E{episode:02d}"
)
print(
f"{GREEN}✓{RESET} {show}"
)
if parts:
print(
f" "
f"{GREEN}✓{RESET} S{season:02d} "
+ " ".join(parts)
)
else:
print(
f"{DIM}"
"No TV library entries yet."
f"{RESET}"
)
print()
print(
f"{BOLD}MOVIE LIBRARY{RESET}"
)
print("-" * 120)
if movie_library:
movie_name = (
movie_library["title"]
or movie_library["disc_label"]
or "UNKNOWN MOVIE"
)
movie_status = movie_library["status"]
if movie_status == "LIBRARY_MOVED":
print(
f"{GREEN}✓{RESET} "
f"{movie_name}"
)
elif movie_status in (
"RIPPING",
"TRANSCODING",
):
print(
f"{YELLOW}●{RESET} "
f"{movie_name}"
)
elif movie_status in (
"RIP_FAILED",
"TRANSCODE_FAILED",
):
print(
f"{RED}✗{RESET} "
f"{movie_name}"
)
else:
print(
f"{YELLOW}●{RESET} "
f"{movie_name}"
)
else:
print(
f"{DIM}"
"No recent movie processing."
f"{RESET}"
)
# -------------------------------------------------------
# ACTIVE PROCESSES
# -------------------------------------------------------
print()
print(
f"{BOLD}ACTIVE PROCESSES{RESET}"
)
print("-" * 120)
makemkv, handbrake, processes = (
process_summary()
)
if makemkv:
for line in makemkv:
print(line[:119])
if handbrake:
for line in handbrake:
print(line[:119])
if not makemkv and not handbrake:
print(
f"{DIM}"
"No MakeMKV or HandBrake process "
"currently detected."
f"{RESET}"
)
# -------------------------------------------------------
# RIP OUTPUT
# -------------------------------------------------------
print()
print(
f"{BOLD}RIP OUTPUT{RESET}"
)
print("-" * 120)
if os.path.isdir(RIP_ROOT):
entries = sorted(
os.scandir(RIP_ROOT),
key=lambda entry: entry.name,
)
found = False
for entry in entries:
if not entry.is_dir():
continue
found = True
size = directory_size(
entry.path
)
old = previous_sizes.get(
entry.name
)
growth = ""
if old is not None:
delta = size - old
if delta > 0:
growth = (
f" {GREEN}+"
f"{bytes_text(delta)}"
f" since last refresh"
f"{RESET}"
)
elif delta == 0:
growth = (
f" {YELLOW}"
"no growth since last refresh"
f"{RESET}"
)
previous_sizes[
entry.name
] = size
print(
f"{entry.name:<30}"
f"{bytes_text(size):>12}"
f"{growth}"
)
if not found:
print(
f"{DIM}"
"No active rip directories."
f"{RESET}"
)
else:
print(
f"{DIM}"
f"{RIP_ROOT} does not exist."
f"{RESET}"
)
# -------------------------------------------------------
# RECENT ARM ACTIVITY
# -------------------------------------------------------
print()
print(
f"{BOLD}RECENT ARM ACTIVITY{RESET}"
)
print("-" * 120)
events = conn.execute("""
SELECT
timestamp,
level,
event,
message
FROM events
ORDER BY timestamp DESC
LIMIT 8
""").fetchall()
for event in reversed(events):
timestamp = (
event["timestamp"]
or ""
)
timestamp = (
timestamp
.replace("T", " ")
[:19]
)
message = (
event["message"]
or ""
).replace(
"\n",
" ",
)
print(
f"{timestamp} "
f"{event['level']:<5} "
f"{event['event']:<28} "
f"{message[:55]}"
)
except Exception as exc:
print()
print(
f"{RED}"
f"DATABASE ERROR: {exc}"
f"{RESET}"
)
# ---------------------------------------------------------------
# SYSTEM
# ---------------------------------------------------------------
print()
print(
f"{BOLD}SYSTEM{RESET}"
)
print("-" * 120)
load = run([
"awk",
"{print $1}",
"/proc/loadavg",
])
memory = run([
"bash",
"-c",
"free -h | "
"awk '/^Mem:/ "
"{print $3 \" / \" $2}'",
])
disk = run([
"bash",
"-c",
"df -h /rips | "
"awk 'NR==2 "
"{print $3 \" / \" $2 \" (\" $5 \")\"}'",
])
print(
f"Load: {load or '-':<10}"
f"RAM: {memory or '-':<20}"
f"/rips: {disk or '-'}"
)
# ---------------------------------------------------------------
# KEYBOARD CONTROLS
# ---------------------------------------------------------------
print()
print(
f"{BOLD}"
"[E] Eject [A] Auto-eject "
"[C] Correct [D] Delete [Q] Quit"
f"{RESET}"
)
print(
f"{DIM}"
"Press a key at any time • "
"Refreshing every 2 seconds"
f"{RESET}"
)
# Wait for one key, but refresh automatically after 2 seconds.
key = read_key(timeout=2)
if key is None:
continue
key = key.lower()
if key == "q":
clear()
print(
"ARM status dashboard closed."
)
break
if key == "e":
eject_menu()
if key == "a":
toggle_auto_eject()
if key == "c":
job_action_menu(
"correct",
old_settings,
)
if key == "d":
job_action_menu(
"delete",
old_settings,
)
finally:
# Always restore the user's terminal settings.
termios.tcsetattr(
sys.stdin,
termios.TCSADRAIN,
old_settings,
)
clear()
if __name__ == "__main__":
main()scripts/arm_config.py
#!/usr/bin/env python3
from configparser import ConfigParser
from pathlib import Path
CONFIG_FILE = Path("/opt/arm/config/arm.conf")
def load_config():
if not CONFIG_FILE.exists():
raise FileNotFoundError(f"ARM configuration not found: {CONFIG_FILE}")
config = ConfigParser()
config.read(CONFIG_FILE)
return config
def get_paths(config):
return {
"rip_root": Path(config["ripping"]["rip_root"]),
"rip_work": Path(config["ripping"]["rip_work"]),
"media_root": Path(config["jellyfin"]["media_root"]),
"state": Path(config["state"]["root"]),
"logs": Path(config["state"]["logs"]),
"backups": Path(config["state"]["backups"]),
}
def get_drives(config):
drives = []
for key, value in config["optical"].items():
if key.startswith("drive_"):
drives.append(value)
return drives
if __name__ == "__main__":
config = load_config()
paths = get_paths(config)
drives = get_drives(config)
print("ARM configuration loaded successfully.")
print()
print("Drives:")
for drive in drives:
print(f" {drive}")
print()
print("Paths:")
for name, path in paths.items():
print(f" {name}: {path}")scripts/arm_logger.py
#!/usr/bin/env python3
import logging
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
DB = Path("/opt/arm/state/arm.db")
LOG_DIR = Path("/opt/arm/logs")
LOG_FILE = LOG_DIR / "arm.log"
def utc_now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def get_logger():
LOG_DIR.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("ARM")
logger.setLevel(logging.INFO)
if not logger.handlers:
handler = logging.FileHandler(LOG_FILE)
formatter = logging.Formatter(
"%(asctime)s [%(levelname)s] %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def log_event(level, event, message, job_id=None):
timestamp = utc_now()
logger = get_logger()
log_method = getattr(logger, level.lower(), logger.info)
log_method(f"{event}: {message}")
with sqlite3.connect(DB) as conn:
conn.execute(
"""
INSERT INTO events (
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(
job_id,
timestamp,
level.upper(),
event,
message,
),
)
def info(event, message, job_id=None):
log_event("INFO", event, message, job_id)
def warning(event, message, job_id=None):
log_event("WARNING", event, message, job_id)
def error(event, message, job_id=None):
log_event("ERROR", event, message, job_id)
if __name__ == "__main__":
info(
"SYSTEM_TEST",
"ARM logging subsystem initialized successfully"
)
print(f"Log file: {LOG_FILE}")
print("Database event recorded.")scripts/init_db.py
#!/usr/bin/env python3
import sqlite3
from pathlib import Path
DB = Path("/opt/arm/state/arm.db")
DB.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(DB) as conn:
conn.execute("PRAGMA foreign_keys = ON")
conn.executescript("""
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_uuid TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
drive TEXT,
disc_label TEXT,
disc_source TEXT,
media_type TEXT,
title TEXT,
year INTEGER,
status TEXT NOT NULL,
rip_started_at TEXT,
rip_completed_at TEXT,
transcode_started_at TEXT,
transcode_completed_at TEXT,
destination TEXT,
error_code TEXT,
error_message TEXT,
retry_count INTEGER NOT NULL DEFAULT 0,
notes TEXT
);
CREATE TABLE IF NOT EXISTS job_titles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_id INTEGER NOT NULL,
title_number INTEGER,
name TEXT,
duration_seconds INTEGER,
duration TEXT,
size_bytes INTEGER,
size_display TEXT,
chapters INTEGER,
source_file TEXT,
selected INTEGER NOT NULL DEFAULT 0,
makemkv_title_id INTEGER,
FOREIGN KEY(job_id) REFERENCES jobs(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_id INTEGER,
timestamp TEXT NOT NULL,
level TEXT NOT NULL,
event TEXT NOT NULL,
message TEXT,
FOREIGN KEY(job_id) REFERENCES jobs(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS system_state (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_jobs_status
ON jobs(status);
CREATE INDEX IF NOT EXISTS idx_jobs_created
ON jobs(created_at);
CREATE INDEX IF NOT EXISTS idx_events_job
ON events(job_id);
""")
print(f"ARM database initialized: {DB}")scripts/arm_orchestrator.py
#!/usr/bin/env python3
import configparser
import sqlite3
import contextlib
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
# Active disc_inspector.py children, keyed by optical config key.
# Example: drive_0 -> (device, label, Popen object)
ACTIVE_INSPECTIONS = {}
# Active ripper.py children, keyed by device path.
# Example: /dev/sr0 -> Popen object
ACTIVE_RIPPERS = {}
# Exactly one transcoder.py worker may run at a time.
ACTIVE_TRANSCODER = None
@contextlib.contextmanager
def db_connection():
"""
Open an ARM SQLite connection and always close it.
sqlite3.Connection's own context manager commits/rolls back but
does not close the connection, which leaks file descriptors in
this long-running orchestrator.
"""
conn = sqlite3.connect(DB)
try:
with conn:
yield conn
finally:
conn.close()
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def log_event(conn, level, event, message, job_id=None):
conn.execute(
"""
INSERT INTO events(
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(
job_id,
now(),
level,
event,
message,
),
)
conn.commit()
def process_exists(pattern):
import subprocess
result = subprocess.run(
["pgrep", "-f", pattern],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
return bool(result.stdout.strip())
def recover_orphaned_jobs():
"""
Recover jobs left active in the database after an interrupted process.
This only repairs database state. It does not delete files or retry work.
"""
with db_connection() as conn:
jobs = conn.execute(
"""
SELECT id, disc_label, status
FROM jobs
WHERE status IN ('RIPPING','TRANSCODING')
ORDER BY id
"""
).fetchall()
for job_id, label, status in jobs:
if status == "RIPPING":
if not process_exists("makemkvcon.*mkv"):
print(
f"RECOVERY: Job {job_id} "
f"{label} reset RIPPING -> TITLE_SELECTED"
)
conn.execute(
"""
UPDATE jobs
SET status='TITLE_SELECTED'
WHERE id=?
""",
(job_id,),
)
log_event(
conn,
"WARNING",
"JOB_RECOVERED",
"Reset orphaned RIPPING job",
job_id,
)
elif status == "TRANSCODING":
if not process_exists("HandBrakeCLI"):
print(
f"RECOVERY: Job {job_id} "
f"{label} reset TRANSCODING -> RIPPED"
)
conn.execute(
"""
UPDATE jobs
SET status='RIPPED'
WHERE id=?
""",
(job_id,),
)
log_event(
conn,
"WARNING",
"JOB_RECOVERED",
"Reset orphaned TRANSCODING job",
job_id,
)
conn.commit()
def get_state(conn, key):
row = conn.execute(
"""
SELECT value
FROM system_state
WHERE key = ?
""",
(key,),
).fetchone()
return row[0] if row else None
def set_state(conn, key, value):
conn.execute(
"""
INSERT INTO system_state(key, value, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(key)
DO UPDATE SET
value = excluded.value,
updated_at = excluded.updated_at
""",
(
key,
value,
now(),
),
)
conn.commit()
def run_script(script, *args):
command = [
"python3",
str(script),
*args,
]
print()
print("=" * 70)
print(f"RUNNING: {script}")
print("=" * 70)
result = subprocess.run(
command,
check=False,
)
print(f"RESULT: {script.name} returncode={result.returncode}")
return result.returncode
def detect_disc(drive):
"""
Return the MakeMKV disc label when a disc is present.
Return None when the drive is empty or MakeMKV cannot identify
a disc.
"""
try:
result = subprocess.run(
[
"makemkvcon",
"-r",
"info",
f"disc:{drive}",
],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
timeout=120,
check=False,
)
except Exception as exc:
print(
f"Disc detection exception on {drive}: {exc}"
)
return None
# MakeMKV can return a non-zero status while still providing
# valid DRV records describing the installed optical discs.
# Parse the DRV records instead of rejecting the output solely
# because of the process return code.
for line in result.stdout.splitlines():
if not line.startswith("DRV:"):
continue
parts = line.split(",")
if len(parts) < 7:
continue
device = parts[6].strip('"').strip()
if device != drive:
continue
label = parts[5].strip('"').strip()
if label:
return label
return None
def auto_eject_drive(config, drive):
"""
Eject a drive after its optical rip is completely finished.
This never closes a tray. It only issues an eject/open command
when [automation] auto_eject is enabled.
"""
automation = config["automation"]
if not automation.getboolean(
"auto_eject",
fallback=False,
):
return False
# Confirm this drive has no remaining selected optical work.
with db_connection() as conn:
remaining = conn.execute(
"""
SELECT COUNT(*)
FROM jobs j
JOIN job_titles jt
ON jt.job_id = j.id
WHERE j.drive = ?
AND j.status IN (
'TITLE_SELECTED',
'RIPPING'
)
AND jt.selected = 1
""",
(drive,),
).fetchone()[0]
if remaining:
print(
f"{drive}: auto-eject deferred; "
f"{remaining} selected title(s) still need optical work"
)
return False
print(f"{drive}: optical work complete; auto-ejecting tray...")
try:
result = subprocess.run(
["eject", drive],
capture_output=True,
text=True,
timeout=10,
)
except Exception as exc:
print(
f"{drive}: auto-eject failed: {exc}"
)
with db_connection() as conn:
log_event(
conn,
"ERROR",
"AUTO_EJECT_FAILED",
f"drive={drive}; error={exc}",
)
return False
if result.returncode != 0:
message = (
result.stderr.strip()
or result.stdout.strip()
or f"eject returned {result.returncode}"
)
print(
f"{drive}: auto-eject failed: {message}"
)
with db_connection() as conn:
log_event(
conn,
"ERROR",
"AUTO_EJECT_FAILED",
f"drive={drive}; error={message}",
)
return False
with db_connection() as conn:
# Clear remembered media immediately. The next closed/new
# disc can therefore be treated as a genuine insertion.
for drive_key, device in config["optical"].items():
if (
drive_key.startswith("drive_")
and device == drive
):
set_state(
conn,
f"{drive_key}_disc_label",
"",
)
set_state(
conn,
f"{drive_key}_tray_open",
"1",
)
break
log_event(
conn,
"INFO",
"AUTO_EJECTED",
f"drive={drive}",
)
print(f"{drive}: tray ejected; ready for replacement.")
return True
def process_pipeline(config):
"""
Advance jobs without blocking the ARM main loop.
Optical drives rip independently, with at most one ripper worker
per drive. Exactly one transcoder worker may run at a time.
This allows inspection, title selection, ripping, transcoding,
and library movement for different jobs to overlap safely.
"""
global ACTIVE_RIPPERS
global ACTIVE_TRANSCODER
automation = config["automation"]
# ------------------------------------------------------------
# Reap completed ripper workers.
# ------------------------------------------------------------
for drive, process in list(ACTIVE_RIPPERS.items()):
rc = process.poll()
if rc is None:
continue
print(
f"Ripper finished for {drive}; "
f"returncode={rc}"
)
del ACTIVE_RIPPERS[drive]
# The optical disc is no longer required once ripping
# finishes successfully. Transcoding/library work continues
# independently from SSD staging.
if rc == 0:
auto_eject_drive(
config,
drive,
)
# ------------------------------------------------------------
# Reap the single transcode worker.
# ------------------------------------------------------------
if ACTIVE_TRANSCODER is not None:
rc = ACTIVE_TRANSCODER.poll()
if rc is not None:
print(
"Transcoder worker finished; "
f"returncode={rc}"
)
ACTIVE_TRANSCODER = None
# ------------------------------------------------------------
# Title selection.
#
# This is short-lived and may safely run synchronously.
# ------------------------------------------------------------
if automation.getboolean(
"auto_rip",
fallback=True,
):
rc = run_script(
Path("/opt/arm/scripts/title_selector.py")
)
if rc != 0:
print(
"Pipeline stage failed: title_selector.py"
)
return False
# --------------------------------------------------------
# Start one ripper per eligible optical drive.
#
# Do not wait for these processes. The main ARM loop must
# remain free to inspect and advance other drives.
# --------------------------------------------------------
with db_connection() as conn:
for key, drive in config["optical"].items():
if not key.startswith("drive_"):
continue
active = ACTIVE_RIPPERS.get(drive)
if active is not None:
if active.poll() is None:
continue
del ACTIVE_RIPPERS[drive]
eligible = conn.execute(
"""
SELECT 1
FROM jobs j
JOIN job_titles jt
ON jt.job_id = j.id
WHERE j.status = 'TITLE_SELECTED'
AND j.drive = ?
AND jt.selected = 1
AND NOT EXISTS (
SELECT 1
FROM job_files jf
WHERE jf.job_id = j.id
AND jf.job_title_id = jt.id
AND jf.status = 'RIPPED'
)
LIMIT 1
""",
(drive,),
).fetchone()
if not eligible:
continue
command = [
"python3",
"/opt/arm/scripts/ripper.py",
drive,
]
process = subprocess.Popen(command)
ACTIVE_RIPPERS[drive] = process
print(
f"Started ripper for {drive}; "
f"pid={process.pid}"
)
# ------------------------------------------------------------
# Single transcode queue worker.
#
# If any RIPPED job exists and no transcoder worker is active,
# start exactly one transcoder.py process.
# ------------------------------------------------------------
if automation.getboolean(
"auto_transcode",
fallback=True,
):
if ACTIVE_TRANSCODER is None:
with db_connection() as conn:
eligible = conn.execute(
"""
SELECT id
FROM jobs
WHERE status = 'RIPPED'
ORDER BY id
LIMIT 1
"""
).fetchone()
if eligible:
ACTIVE_TRANSCODER = subprocess.Popen(
[
"python3",
"/opt/arm/scripts/transcoder.py",
]
)
print(
"Started transcode worker; "
f"pid={ACTIVE_TRANSCODER.pid}; "
f"queued_job={eligible[0]}"
)
# ------------------------------------------------------------
# Library movement.
#
# This stage is intentionally synchronous because moves are
# short. It may move a TRANSCODED job while other drives rip
# or another job transcodes.
# ------------------------------------------------------------
if automation.getboolean(
"auto_library_move",
fallback=True,
):
rc = run_script(
Path("/opt/arm/scripts/library_mover.py")
)
if rc != 0:
return False
if automation.getboolean(
"auto_cleanup_rips",
fallback=True,
):
cleanup_script = Path(
"/opt/arm/scripts/cleanup_rips.py"
)
if cleanup_script.exists():
rc = run_script(
cleanup_script,
"--execute",
)
if rc != 0:
return False
return True
def inspect_new_discs(config):
"""
Detect insertion/removal transitions and run inspections independently.
A drive must never be probed while either disc_inspector.py or
ripper.py owns it. MakeMKV info calls against an actively-ripping
optical drive can temporarily make media detection look like a
removal/reinsertion and create duplicate jobs.
"""
global ACTIVE_INSPECTIONS
global ACTIVE_RIPPERS
drives = [
(key, value)
for key, value in config["optical"].items()
if key.startswith("drive_")
]
# Reap completed inspector children first.
for drive_key, active in list(ACTIVE_INSPECTIONS.items()):
device, label, process = active
rc = process.poll()
if rc is None:
continue
print(
f"Disc inspection finished for {device}; "
f"returncode={rc}"
)
with db_connection() as conn:
if rc == 0:
# Only mark the disc as handled after inspection succeeds.
set_state(
conn,
f"{drive_key}_disc_label",
label,
)
else:
log_event(
conn,
"ERROR",
"DISC_INSPECTION_FAILED",
(
f"disc inspector returned {rc}; "
f"drive={device}; label={label}"
),
)
del ACTIVE_INSPECTIONS[drive_key]
with db_connection() as conn:
for drive_key, device in drives:
# Never probe a drive while its inspector owns it.
active = ACTIVE_INSPECTIONS.get(drive_key)
if active is not None:
_, label, process = active
print(
f"{device}: inspection active "
f"(pid={process.pid}, label={label})"
)
continue
# Never run MakeMKV/media detection against a drive while
# ripper.py owns it. The ripper must have exclusive access
# to the optical device until the rip completes.
ripper = ACTIVE_RIPPERS.get(device)
if ripper is not None:
if ripper.poll() is None:
print(
f"{device}: rip active "
f"(pid={ripper.pid}); inspection deferred"
)
continue
current_label = detect_disc(device)
state_key = f"{drive_key}_disc_label"
previous_label = get_state(
conn,
state_key,
)
if current_label:
print(
f"{device}: disc present: "
f"{current_label}"
)
if previous_label != current_label:
print(
f"NEW DISC DETECTED: "
f"{device} -> {current_label}"
)
log_event(
conn,
"INFO",
"DISC_INSERTED",
(
f"drive={device}; "
f"label={current_label}"
),
)
process = subprocess.Popen(
[
"python3",
"/opt/arm/scripts/disc_inspector.py",
device,
],
)
ACTIVE_INSPECTIONS[drive_key] = (
device,
current_label,
process,
)
print(
f"Started disc inspection for "
f"{device}; pid={process.pid}"
)
else:
print(
f"{device}: same disc still present."
)
else:
if previous_label:
print(
f"{device}: disc removed."
)
log_event(
conn,
"INFO",
"DISC_REMOVED",
(
f"drive={device}; "
f"previous_label={previous_label}"
),
)
set_state(
conn,
state_key,
"",
)
else:
print(
f"{device}: empty."
)
def main():
config = load_config()
poll_seconds = config["automation"].getint(
"poll_seconds",
fallback=15,
)
print("ARM orchestrator starting.")
print(f"Database: {DB}")
print(f"Poll interval: {poll_seconds}s")
if not DB.exists():
raise SystemExit(
f"ARM database does not exist: {DB}"
)
while True:
try:
# Reload configuration every cycle so dashboard toggles
# such as auto_eject take effect without restarting ARM.
config = load_config()
recover_orphaned_jobs()
inspect_new_discs(config)
process_pipeline(config)
except KeyboardInterrupt:
print()
print("ARM orchestrator stopped.")
return 0
except Exception as exc:
print(
f"ARM orchestrator exception: {exc}"
)
try:
with db_connection() as conn:
log_event(
conn,
"ERROR",
"ORCHESTRATOR_EXCEPTION",
str(exc),
)
except Exception:
pass
time.sleep(poll_seconds)
if __name__ == "__main__":
raise SystemExit(main())scripts/disc_inspector.py
#!/usr/bin/env python3
import configparser
import hashlib
import json
import re
import sqlite3
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def run_makemkv(drive):
try:
result = subprocess.run(
["makemkvcon", "-r", "info", f"disc:{drive}"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
timeout=600,
check=False,
)
return result.returncode, result.stdout, None
except subprocess.TimeoutExpired as exc:
output = exc.stdout or ""
if isinstance(output, bytes):
output = output.decode(errors="replace")
return None, output, "INSPECTION_TIMEOUT"
def parse_makemkv(output):
disc = {
"label": None,
"source": None,
"titles": [],
}
titles = {}
for line in output.splitlines():
if line.startswith("CINFO:"):
parts = line.split(",", 2)
if len(parts) != 3:
continue
field = parts[0].split(":", 1)[1]
value = parts[2].strip('"')
if field == "2":
disc["label"] = value
elif field == "32":
disc["source"] = value
elif line.startswith("TINFO:"):
match = re.match(
r'^TINFO:(\d+),(\d+),\d+,"(.*)"$',
line,
)
if not match:
continue
makemkv_title_id = int(match.group(1))
field = match.group(2)
value = match.group(3)
if makemkv_title_id not in titles:
titles[makemkv_title_id] = {
"title_number": makemkv_title_id,
"name": None,
"makemkv_title_id": makemkv_title_id,
"duration": None,
"size_display": None,
"size_bytes": None,
"chapters": None,
"source_file": None,
}
title = titles[makemkv_title_id]
if field == "2":
title["name"] = value
elif field == "8":
title["chapters"] = int(value)
elif field == "9":
title["duration"] = value
elif field == "10":
title["size_display"] = value
elif field == "11":
title["size_bytes"] = int(value)
elif field == "27":
title["source_file"] = value
ordered_titles = [
titles[number]
for number in sorted(titles)
]
for database_title_number, title in enumerate(
ordered_titles
):
title["title_number"] = database_title_number
disc["titles"] = ordered_titles
return disc
def duration_seconds(value):
if not value:
return None
try:
h, m, s = value.split(":")
return int(h) * 3600 + int(m) * 60 + int(s)
except ValueError:
return None
def disc_fingerprint(disc):
"""
Build a stable SHA-256 fingerprint from normalized MakeMKV
title metadata.
Deliberately excludes:
- drive path
- drive model
- timestamps
- physical disc label
so the same physical disc should fingerprint identically
in either optical drive.
Exact byte size is normalized to MiB to reduce sensitivity
to tiny reporting differences.
"""
parts = []
for title in disc.get("titles", []):
duration = duration_seconds(
title.get("duration")
) or 0
size_bytes = title.get("size_bytes") or 0
size_mib = int(
round(
size_bytes
/ (1024 * 1024)
)
)
chapters = title.get("chapters") or 0
makemkv_id = title.get(
"makemkv_title_id"
)
if makemkv_id is None:
makemkv_id = title.get(
"title_number",
0,
)
parts.append(
(
int(makemkv_id),
int(duration),
int(size_mib),
int(chapters),
)
)
normalized = "|".join(
f"{title_id}:{duration}:{size_mib}:{chapters}"
for title_id, duration, size_mib, chapters
in parts
)
return hashlib.sha256(
normalized.encode("utf-8")
).hexdigest()
def is_generic_disc_label(label):
"""
Return True for physical labels that do not meaningfully
identify the movie.
"""
if not label:
return True
normalized = re.sub(
r"[^A-Z0-9]+",
"_",
label.upper(),
).strip("_")
generic = {
"DVD_VIDEO",
"VIDEO_DVD",
"DVDVIDEO",
"VIDEO",
"DVD",
"UNTITLED",
"NO_LABEL",
"NOLABEL",
"UNKNOWN",
}
return normalized in generic
def generic_fingerprint_title(label, fingerprint):
"""
Build the unique human-facing temporary title used by ARM
and Jellyfin for a generic physical disc label.
"""
normalized = re.sub(
r"[^A-Z0-9]+",
"_",
(label or "DVD_VIDEO").upper(),
).strip("_")
if not normalized:
normalized = "DVD_VIDEO"
return f"{normalized}_{fingerprint[:12]}"
def ambiguous_tv_volume_label(label):
"""
Recognize publisher-style volume/disc labels that strongly
suggest episodic media but do NOT provide a trustworthy season.
Example:
Seinfeld - Vol. 2 (Disc 1)
Volume numbers are deliberately NOT treated as season numbers.
Returns a normalized possible show title, or None.
"""
if not label:
return None
name = label.replace("_", " ")
has_volume = re.search(
r"\b(?:VOL\.?|VOLUME)\s*\d+\b",
name,
flags=re.IGNORECASE,
)
has_disc = re.search(
r"\bDISC\s*\d+\b",
name,
flags=re.IGNORECASE,
)
if not (has_volume and has_disc):
return None
# Remove "(Disc 1)" or "Disc 1".
name = re.sub(
r"\(?\bDISC\s*\d+\b\)?",
"",
name,
flags=re.IGNORECASE,
)
# Remove "Vol. 2" or "Volume 2".
name = re.sub(
r"\b(?:VOL\.?|VOLUME)\s*\d+\b",
"",
name,
flags=re.IGNORECASE,
)
# Clean separators left behind.
name = re.sub(r"[-–—]+", " ", name)
name = re.sub(r"\s+", " ", name).strip()
name = name.title().replace(" Of ", " of ")
return name or None
def classify_disc(label, config=None):
"""
Classify a disc conservatively.
TV detection order:
1. Explicit season marker in the disc label.
2. Configured [tv_overrides] label prefix.
3. Otherwise treat the disc as a movie.
Returns:
(media_type, season_number, show_title)
"""
if not label:
return "MOVIE", None, None
normalized = label.upper().replace("-", "_")
patterns = [
r"(?:^|[_\s])SEASON[_\s]?(\d{1,2})(?:$|[_\s])",
r"(?:^|[_\s])S(\d{1,2})(?:$|[_\s])",
]
for pattern in patterns:
match = re.search(pattern, normalized)
if match:
season = int(match.group(1))
if 1 <= season <= 99:
return "TV", season, None
# Some commercial TV sets use spelled-out season numbers,
# for example BOONDOCKS_SEASON_ONE_DISC_ONE.
number_words = {
"ONE": 1,
"TWO": 2,
"THREE": 3,
"FOUR": 4,
"FIVE": 5,
"SIX": 6,
"SEVEN": 7,
"EIGHT": 8,
"NINE": 9,
"TEN": 10,
"ELEVEN": 11,
"TWELVE": 12,
"THIRTEEN": 13,
"FOURTEEN": 14,
"FIFTEEN": 15,
"SIXTEEN": 16,
"SEVENTEEN": 17,
"EIGHTEEN": 18,
"NINETEEN": 19,
"TWENTY": 20,
}
match = re.search(
r"(?:^|[_\s])SEASON[_\s]+"
r"(ONE|TWO|THREE|FOUR|FIVE|SIX|SEVEN|EIGHT|NINE|TEN|"
r"ELEVEN|TWELVE|THIRTEEN|FOURTEEN|FIFTEEN|SIXTEEN|"
r"SEVENTEEN|EIGHTEEN|NINETEEN|TWENTY)"
r"(?:$|[_\s])",
normalized,
)
if match:
return "TV", number_words[match.group(1)], None
if config is not None and config.has_section("tv_overrides"):
for prefix, value in config["tv_overrides"].items():
prefix_normalized = prefix.upper().replace("-", "_")
if not normalized.startswith(prefix_normalized):
continue
try:
show_title, season_text = value.rsplit("|", 1)
season = int(season_text.strip())
except (ValueError, AttributeError):
continue
if 1 <= season <= 99:
return "TV", season, show_title.strip() or None
return "MOVIE", None, None
def inspect_drive(conn, drive):
started = now()
rc, output, error_code = run_makemkv(drive)
if error_code:
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"ERROR",
"INSPECTION_FAILED",
f"drive=/dev/sr{drive}; error={error_code}",
),
)
print(
f" Inspection failed: {error_code}"
)
return None
# MakeMKV can return non-zero while still giving us useful
# DRV/TINFO information. Therefore parse the output before
# treating the inspection as failed.
disc = parse_makemkv(output)
if not disc["label"]:
conn.execute(
"""
INSERT INTO events(timestamp, level, event, message)
VALUES (?, ?, ?, ?)
""",
(
now(),
"INFO",
"NO_DISC",
f"drive={drive}",
),
)
return None
fingerprint = disc_fingerprint(disc)
media_type, season_number, show_title = classify_disc(
disc["label"],
load_config(),
)
# Publisher "Volume N / Disc N" labels often identify TV sets
# without revealing the real season. Treat these as TV requiring
# metadata confirmation rather than guessing that volume=season.
if media_type == "MOVIE":
ambiguous_show = ambiguous_tv_volume_label(
disc["label"]
)
if ambiguous_show:
media_type = "TV"
season_number = None
show_title = ambiguous_show
print(
" Ambiguous TV volume/disc label detected."
)
print(
f" Possible show: {show_title}"
)
print(
" Season number requires confirmation."
)
duplicate_job_id = None
if (
media_type == "MOVIE"
and is_generic_disc_label(disc["label"])
):
show_title = generic_fingerprint_title(
disc["label"],
fingerprint,
)
print(
f" Generic label detected: {disc['label']}"
)
print(
f" Effective title: {show_title}"
)
duplicate = conn.execute(
"""
SELECT id
FROM jobs
WHERE disc_fingerprint = ?
AND status = 'LIBRARY_MOVED'
ORDER BY id DESC
LIMIT 1
""",
(fingerprint,),
).fetchone()
if duplicate:
duplicate_job_id = duplicate[0]
print(
f" Known fingerprint: already completed "
f"as Job {duplicate_job_id}"
)
job_uuid = (
f"inspect-{drive}-"
f"{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S%f')}"
)
conn.execute(
"""
INSERT INTO jobs(
job_uuid,
created_at,
updated_at,
drive,
disc_label,
disc_source,
media_type,
title,
status,
notes,
disc_fingerprint
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
job_uuid,
started,
now(),
f"/dev/sr{drive}",
disc["label"],
disc["source"],
media_type,
show_title,
(
"SUPERSEDED"
if duplicate_job_id is not None
else (
"TV_METADATA_NEEDED"
if (
media_type == "TV"
and season_number is None
)
else "INSPECTED"
)
),
(
(
"Duplicate generic disc; "
f"already completed as Job {duplicate_job_id}"
)
if duplicate_job_id is not None
else (
"Created by disc inspector"
if media_type == "MOVIE"
else (
(
"Created by disc inspector; "
"TV season requires confirmation"
)
if season_number is None
else (
"Created by disc inspector; "
f"TV season {season_number}"
)
)
)
),
fingerprint,
),
)
job_id = conn.execute(
"SELECT id FROM jobs WHERE job_uuid = ?",
(job_uuid,),
).fetchone()[0]
for title in disc["titles"]:
if (
not title.get("duration")
and not title.get("source_file")
):
continue
conn.execute(
"""
INSERT INTO job_titles(
job_id,
title_number,
name,
duration_seconds,
duration,
size_bytes,
size_display,
chapters,
source_file,
selected,
makemkv_title_id,
season_number,
episode_number
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)
""",
(
job_id,
title["title_number"],
title.get("name"),
duration_seconds(
title.get("duration")
),
title.get("duration"),
title.get("size_bytes"),
title.get("size_display"),
title.get("chapters"),
title.get("source_file"),
title.get("makemkv_title_id"),
season_number,
None,
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"INFO",
"DISC_INSPECTED",
json.dumps({
"job_id": job_id,
"drive": drive,
"label": disc["label"],
"media_type": media_type,
"season_number": season_number,
"titles": len(disc["titles"]),
"fingerprint": fingerprint,
"fingerprint_short": fingerprint[:12],
"duplicate_job_id": duplicate_job_id,
"makemkv_returncode": rc,
}),
),
)
print(
f" Disc classification: {media_type}"
)
print(
f" Fingerprint: {fingerprint[:12]}"
)
if media_type == "TV":
print(
f" Season: {season_number}"
)
if duplicate_job_id is not None:
print(
f" Duplicate disc suppressed: "
f"Job {job_id} -> SUPERSEDED"
)
return job_id
def main():
config = load_config()
if len(sys.argv) > 1:
devices = [sys.argv[1]]
else:
devices = [
value
for key, value in config["optical"].items()
if key.startswith("drive_")
]
with sqlite3.connect(DB) as conn:
for device in devices:
match = re.search(
r"sr(\d+)$",
device,
)
if not match:
print(
f"Skipping invalid optical device: "
f"{device}"
)
continue
drive = match.group(1)
print(
f"Inspecting {device}..."
)
job_id = inspect_drive(
conn,
drive,
)
if job_id:
print(
f" Disc detected. "
f"Job ID: {job_id}"
)
else:
print(
" No disc detected."
)
conn.commit()
if __name__ == "__main__":
main()scripts/title_selector.py
#!/usr/bin/env python3
import configparser
import json
import re
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def score_title(title):
"""
Score a title as a likely main feature.
Used for MOVIE jobs only.
"""
duration = title["duration_seconds"] or 0
chapters = title["chapters"] or 0
size = title["size_bytes"] or 0
score = 0
if duration >= 5400:
score += 100
elif duration >= 3600:
score += 60
elif duration >= 1800:
score += 20
score += min(duration // 60, 180)
if chapters >= 10:
score += 30
elif chapters >= 5:
score += 15
elif chapters >= 2:
score += 5
score += min(
size // (1024 * 1024 * 100),
20,
)
return score
def tv_show_name(disc_label, title):
"""
Determine the canonical Jellyfin TV series name.
Supports numeric and spelled-out season/disc markers.
"""
if title:
return title.strip()
name = disc_label or ""
name = name.replace("_", " ")
number_word = (
r"(?:ONE|TWO|THREE|FOUR|FIVE|SIX|SEVEN|EIGHT|NINE|TEN|"
r"ELEVEN|TWELVE|THIRTEEN|FOURTEEN|FIFTEEN|SIXTEEN|"
r"SEVENTEEN|EIGHTEEN|NINETEEN|TWENTY)"
)
name = re.sub(
rf"\bDISC\s*(?:\d+|{number_word})\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(
rf"\bSEASON\s*(?:\d+|{number_word})\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(
r"\bS\d{1,2}\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(r"\s+", " ", name).strip()
name = name.title().replace(" Of ", " of ")
return name or "Unknown TV Show"
def promote_selected_title(conn, job_id, title_name):
"""
Copy a selected title name into jobs.title.
For TV jobs this is informational only; the library
mover will ultimately use the disc label/show name.
"""
if not title_name:
return False
conn.execute(
"""
UPDATE jobs
SET
title = CASE
WHEN title IS NULL OR TRIM(title) = ''
THEN ?
ELSE title
END,
updated_at = ?
WHERE id = ?
""",
(
title_name.strip(),
now(),
job_id,
),
)
return True
def is_tv_episode_candidate(title):
"""
Broad first-pass TV episode detector.
The final selection is made by select_tv_titles(), which groups
short-form titles by consistent chapter structure.
"""
duration = title["duration_seconds"] or 0
chapters = title["chapters"] or 0
size = title["size_bytes"] or 0
# Permit slightly-under-20-minute broadcast episodes.
if duration < 19 * 60:
return False
if duration > 75 * 60:
return False
if size and size < 500 * 1024 * 1024:
return False
if chapters < 1:
return False
return True
def choose_tv_episode_group(candidates):
"""
Choose the strongest coherent TV episode group.
For short-form television (19-35 minutes), DVDs sometimes
contain multiple alternate program-chain groups with nearly
identical durations but different chapter structures.
Conservative behavior:
- One short-form chapter-count group: keep all candidates.
- If competing short-form groups exist, automatically choose
one only when it has more titles than every other group
and contains at least 3 titles.
- If long-form TV candidates are mixed in, do not guess.
"""
if not candidates:
return []
short = [
item
for item in candidates
if (item["duration_seconds"] or 0) <= 35 * 60
]
long_form = [
item
for item in candidates
if (item["duration_seconds"] or 0) > 35 * 60
]
if not short:
return list(candidates)
by_chapters = {}
for item in short:
chapters = int(item["chapters"] or 0)
by_chapters.setdefault(
chapters,
[],
).append(item)
# No competing short-form structures.
if len(by_chapters) <= 1:
return list(candidates)
ranked = sorted(
by_chapters.items(),
key=lambda pair: (
len(pair[1]),
sum(
int(item["size_bytes"] or 0)
for item in pair[1]
),
),
reverse=True,
)
winning_chapters, winning_group = ranked[0]
runner_up_count = len(ranked[1][1])
# Mixed short/long discs are ambiguous.
if long_form:
return list(candidates)
if (
len(winning_group) >= 3
and len(winning_group) > runner_up_count
):
return list(winning_group)
return list(candidates)
def select_tv_titles(conn, job, titles):
"""
Select episode titles for a confidently classified TV disc.
Strategy:
1. Find independently strong episode candidates.
2. Reject combined titles, short extras, low-chapter bonus
features, and tiny files using is_tv_episode_candidate().
3. Select all remaining candidates in disc/title order.
4. Allow a single strong candidate for legitimate
single-episode TV discs.
5. Assign sequential episode numbers.
"""
job_id = job["id"]
candidates = [
dict(row)
for row in titles
if is_tv_episode_candidate(dict(row))
]
if not candidates:
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?,
error_code = ?,
error_message = ?
WHERE id = ?
""",
(
"SELECTION_FAILED",
now(),
"TV_EPISODES_NOT_FOUND",
"Could not identify any strong episode-sized titles.",
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"ERROR",
"TITLE_SELECTION_FAILED",
json.dumps({
"job_id": job_id,
"media_type": "TV",
"reason": "no strong episode candidates",
"candidate_count": 0,
}),
),
)
return False
selected_group = choose_tv_episode_group(
candidates
)
# Sort episodes in physical disc/title order.
selected_group.sort(
key=lambda item: item["title_number"]
)
if len(selected_group) != len(candidates):
rejected = [
item["title_number"]
for item in candidates
if item not in selected_group
]
print(
" TV structural grouping: "
f"selected {len(selected_group)} of "
f"{len(candidates)} candidates; "
f"rejected titles {rejected}"
)
# Clear every previous selection.
conn.execute(
"""
UPDATE job_titles
SET
selected = 0,
season_number = NULL,
episode_number = NULL
WHERE job_id = ?
""",
(job_id,),
)
season_number = job["season_number"]
# Determine the normalized show identity first. Episode numbering
# must continue only within the same show and season.
show_name = tv_show_name(
job["disc_label"],
job["title"],
)
# Continue TV episode numbering across successfully completed discs
# belonging to THIS show and season only.
#
# We compare normalized show names in Python because older jobs may
# not have jobs.title populated and instead derive their show name
# from the physical disc label.
previous_rows = conn.execute(
"""
SELECT
j.disc_label,
j.title,
jt.episode_number
FROM job_titles jt
JOIN jobs j
ON j.id = jt.job_id
WHERE j.media_type = 'TV'
AND j.status = 'LIBRARY_MOVED'
AND jt.selected = 1
AND jt.season_number = ?
AND jt.episode_number IS NOT NULL
""",
(season_number,),
).fetchall()
previous_max_episode = 0
for previous_row in previous_rows:
previous_show_name = tv_show_name(
previous_row["disc_label"],
previous_row["title"],
)
if previous_show_name != show_name:
continue
previous_max_episode = max(
previous_max_episode,
int(previous_row["episode_number"]),
)
# Also account for episodes already present in the Jellyfin
# library. This protects against manually added episodes and
# previous runs whose database state was not recorded as
# LIBRARY_MOVED.
config = load_config()
media_root = Path(config["jellyfin"]["media_root"])
tv_root = media_root / "TV Shows"
season_dir = (
tv_root
/ show_name
/ f"Season {int(season_number):02d}"
)
library_max_episode = 0
if season_dir.is_dir():
pattern = re.compile(
rf"^.+ - S{int(season_number):02d}E(\d{{2}})\.mp4$",
re.IGNORECASE,
)
for library_file in season_dir.iterdir():
if not library_file.is_file():
continue
match = pattern.match(library_file.name)
if match:
library_max_episode = max(
library_max_episode,
int(match.group(1)),
)
effective_max_episode = max(
previous_max_episode,
library_max_episode,
)
start_episode = effective_max_episode + 1
print(
f" Season {int(season_number):02d} episode state: "
f"DB=E{previous_max_episode:02d}, "
f"library=E{library_max_episode:02d}, "
f"next=E{start_episode:02d}"
)
for episode_number, title in enumerate(
selected_group,
start=start_episode,
):
conn.execute(
"""
UPDATE job_titles
SET
selected = 1,
season_number = ?,
episode_number = ?
WHERE id = ?
""",
(
season_number,
episode_number,
title["id"],
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"INFO",
"TV_EPISODE_SELECTED",
json.dumps({
"job_id": job_id,
"title_number": title["title_number"],
"season_number": season_number,
"episode_number": episode_number,
"duration": title["duration"],
"size": title["size_display"],
"chapters": title["chapters"],
"source_file": title["source_file"],
}),
),
)
print(
f" Selected episode "
f"S{season_number:02d}E{episode_number:02d}: "
f"title {title['title_number']} "
f"'{title['name']}' "
f"({title['duration']}, "
f"{title['size_display']}, "
f"{title['chapters']} chapters)"
)
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?,
error_code = NULL,
error_message = NULL
WHERE id = ?
""",
(
"TITLE_SELECTED",
now(),
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"INFO",
"TV_TITLES_SELECTED",
json.dumps({
"job_id": job_id,
"season_number": season_number,
"episode_count": len(selected_group),
"title_numbers": [
item["title_number"]
for item in selected_group
],
}),
),
)
return True
def select_movie_title(conn, job, titles):
"""
Existing movie behavior: select exactly one best feature.
"""
job_id = job["id"]
candidates = []
for row in titles:
title = dict(row)
title["score"] = score_title(title)
candidates.append(title)
candidates.sort(
key=lambda x: (
x["score"],
x["duration_seconds"] or 0,
x["size_bytes"] or 0,
),
reverse=True,
)
selected = candidates[0]
conn.execute(
"""
UPDATE job_titles
SET
selected = 0,
season_number = NULL,
episode_number = NULL
WHERE job_id = ?
""",
(job_id,),
)
conn.execute(
"""
UPDATE job_titles
SET selected = 1
WHERE id = ?
""",
(selected["id"],),
)
promote_selected_title(
conn,
job_id,
selected["name"],
)
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?,
error_code = NULL,
error_message = NULL
WHERE id = ?
""",
(
"TITLE_SELECTED",
now(),
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"INFO",
"TITLE_SELECTED",
json.dumps({
"job_id": job_id,
"title_number": selected["title_number"],
"name": selected["name"],
"duration": selected["duration"],
"size": selected["size_display"],
"chapters": selected["chapters"],
"source_file": selected["source_file"],
"score": selected["score"],
"reason": "highest feature candidate score",
}),
),
)
print(
f" Selected title "
f"{selected['title_number']} "
f"'{selected['name']}' "
f"({selected['duration']}, "
f"{selected['size_display']}, "
f"{selected['chapters']} chapters)"
)
return True
def select_job(conn, job):
job_id = job["id"]
titles = conn.execute(
"""
SELECT
id,
title_number,
name,
duration_seconds,
duration,
size_bytes,
size_display,
chapters,
source_file,
selected,
makemkv_title_id,
season_number,
episode_number
FROM job_titles
WHERE job_id = ?
ORDER BY title_number
""",
(job_id,),
).fetchall()
if not titles:
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?,
error_code = ?,
error_message = ?
WHERE id = ?
""",
(
"SELECTION_FAILED",
now(),
"NO_TITLES",
"No titles were available for selection.",
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?)
""",
(
now(),
"ERROR",
"TITLE_SELECTION_FAILED",
json.dumps({
"job_id": job_id,
"reason": "NO_TITLES",
}),
),
)
return False
# Never overwrite an existing selection.
existing = [
row for row in titles
if row["selected"] == 1
]
if existing:
if job["media_type"] == "TV":
print(
f" Existing TV selection: "
f"{len(existing)} title(s)"
)
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?
WHERE id = ?
""",
(
"TITLE_SELECTED",
now(),
job_id,
),
)
return True
selected = existing[0]
promote_selected_title(
conn,
job_id,
selected["name"],
)
conn.execute(
"""
UPDATE jobs
SET
status = ?,
updated_at = ?
WHERE id = ?
""",
(
"TITLE_SELECTED",
now(),
job_id,
),
)
print(
f" Existing selection: title "
f"{selected['title_number']} "
f"'{selected['name']}'"
)
return True
if job["media_type"] == "TV":
return select_tv_titles(
conn,
job,
titles,
)
return select_movie_title(
conn,
job,
titles,
)
def main():
if not DB.exists():
raise SystemExit(
f"Database does not exist: {DB}"
)
with sqlite3.connect(DB) as conn:
conn.row_factory = sqlite3.Row
jobs = conn.execute(
"""
SELECT
id,
drive,
disc_label,
media_type,
title,
year,
status,
(
SELECT season_number
FROM job_titles jt
WHERE jt.job_id = jobs.id
AND jt.season_number IS NOT NULL
LIMIT 1
) AS season_number
FROM jobs
WHERE status = 'INSPECTED'
ORDER BY id
"""
).fetchall()
if not jobs:
print(
"No INSPECTED jobs require title selection."
)
return
for job in jobs:
print(
f"Selecting titles for Job {job['id']} "
f"({job['disc_label']})..."
)
select_job(
conn,
job,
)
conn.commit()
print("Title selection complete.")
if __name__ == "__main__":
main()scripts/ripper.py
#!/usr/bin/env python3
import configparser
import json
import sqlite3
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def log_event(conn, job_id, level, event, message):
conn.execute(
"""
INSERT INTO events(
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(job_id, now(), level, event, message),
)
def get_next_title(conn, drive):
"""
Return the next selected title that has not already been ripped.
Selected titles remain selected in job_titles. Completion is
tracked separately in job_files.
"""
return conn.execute(
"""
SELECT
j.id,
j.drive,
j.disc_label,
j.media_type,
jt.id,
jt.title_number,
jt.makemkv_title_id,
jt.source_file,
jt.season_number,
jt.episode_number
FROM jobs j
JOIN job_titles jt
ON jt.job_id = j.id
WHERE j.status = 'TITLE_SELECTED'
AND j.drive = ?
AND jt.selected = 1
AND NOT EXISTS (
SELECT 1
FROM job_files jf
WHERE jf.job_id = j.id
AND jf.job_title_id = jt.id
AND jf.status = 'RIPPED'
)
ORDER BY
j.id,
jt.episode_number IS NULL,
jt.episode_number,
jt.title_number
LIMIT 1
""",
(drive,),
).fetchone()
def selected_titles_total(conn, job_id):
return conn.execute(
"""
SELECT COUNT(*)
FROM job_titles
WHERE job_id = ?
AND selected = 1
""",
(job_id,),
).fetchone()[0]
def ripped_selected_titles(conn, job_id):
return conn.execute(
"""
SELECT COUNT(*)
FROM job_titles jt
WHERE jt.job_id = ?
AND jt.selected = 1
AND EXISTS (
SELECT 1
FROM job_files jf
WHERE jf.job_id = jt.job_id
AND jf.job_title_id = jt.id
AND jf.status = 'RIPPED'
)
""",
(job_id,),
).fetchone()[0]
def ensure_job_file(
conn,
job_id,
title_row_id,
season_number,
episode_number,
):
"""
Create the per-title tracking row if it does not already exist.
"""
row = conn.execute(
"""
SELECT id
FROM job_files
WHERE job_id = ?
AND job_title_id = ?
LIMIT 1
""",
(
job_id,
title_row_id,
),
).fetchone()
if row:
return row[0]
timestamp = now()
cursor = conn.execute(
"""
INSERT INTO job_files(
job_id,
job_title_id,
season_number,
episode_number,
status,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, 'PENDING', ?, ?)
""",
(
job_id,
title_row_id,
season_number,
episode_number,
timestamp,
timestamp,
),
)
return cursor.lastrowid
def mark_job_file_failed(
conn,
job_file_id,
error_code,
message,
):
conn.execute(
"""
UPDATE job_files
SET
status = 'RIP_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
error_code,
message,
now(),
job_file_id,
),
)
def mark_job_failed(
conn,
job_id,
error_code,
message,
):
conn.execute(
"""
UPDATE jobs
SET
status = 'RIP_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
error_code,
message,
now(),
job_id,
),
)
log_event(
conn,
job_id,
"ERROR",
"RIP_FAILED",
message,
)
conn.commit()
def rip_title(conn, config, job):
(
job_id,
drive,
disc_label,
media_type,
title_row_id,
title_number,
makemkv_title_id,
source_file,
season_number,
episode_number,
) = job
rip_work = Path(config["ripping"]["rip_work"])
log_root = Path(config["state"]["logs"])
job_dir = rip_work / f"job-{job_id}"
if (
media_type == "TV"
and season_number is not None
and episode_number is not None
):
episode_dir = (
job_dir
/ f"S{season_number:02d}E{episode_number:02d}"
)
else:
episode_dir = job_dir
episode_dir.mkdir(
parents=True,
exist_ok=True,
)
log_suffix = (
f"-S{season_number:02d}E{episode_number:02d}"
if (
media_type == "TV"
and season_number is not None
and episode_number is not None
)
else ""
)
log_file = (
log_root
/ f"job-{job_id}-rip{log_suffix}.log"
)
if makemkv_title_id is None:
raise RuntimeError(
f"Job {job_id}, title {title_number} "
f"has no MakeMKV title ID"
)
makemkv_title_id = int(makemkv_title_id)
# MakeMKV title IDs are used directly by makemkvcon.
rip_title_id = makemkv_title_id
job_file_id = ensure_job_file(
conn,
job_id,
title_row_id,
season_number,
episode_number,
)
conn.execute(
"""
UPDATE job_files
SET
status = 'RIPPING',
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
job_file_id,
),
)
total_selected = selected_titles_total(
conn,
job_id,
)
ripped_before = ripped_selected_titles(
conn,
job_id,
)
conn.execute(
"""
UPDATE jobs
SET
status = 'RIPPING',
rip_started_at = COALESCE(rip_started_at, ?),
updated_at = ?
WHERE id = ?
""",
(
now(),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"RIP_STARTED",
json.dumps({
"job_file_id": job_file_id,
"media_type": media_type,
"drive": drive,
"database_title_number": title_number,
"job_title_row_id": title_row_id,
"makemkv_reported_title_id": makemkv_title_id,
"rip_selection_id": rip_title_id,
"source_file": source_file,
"destination": str(episode_dir),
"season_number": season_number,
"episode_number": episode_number,
"selected_titles_total": total_selected,
"ripped_selected_before": ripped_before,
}),
)
conn.commit()
device_number = drive.replace(
"/dev/sr",
"",
)
command = [
"makemkvcon",
"-r",
"mkv",
f"disc:{device_number}",
str(rip_title_id),
str(episode_dir),
]
print(
f"Ripping Job {job_id}: {disc_label}"
)
if media_type == "TV":
print(
f" Episode: "
f"S{season_number:02d}E{episode_number:02d}"
)
print(f" Drive: {drive}")
print(f" Database title: {title_number}")
print(
f" MakeMKV reported title: "
f"{makemkv_title_id}"
)
print(
f" MakeMKV rip selection: "
f"{rip_title_id}"
)
print(f" Output: {episode_dir}")
print(f" Log: {log_file}")
print()
try:
with log_file.open("w") as log:
result = subprocess.run(
command,
stdout=log,
stderr=subprocess.STDOUT,
text=True,
check=False,
)
except Exception as exc:
message = (
f"Exception while running MakeMKV: {exc}"
)
mark_job_file_failed(
conn,
job_file_id,
"RIP_EXCEPTION",
message,
)
mark_job_failed(
conn,
job_id,
"RIP_EXCEPTION",
message,
)
print("RIP FAILED")
print(f" {message}")
return False
output_files = [
path
for path in episode_dir.glob("*.mkv")
if path.is_file()
and path.stat().st_size > 0
]
expected_file = (
episode_dir / source_file
if source_file
else None
)
if (
result.returncode != 0
or expected_file is None
or not expected_file.is_file()
):
message = (
f"MakeMKV returncode={result.returncode}; "
f"expected_file={expected_file}; "
f"valid_mkv_files={len(output_files)}"
)
mark_job_file_failed(
conn,
job_file_id,
"RIP_OUTPUT_INVALID",
message,
)
mark_job_failed(
conn,
job_id,
"RIP_OUTPUT_INVALID",
message,
)
print("RIP FAILED")
print(f" {message}")
if output_files:
print(" Actual MKV files:")
for path in output_files:
print(f" {path}")
return False
if expected_file.stat().st_size <= 0:
message = (
f"Expected output file is empty: "
f"{expected_file}"
)
mark_job_file_failed(
conn,
job_file_id,
"RIP_OUTPUT_EMPTY",
message,
)
mark_job_failed(
conn,
job_id,
"RIP_OUTPUT_EMPTY",
message,
)
print("RIP FAILED")
print(f" {message}")
return False
output_size = expected_file.stat().st_size
conn.execute(
"""
UPDATE job_files
SET
source_file = ?,
rip_file = ?,
status = 'RIPPED',
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
source_file,
str(expected_file),
now(),
job_file_id,
),
)
total_selected = selected_titles_total(
conn,
job_id,
)
ripped_after = ripped_selected_titles(
conn,
job_id,
)
all_ripped = (
total_selected > 0
and ripped_after >= total_selected
)
if all_ripped:
next_status = "RIPPED"
# Preserve the old destination field for compatibility,
# but the authoritative TV file list is now job_files.
destination = str(expected_file)
else:
next_status = "TITLE_SELECTED"
# Do not overwrite jobs.destination for an unfinished TV job.
destination = None
conn.execute(
"""
UPDATE jobs
SET
status = ?,
rip_completed_at = CASE
WHEN ? = 'RIPPED' THEN ?
ELSE rip_completed_at
END,
destination = CASE
WHEN ? = 'RIPPED' THEN ?
ELSE destination
END,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
next_status,
next_status,
now(),
next_status,
destination,
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"RIP_COMPLETED",
json.dumps({
"job_file_id": job_file_id,
"file": str(expected_file),
"size_bytes": output_size,
"files": [
str(path)
for path in output_files
],
"title_number": title_number,
"season_number": season_number,
"episode_number": episode_number,
"selected_titles_total": total_selected,
"ripped_selected_after": ripped_after,
"next_status": next_status,
}),
)
conn.commit()
print("RIP COMPLETED")
print(f" File: {expected_file}")
print(f" Size: {output_size:,} bytes")
if not all_ripped:
print(
f" Episodes ripped: "
f"{ripped_after}/{total_selected}"
)
print(
" Job remains TITLE_SELECTED "
"for the next episode."
)
else:
print(
" All selected titles for this job "
"are ripped."
)
return True
def main():
config = load_config()
if len(sys.argv) != 2:
print("Usage: ripper.py <drive>")
return 2
drive = sys.argv[1]
with sqlite3.connect(DB) as conn:
job = get_next_title(conn, drive)
if not job:
print(
"No selected titles available "
"for ripping."
)
return 0
try:
return 0 if rip_title(
conn,
config,
job,
) else 1
except Exception as exc:
job_id = job[0]
message = (
f"Unexpected ripper error "
f"for Job {job_id}: {exc}"
)
conn.execute(
"""
UPDATE jobs
SET
status = 'RIP_FAILED',
error_code = 'RIP_UNEXPECTED',
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
message,
now(),
job_id,
),
)
log_event(
conn,
job_id,
"ERROR",
"RIP_FAILED",
message,
)
conn.commit()
print("RIP FAILED")
print(f" {message}")
return 1
if __name__ == "__main__":
sys.exit(main())scripts/transcoder.py
#!/usr/bin/env python3
import configparser
import json
import sqlite3
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def log_event(conn, job_id, level, event, message):
conn.execute(
"""
INSERT INTO events(
job_id, timestamp, level, event, message
)
VALUES (?, ?, ?, ?, ?)
""",
(job_id, now(), level, event, message),
)
def get_next_movie_job(conn):
return conn.execute(
"""
SELECT id, disc_label, destination
FROM jobs
WHERE status = 'RIPPED'
AND COALESCE(media_type, '') != 'TV'
AND destination IS NOT NULL
ORDER BY id
LIMIT 1
"""
).fetchone()
def get_next_tv_file(conn):
return conn.execute(
"""
SELECT
jf.id,
jf.job_id,
jf.rip_file,
jf.season_number,
jf.episode_number,
j.disc_label
FROM job_files jf
JOIN jobs j ON j.id = jf.job_id
WHERE j.status = 'RIPPED'
AND j.media_type = 'TV'
AND jf.status = 'RIPPED'
AND jf.rip_file IS NOT NULL
ORDER BY
jf.job_id,
jf.season_number,
jf.episode_number,
jf.id
LIMIT 1
"""
).fetchone()
def fail_job(conn, job_id, error_code, message):
conn.execute(
"""
UPDATE jobs
SET
status = 'TRANSCODE_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(error_code, message, now(), job_id),
)
log_event(
conn,
job_id,
"ERROR",
"TRANSCODE_FAILED",
message,
)
conn.commit()
def fail_job_file(
conn,
job_file_id,
error_code,
message,
):
conn.execute(
"""
UPDATE job_files
SET
status = 'TRANSCODE_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
error_code,
message,
now(),
job_file_id,
),
)
def all_tv_files_transcoded(conn, job_id):
row = conn.execute(
"""
SELECT COUNT(*)
FROM job_files
WHERE job_id = ?
AND status != 'TRANSCODED'
""",
(job_id,),
).fetchone()
return row[0] == 0
def build_command(source, output_file):
return [
"HandBrakeCLI",
"--input",
str(source),
"--output",
str(output_file),
"--encoder",
"nvenc_h264",
"--quality",
"15",
"--rate",
"auto",
"--crop-mode",
"none",
"--custom-anamorphic",
"--keep-display-aspect",
"--all-audio",
"--all-subtitles",
"--format",
"av_mp4",
"--optimize",
]
def run_transcode(
config,
job_id,
source,
output_file,
label,
):
log_root = Path(config["state"]["logs"])
log_file = (
log_root
/ f"job-{job_id}-{label}-transcode.log"
)
print(f"Transcoding Job {job_id}: {label}")
print(f" Source: {source}")
print(f" Output: {output_file}")
print(" Encoder: nvenc_h264")
print(f" Log: {log_file}")
print()
command = build_command(
source,
output_file,
)
try:
with log_file.open("w") as log:
result = subprocess.run(
command,
stdout=log,
stderr=subprocess.STDOUT,
text=True,
check=False,
)
except Exception as exc:
return (
False,
f"Exception while running HandBrakeCLI: {exc}",
)
if result.returncode != 0:
return (
False,
(
f"HandBrakeCLI returncode="
f"{result.returncode}; "
f"output={output_file}"
),
)
if not output_file.is_file():
return (
False,
(
"HandBrake completed but output "
f"is missing: {output_file}"
),
)
if output_file.stat().st_size <= 0:
return (
False,
(
"HandBrake produced an empty output: "
f"{output_file}"
),
)
return True, None
def transcode_movie(conn, config, job):
job_id, disc_label, destination = job
source = Path(destination)
transcode_root = Path(
config["transcoding"].get(
"transcode_root",
"/rips/transcoding",
)
)
output_dir = (
transcode_root
/ "output"
/ f"job-{job_id}"
)
output_dir.mkdir(
parents=True,
exist_ok=True,
)
if not source.is_file():
message = (
f"Source MKV does not exist: {source}"
)
fail_job(
conn,
job_id,
"TRANSCODE_SOURCE_MISSING",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
if source.stat().st_size <= 0:
message = (
f"Source MKV is empty: {source}"
)
fail_job(
conn,
job_id,
"TRANSCODE_SOURCE_EMPTY",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
output_file = (
output_dir
/ f"{source.stem}.mp4"
)
if output_file.exists():
output_file.unlink()
conn.execute(
"""
UPDATE jobs
SET
status = 'TRANSCODING',
transcode_started_at = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
now(),
job_id,
),
)
conn.commit()
success, message = run_transcode(
config,
job_id,
source,
output_file,
"movie",
)
if not success:
fail_job(
conn,
job_id,
"TRANSCODE_PROCESS_FAILED",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
output_size = output_file.stat().st_size
conn.execute(
"""
UPDATE jobs
SET
status = 'TRANSCODED',
transcode_completed_at = ?,
destination = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
str(output_file),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"TRANSCODE_COMPLETED",
json.dumps({
"source": str(source),
"output": str(output_file),
"size_bytes": output_size,
}),
)
conn.commit()
print("TRANSCODE COMPLETED")
print(f" File: {output_file}")
print(f" Size: {output_size:,} bytes")
return True
def transcode_tv_file(conn, config, row):
(
job_file_id,
job_id,
rip_file,
season_number,
episode_number,
disc_label,
) = row
source = Path(rip_file)
transcode_root = Path(
config["transcoding"].get(
"transcode_root",
"/rips/transcoding",
)
)
episode_label = (
f"S{season_number:02d}"
f"E{episode_number:02d}"
)
output_dir = (
transcode_root
/ "output"
/ f"job-{job_id}"
/ episode_label
)
output_dir.mkdir(
parents=True,
exist_ok=True,
)
output_file = (
output_dir
/ f"{source.stem}.mp4"
)
if not source.is_file():
message = (
f"Source MKV does not exist: {source}"
)
fail_job_file(
conn,
job_file_id,
"TRANSCODE_SOURCE_MISSING",
message,
)
fail_job(
conn,
job_id,
"TRANSCODE_SOURCE_MISSING",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
if source.stat().st_size <= 0:
message = (
f"Source MKV is empty: {source}"
)
fail_job_file(
conn,
job_file_id,
"TRANSCODE_SOURCE_EMPTY",
message,
)
fail_job(
conn,
job_id,
"TRANSCODE_SOURCE_EMPTY",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
if (
output_file.exists()
and output_file.stat().st_size > 0
):
conn.execute(
"""
UPDATE job_files
SET
transcode_file = ?,
status = 'TRANSCODED',
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_file_id,
),
)
if all_tv_files_transcoded(
conn,
job_id,
):
conn.execute(
"""
UPDATE jobs
SET
status = 'TRANSCODED',
transcode_completed_at = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
now(),
job_id,
),
)
conn.commit()
print("Existing transcode reused")
print(f" File: {output_file}")
return True
if output_file.exists():
output_file.unlink()
conn.execute(
"""
UPDATE job_files
SET
status = 'TRANSCODING',
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
job_file_id,
),
)
conn.execute(
"""
UPDATE jobs
SET
status = 'RIPPED',
transcode_started_at =
COALESCE(
transcode_started_at,
?
),
updated_at = ?
WHERE id = ?
""",
(
now(),
now(),
job_id,
),
)
conn.commit()
success, message = run_transcode(
config,
job_id,
source,
output_file,
episode_label,
)
if not success:
fail_job_file(
conn,
job_file_id,
"TRANSCODE_PROCESS_FAILED",
message,
)
fail_job(
conn,
job_id,
"TRANSCODE_PROCESS_FAILED",
message,
)
print(f"TRANSCODE FAILED\n {message}")
return False
output_size = output_file.stat().st_size
conn.execute(
"""
UPDATE job_files
SET
transcode_file = ?,
status = 'TRANSCODED',
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_file_id,
),
)
log_event(
conn,
job_id,
"INFO",
"TV_EPISODE_TRANSCODE_COMPLETED",
json.dumps({
"job_file_id": job_file_id,
"season_number": season_number,
"episode_number": episode_number,
"source": str(source),
"output": str(output_file),
"size_bytes": output_size,
}),
)
if all_tv_files_transcoded(
conn,
job_id,
):
conn.execute(
"""
UPDATE jobs
SET
status = 'TRANSCODED',
transcode_completed_at = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
now(),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"TRANSCODE_COMPLETED",
"All TV episodes transcoded.",
)
conn.commit()
print("TV EPISODE TRANSCODE COMPLETED")
print(f" File: {output_file}")
print(f" Size: {output_size:,} bytes")
return True
def main():
config = load_config()
with sqlite3.connect(DB) as conn:
tv_file = get_next_tv_file(conn)
if tv_file:
return (
0
if transcode_tv_file(
conn,
config,
tv_file,
)
else 1
)
movie_job = get_next_movie_job(conn)
if movie_job:
return (
0
if transcode_movie(
conn,
config,
movie_job,
)
else 1
)
print("No RIPPED jobs available.")
return 0
if __name__ == "__main__":
sys.exit(main())scripts/library_mover.py
#!/usr/bin/env python3
import configparser
import hashlib
import re
import sqlite3
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG = Path("/opt/arm/config/arm.conf")
DB = Path("/opt/arm/state/arm.db")
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def load_config():
config = configparser.ConfigParser()
config.read(CONFIG)
return config
def log_event(conn, job_id, level, event, message):
conn.execute(
"""
INSERT INTO events(job_id, timestamp, level, event, message)
VALUES (?, ?, ?, ?, ?)
""",
(job_id, now(), level, event, message),
)
def get_next_job(conn):
"""
Return the next completed job requiring library movement.
TV jobs do not use jobs.destination because each TV episode
has its own destination in job_files.transcode_file.
"""
return conn.execute(
"""
SELECT
id,
media_type,
disc_label,
title,
year,
destination
FROM jobs
WHERE status = 'TRANSCODED'
ORDER BY id
LIMIT 1
"""
).fetchone()
def clean_name(value):
if not value:
return ""
value = value.strip()
# Remove common MakeMKV/disc suffixes.
value = re.sub(
r"\s+DISC\s+\d+\b",
"",
value,
flags=re.IGNORECASE,
)
value = re.sub(
r"[-_ ]+[A-Z]\d+_t\d+\b",
"",
value,
flags=re.IGNORECASE,
)
value = re.sub(
r"[-_ ]+t\d+\b",
"",
value,
flags=re.IGNORECASE,
)
# Collapse whitespace.
value = re.sub(r"\s+", " ", value)
# Filesystem-safe characters.
value = re.sub(r'[<>:"/\\|?*]', "", value)
# Avoid trailing dots/spaces.
value = value.strip(" .")
return value
def movie_name(disc_label, title, year):
name = clean_name(title) if title else clean_name(disc_label)
if not name:
name = "Unknown Movie"
if year:
return f"{name} ({int(year)})"
return name
def tv_show_name(disc_label, title, year):
"""
Determine the canonical Jellyfin TV series directory name.
Supports numeric and spelled-out season/disc markers.
"""
if title:
name = clean_name(title)
else:
name = clean_name(disc_label)
name = name.replace("_", " ")
number_word = (
r"(?:ONE|TWO|THREE|FOUR|FIVE|SIX|SEVEN|EIGHT|NINE|TEN|"
r"ELEVEN|TWELVE|THIRTEEN|FOURTEEN|FIFTEEN|SIXTEEN|"
r"SEVENTEEN|EIGHTEEN|NINETEEN|TWENTY)"
)
name = re.sub(
rf"\bDISC\s*(?:\d+|{number_word})\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(
rf"\bSEASON\s*(?:\d+|{number_word})\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(
r"\bS\d{1,2}\b",
"",
name,
flags=re.IGNORECASE,
)
name = re.sub(r"\s+", " ", name).strip()
name = name.title().replace(" Of ", " of ")
name = re.sub(r"\bOf\b", "of", name)
if not name:
name = "Unknown TV Show"
return name
def tv_episode_name(show_name, season_number, episode_number):
return (
f"{show_name} - "
f"S{int(season_number):02d}"
f"E{int(episode_number):02d}.mp4"
)
def fail_job(conn, job_id, error_code, message):
conn.execute(
"""
UPDATE jobs
SET
status = 'LIBRARY_MOVE_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
error_code,
message,
now(),
job_id,
),
)
log_event(
conn,
job_id,
"ERROR",
"LIBRARY_MOVE_FAILED",
message,
)
conn.commit()
def fail_job_file(
conn,
job_file_id,
error_code,
message,
):
conn.execute(
"""
UPDATE job_files
SET
status = 'LIBRARY_MOVE_FAILED',
error_code = ?,
error_message = ?,
updated_at = ?
WHERE id = ?
""",
(
error_code,
message,
now(),
job_file_id,
),
)
def sha256_file(path, chunk_size=8 * 1024 * 1024):
"""
Return the SHA-256 digest of a file without loading the
entire file into memory.
"""
digest = hashlib.sha256()
with Path(path).open("rb") as handle:
while True:
chunk = handle.read(chunk_size)
if not chunk:
break
digest.update(chunk)
return digest.hexdigest()
def validate_source(source):
if not source.is_file():
return (
False,
f"Transcoded source does not exist: {source}",
)
if source.stat().st_size <= 0:
return (
False,
f"Transcoded source is empty: {source}",
)
return True, None
def move_tv_job(conn, config, job):
"""
Move every transcoded TV episode belonging to a job.
Each job_file represents exactly one episode.
The job is only marked LIBRARY_MOVED after all files
have successfully reached the Jellyfin library.
"""
(
job_id,
media_type,
disc_label,
title,
year,
destination,
) = job
media_root = Path(
config["jellyfin"]["media_root"]
)
tv_root = media_root / "TV Shows"
files = conn.execute(
"""
SELECT
id,
season_number,
episode_number,
transcode_file,
library_file,
status
FROM job_files
WHERE job_id = ?
ORDER BY
season_number,
episode_number,
id
""",
(job_id,),
).fetchall()
if not files:
message = (
f"TV job {job_id} has no job_files."
)
fail_job(
conn,
job_id,
"NO_TV_FILES",
message,
)
print(f"LIBRARY MOVE FAILED\n {message}")
return False
show_name = tv_show_name(
disc_label,
title,
year,
)
print(
f"Moving TV Job {job_id}: "
f"{show_name}"
)
print(
f" Library root: {tv_root}"
)
moved_count = 0
for file_row in files:
(
job_file_id,
season_number,
episode_number,
transcode_file,
library_file,
file_status,
) = file_row
if not season_number or not episode_number:
message = (
f"Job file {job_file_id} is missing "
f"season/episode metadata."
)
fail_job_file(
conn,
job_file_id,
"TV_EPISODE_METADATA_MISSING",
message,
)
fail_job(
conn,
job_id,
"TV_EPISODE_METADATA_MISSING",
message,
)
conn.commit()
print(f"LIBRARY MOVE FAILED\n {message}")
return False
# Already completed files are left alone.
if file_status == "LIBRARY_MOVED":
if library_file:
existing_library_file = Path(library_file)
if existing_library_file.is_file():
print(
f" Already moved: "
f"S{int(season_number):02d}"
f"E{int(episode_number):02d}"
)
continue
# Database says moved but file is gone.
message = (
f"Job file {job_file_id} is marked "
f"LIBRARY_MOVED but library file is missing."
)
fail_job_file(
conn,
job_file_id,
"LIBRARY_FILE_MISSING",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_FILE_MISSING",
message,
)
conn.commit()
print(f"LIBRARY MOVE FAILED\n {message}")
return False
source = Path(transcode_file) if transcode_file else None
if source is None:
message = (
f"Job file {job_file_id} has no "
f"transcode_file."
)
fail_job_file(
conn,
job_file_id,
"TV_TRANSCODE_FILE_MISSING",
message,
)
fail_job(
conn,
job_id,
"TV_TRANSCODE_FILE_MISSING",
message,
)
conn.commit()
print(f"LIBRARY MOVE FAILED\n {message}")
return False
valid, message = validate_source(source)
if not valid:
fail_job_file(
conn,
job_file_id,
"LIBRARY_SOURCE_MISSING",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_SOURCE_MISSING",
message,
)
conn.commit()
print(f"LIBRARY MOVE FAILED\n {message}")
return False
season_dir = (
tv_root
/ show_name
/ f"Season {int(season_number):02d}"
)
output_file = (
season_dir
/ tv_episode_name(
show_name,
season_number,
episode_number,
)
)
season_dir.mkdir(
parents=True,
exist_ok=True,
)
# Never silently overwrite a library episode.
#
# If the destination already exists, determine whether this
# is a recoverable partially-recorded move. A matching size
# alone is not sufficient; SHA-256 must also match.
if output_file.exists():
existing_size = output_file.stat().st_size
source_size = source.stat().st_size
recovered = False
source_hash = None
if (
existing_size > 0
and source_size > 0
and existing_size == source_size
):
print(
" Existing destination has matching size; "
"verifying SHA-256..."
)
source_hash = sha256_file(source)
destination_hash = sha256_file(
output_file
)
if source_hash == destination_hash:
recovered = True
if recovered:
print(
f" Recovered existing library episode: "
f"S{int(season_number):02d}"
f"E{int(episode_number):02d}"
)
conn.execute(
"""
UPDATE job_files
SET
status = 'LIBRARY_MOVED',
library_file = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_file_id,
),
)
log_event(
conn,
job_id,
"INFO",
"TV_EPISODE_LIBRARY_RECONCILED",
(
f"job_file_id={job_file_id}; "
f"season={season_number}; "
f"episode={episode_number}; "
f"source={source}; "
f"destination={output_file}; "
f"size_bytes={existing_size}; "
f"sha256={source_hash}"
),
)
conn.commit()
moved_count += 1
continue
message = (
"TV episode destination already exists and "
f"does not match source: {output_file}"
)
fail_job_file(
conn,
job_file_id,
"LIBRARY_DESTINATION_EXISTS",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_EXISTS",
message,
)
conn.commit()
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
print(
f" S{int(season_number):02d}"
f"E{int(episode_number):02d}"
)
print(f" Source: {source}")
print(f" Destination: {output_file}")
try:
shutil.move(
str(source),
str(output_file),
)
except Exception as exc:
message = (
f"Move failed for job_file "
f"{job_file_id}: {exc}"
)
fail_job_file(
conn,
job_file_id,
"LIBRARY_MOVE_EXCEPTION",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_MOVE_EXCEPTION",
message,
)
conn.commit()
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
if not output_file.is_file():
message = (
f"Move reported success but destination "
f"is missing: {output_file}"
)
fail_job_file(
conn,
job_file_id,
"LIBRARY_DESTINATION_MISSING",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_MISSING",
message,
)
conn.commit()
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
output_size = output_file.stat().st_size
if output_size <= 0:
message = (
f"Destination file is empty: "
f"{output_file}"
)
fail_job_file(
conn,
job_file_id,
"LIBRARY_DESTINATION_EMPTY",
message,
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_EMPTY",
message,
)
conn.commit()
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
conn.execute(
"""
UPDATE job_files
SET
status = 'LIBRARY_MOVED',
library_file = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_file_id,
),
)
log_event(
conn,
job_id,
"INFO",
"TV_EPISODE_LIBRARY_MOVED",
(
f"job_file_id={job_file_id}; "
f"season={season_number}; "
f"episode={episode_number}; "
f"source={source}; "
f"destination={output_file}; "
f"size_bytes={output_size}"
),
)
conn.commit()
moved_count += 1
print(
f" Moved successfully: "
f"{output_size:,} bytes"
)
remaining = conn.execute(
"""
SELECT COUNT(*)
FROM job_files
WHERE job_id = ?
AND status != 'LIBRARY_MOVED'
""",
(job_id,),
).fetchone()[0]
if remaining != 0:
message = (
f"TV job {job_id} still has "
f"{remaining} file(s) not moved."
)
fail_job(
conn,
job_id,
"TV_FILES_REMAINING",
message,
)
print(f"LIBRARY MOVE FAILED\n {message}")
return False
conn.execute(
"""
UPDATE jobs
SET
status = 'LIBRARY_MOVED',
destination = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(tv_root / show_name),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"LIBRARY_MOVE_COMPLETED",
(
f"media_type=TV; "
f"show={show_name}; "
f"episodes_moved={moved_count}; "
f"destination={tv_root / show_name}"
),
)
conn.commit()
print()
print("TV LIBRARY MOVE COMPLETED")
print(f" Show: {show_name}")
print(f" Episodes: {moved_count}")
print(f" Library: {tv_root / show_name}")
return True
def move_movie_job(conn, config, job):
(
job_id,
media_type,
disc_label,
title,
year,
destination,
) = job
source = Path(destination)
media_root = Path(
config["jellyfin"]["media_root"]
)
movies_root = media_root / "Movies"
valid, message = validate_source(source)
if not valid:
fail_job(
conn,
job_id,
(
"LIBRARY_SOURCE_MISSING"
if not source.is_file()
else "LIBRARY_SOURCE_EMPTY"
),
message,
)
print(f"LIBRARY MOVE FAILED\n {message}")
return False
name = movie_name(
disc_label,
title,
year,
)
movie_dir = movies_root / name
output_file = movie_dir / f"{name}.mp4"
movies_root.mkdir(
parents=True,
exist_ok=True,
)
movie_dir.mkdir(
parents=True,
exist_ok=True,
)
# Never silently overwrite an existing movie.
#
# If the destination already exists, determine whether a
# previous filesystem move completed but the database update
# did not persist. Recovery requires both matching size and
# matching SHA-256.
if output_file.exists():
existing_size = output_file.stat().st_size
source_size = source.stat().st_size
recovered = False
source_hash = None
if (
existing_size > 0
and source_size > 0
and existing_size == source_size
):
print(
"Existing movie destination has matching size; "
"verifying SHA-256..."
)
source_hash = sha256_file(source)
destination_hash = sha256_file(
output_file
)
if source_hash == destination_hash:
recovered = True
if recovered:
print(
"Recovered existing movie library file."
)
conn.execute(
"""
UPDATE jobs
SET
status = 'LIBRARY_MOVED',
destination = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"MOVIE_LIBRARY_RECONCILED",
(
f"source={source}; "
f"destination={output_file}; "
f"size_bytes={existing_size}; "
f"sha256={source_hash}"
),
)
conn.commit()
print("MOVIE LIBRARY MOVE RECOVERED")
print(f" File: {output_file}")
return True
message = (
"Movie destination already exists and "
f"does not match source: {output_file}"
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_EXISTS",
message,
)
conn.commit()
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
print(f"Moving Movie Job {job_id}: {disc_label}")
print(f" Source: {source}")
print(f" Destination: {output_file}")
try:
shutil.move(
str(source),
str(output_file),
)
except Exception as exc:
message = f"Move failed: {exc}"
fail_job(
conn,
job_id,
"LIBRARY_MOVE_EXCEPTION",
message,
)
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
if not output_file.is_file():
message = (
f"Move reported success but destination "
f"is missing: {output_file}"
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_MISSING",
message,
)
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
output_size = output_file.stat().st_size
if output_size <= 0:
message = (
f"Destination file is empty: {output_file}"
)
fail_job(
conn,
job_id,
"LIBRARY_DESTINATION_EMPTY",
message,
)
print("LIBRARY MOVE FAILED")
print(f" {message}")
return False
conn.execute(
"""
UPDATE jobs
SET
status = 'LIBRARY_MOVED',
destination = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
str(output_file),
now(),
job_id,
),
)
log_event(
conn,
job_id,
"INFO",
"LIBRARY_MOVE_COMPLETED",
(
f"source={source}; "
f"destination={output_file}; "
f"size_bytes={output_size}"
),
)
conn.commit()
print("LIBRARY MOVE COMPLETED")
print(f" File: {output_file}")
print(f" Size: {output_size:,} bytes")
return True
def move_job(conn, config, job):
media_type = job["media_type"]
if media_type == "TV":
return move_tv_job(
conn,
config,
job,
)
return move_movie_job(
conn,
config,
job,
)
def main():
config = load_config()
with sqlite3.connect(DB) as conn:
conn.row_factory = sqlite3.Row
job = get_next_job(conn)
if not job:
print("No TRANSCODED jobs available.")
return 0
return 0 if move_job(
conn,
config,
job,
) else 1
if __name__ == "__main__":
sys.exit(main())scripts/cleanup_rips.py
#!/usr/bin/env python3
import argparse
import sqlite3
from pathlib import Path
DB = Path("/opt/arm/state/arm.db")
RIP_ROOT = Path("/rips/ripping")
TRANSCODE_ROOT = Path("/rips/transcoding/output")
def size_bytes(path):
if not path.exists():
return 0
if path.is_file():
return path.stat().st_size
return sum(
p.stat().st_size
for p in path.rglob("*")
if p.is_file()
)
def gb(value):
return value / (1024 ** 3)
def library_file_exists(path):
if not path:
return False
return Path(path).is_file()
def inspect_job(conn, job):
job_id = job["id"]
status = job["status"]
media_type = job["media_type"]
rip_dir = RIP_ROOT / f"job-{job_id}"
trans_dir = TRANSCODE_ROOT / f"job-{job_id}"
rip_size = size_bytes(rip_dir)
trans_size = size_bytes(trans_dir)
reasons = []
if status != "LIBRARY_MOVED":
reasons.append(f"status={status}")
if media_type == "TV":
files = conn.execute(
"""
SELECT
id,
status,
library_file
FROM job_files
WHERE job_id = ?
ORDER BY id
""",
(job_id,),
).fetchall()
if not files:
reasons.append("TV job has no job_files")
else:
for f in files:
if f["status"] != "LIBRARY_MOVED":
reasons.append(
f"job_file {f['id']} status={f['status']}"
)
if not library_file_exists(f["library_file"]):
reasons.append(
f"job_file {f['id']} library_file missing"
)
else:
if not job["destination"]:
reasons.append("destination missing")
elif not Path(job["destination"]).is_file():
reasons.append("destination file missing")
eligible = not reasons
return {
"id": job_id,
"disc_label": job["disc_label"],
"status": status,
"media_type": media_type or "LEGACY",
"rip_dir": rip_dir,
"trans_dir": trans_dir,
"rip_size": rip_size,
"trans_size": trans_size,
"eligible": eligible,
"reasons": reasons,
}
def main():
parser = argparse.ArgumentParser(
description="Safely clean completed ARM rip working data."
)
parser.add_argument(
"--execute",
action="store_true",
help="Actually remove eligible working directories.",
)
args = parser.parse_args()
mode = "EXECUTE" if args.execute else "DRY RUN"
print(f"=== ARM RIP CLEANUP: {mode} ===")
print()
with sqlite3.connect(DB) as conn:
conn.row_factory = sqlite3.Row
jobs = conn.execute(
"""
SELECT
id,
disc_label,
media_type,
status,
destination
FROM jobs
ORDER BY id
"""
).fetchall()
safe_total = 0
blocked_total = 0
safe_count = 0
for job in jobs:
result = inspect_job(conn, job)
total_size = (
result["rip_size"]
+ result["trans_size"]
)
if total_size == 0:
continue
if result["eligible"]:
print(
f"Job {result['id']:3d} | "
f"SAFE | "
f"{gb(total_size):6.2f} GB | "
f"{result['disc_label']}"
)
safe_total += total_size
safe_count += 1
if args.execute:
if result["rip_dir"].exists():
import shutil
shutil.rmtree(result["rip_dir"])
if result["trans_dir"].exists():
import shutil
shutil.rmtree(result["trans_dir"])
else:
print(
f"Job {result['id']:3d} | "
f"BLOCKED | "
f"{gb(total_size):6.2f} GB | "
f"{result['disc_label']}"
)
for reason in result["reasons"]:
print(f" {reason}")
blocked_total += total_size
print()
print(f"Eligible jobs: {safe_count}")
print(f"Safe to remove: {gb(safe_total):.2f} GB")
print(f"Blocked working data: {gb(blocked_total):.2f} GB")
if args.execute:
print()
print("Cleanup completed.")
else:
print()
print("DRY RUN ONLY — no files were removed.")
print("Use --execute only after reviewing the list.")
if __name__ == "__main__":
main()scripts/recovery_check.py
#!/usr/bin/env python3
import sqlite3
import subprocess
DB = "/opt/arm/state/arm.db"
def process_exists(pattern):
result = subprocess.run(
["pgrep", "-f", pattern],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
return bool(result.stdout.strip())
with sqlite3.connect(DB) as conn:
rows = conn.execute("""
SELECT id, disc_label, status
FROM jobs
WHERE status IN ('RIPPING','TRANSCODING')
""").fetchall()
if not rows:
print("No active jobs.")
exit(0)
for job_id, label, status in rows:
print()
print(f"Job {job_id}: {label}")
print(f"Status: {status}")
if status == "RIPPING":
running = process_exists("makemkvcon.*mkv")
print(f"MakeMKV running: {running}")
elif status == "TRANSCODING":
running = process_exists("HandBrakeCLI")
print(f"HandBrake running: {running}")scripts/job_corrector.py
#!/usr/bin/env python3
import configparser
import json
import shutil
import sqlite3
import sys
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from title_selector import is_tv_episode_candidate as arm_tv_candidate
DB = Path("/opt/arm/state/arm.db")
BACKUP_DIR = Path("/opt/arm/backups")
CONFIG = Path("/opt/arm/config/arm.conf")
SAFE_MUTATION_STATES = {
"LIBRARY_MOVED",
"LIBRARY_MOVE_FAILED",
"TRANSCODED",
"RIP_FAILED",
"SELECTION_FAILED",
"SUPERSEDED",
"TV_METADATA_NEEDED",
}
def now():
return datetime.now(timezone.utc).isoformat()
def backup_db(tag):
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
target = BACKUP_DIR / f"arm.db.before-{tag}-{stamp}"
shutil.copy2(DB, target)
print()
print(f"Database backup: {target}")
return target
def connect():
conn = sqlite3.connect(DB)
conn.row_factory = sqlite3.Row
return conn
def get_job(conn, job_id):
return conn.execute(
"""
SELECT *
FROM jobs
WHERE id = ?
""",
(job_id,),
).fetchone()
def get_titles(conn, job_id):
return conn.execute(
"""
SELECT *
FROM job_titles
WHERE job_id = ?
ORDER BY title_number
""",
(job_id,),
).fetchall()
def get_files(conn, job_id):
return conn.execute(
"""
SELECT *
FROM job_files
WHERE job_id = ?
ORDER BY id
""",
(job_id,),
).fetchall()
def yes_no(prompt):
answer = input(f"{prompt} [y/N]: ").strip().lower()
return answer == "y"
def print_job(conn, job_id):
job = get_job(conn, job_id)
if not job:
print(f"Job {job_id} not found.")
return False
print()
print("=" * 78)
print(f"JOB {job_id}")
print("=" * 78)
print(f"Disc label : {job['disc_label']}")
print(f"Title : {job['title']}")
print(f"Media type : {job['media_type']}")
print(f"Status : {job['status']}")
print(f"Drive : {job['drive']}")
print(f"Destination: {job['destination']}")
print(
"Fingerprint: "
f"{job['disc_fingerprint'] or '-'}"
)
if job["error_code"]:
print(f"Error : {job['error_code']}")
print(f"Message : {job['error_message']}")
print()
print("TITLES")
for row in get_titles(conn, job_id):
marker = "*" if row["selected"] else " "
season_episode = ""
if (
row["season_number"] is not None
and row["episode_number"] is not None
):
season_episode = (
f"S{int(row['season_number']):02d}"
f"E{int(row['episode_number']):02d}"
)
print(
f"{marker} "
f"title {row['title_number']:<3} "
f"{row['duration'] or '-':>10} "
f"{row['size_display'] or '-':>10} "
f"{row['chapters'] or 0:>3} ch "
f"{season_episode}"
)
files = get_files(conn, job_id)
if files:
print()
print("FILES")
for row in files:
episode = ""
if (
row["season_number"] is not None
and row["episode_number"] is not None
):
episode = (
f"S{int(row['season_number']):02d}"
f"E{int(row['episode_number']):02d}"
)
print(
f"{row['id']:<5} "
f"{episode:<8} "
f"{row['status']:<20} "
f"{row['library_file'] or row['transcode_file'] or row['rip_file'] or '-'}"
)
return True
def is_tv_episode_candidate(row):
"""
Use ARM's production TV episode-candidate rule.
Keeping correction and normal title selection identical
prevents the two workflows from drifting apart.
"""
return arm_tv_candidate(row)
def add_tv_override(prefix, show, season):
config = configparser.ConfigParser()
config.optionxform = str
with CONFIG.open() as fh:
config.read_file(fh)
if not config.has_section("tv_overrides"):
config.add_section("tv_overrides")
desired = f"{show}|{season}"
existing_key = None
for key in config["tv_overrides"]:
if key.upper() == prefix.upper():
existing_key = key
break
if existing_key:
existing = config["tv_overrides"][existing_key]
if existing == desired:
print(
f"Override already present: "
f"{existing_key} = {existing}"
)
return
raise RuntimeError(
f"Override prefix already exists:\n"
f"{existing_key} = {existing}\n"
f"Refusing to replace it automatically."
)
config["tv_overrides"][prefix] = desired
backup = BACKUP_DIR / (
"arm.conf.before-job-correction-"
+ datetime.now().strftime("%Y%m%d-%H%M%S")
)
shutil.copy2(CONFIG, backup)
with CONFIG.open("w") as fh:
config.write(fh)
print(f"Config backup: {backup}")
print(f"Added override: {prefix} = {desired}")
def resolve_tv_metadata(conn, job_id):
"""
Resolve a TV job whose show identity is known but whose season
could not be determined safely from the physical disc label.
This is intentionally pre-rip only. Once released, the normal
TV title selector determines the episode titles and numbering.
"""
job = get_job(conn, job_id)
if not job:
raise RuntimeError("Job not found.")
if job["media_type"] != "TV":
raise RuntimeError(
f"Job {job_id} is {job['media_type']}, not TV."
)
if job["status"] != "TV_METADATA_NEEDED":
raise RuntimeError(
f"Job {job_id} is {job['status']}.\n"
"This operation requires TV_METADATA_NEEDED."
)
files = get_files(conn, job_id)
if files:
raise RuntimeError(
"This metadata-needed job already has job_files.\n"
"Refusing to release a job that may have started ripping."
)
titles = get_titles(conn, job_id)
if not titles:
raise RuntimeError(
"No inspected titles are available for this job."
)
current_show = (
(job["title"] or "").strip()
or (job["disc_label"] or "").strip()
or "Unknown TV Show"
)
candidates = [
row
for row in titles
if arm_tv_candidate(row)
]
print()
print("TV METADATA REQUIRED")
print("-" * 78)
print(f"Job : {job_id}")
print(f"Disc label : {job['disc_label']}")
print(f"Show : {current_show}")
print("Season : UNKNOWN")
print()
print("LIKELY TV EPISODE TITLES")
print("-" * 78)
if candidates:
for row in candidates:
print(
f"title {row['title_number']:<3} "
f"{row['duration']:<10} "
f"{row['size_display']:<10} "
f"{row['chapters']} chapters"
)
else:
print("No strong episode-sized titles detected.")
show = input(
f"\nShow name [{current_show}]: "
).strip()
if not show:
show = current_show
season_text = input(
"Season number: "
).strip()
try:
season = int(season_text)
except ValueError:
print("Cancelled: invalid season.")
return
if not 1 <= season <= 99:
print("Cancelled: season must be 1-99.")
return
print()
print("RELEASE PREVIEW")
print("-" * 78)
print(f"Job : {job_id}")
print(f"Show : {show}")
print(f"Season : {season:02d}")
print(f"Candidates : {len(candidates)}")
print()
print(
"Episode numbering will be determined automatically "
"by the normal TV selector."
)
if not yes_no(
"\nApply TV metadata and release this job"
):
print("Cancelled.")
return
backup_db(
f"tv-metadata-job-{job_id}"
)
try:
conn.execute("BEGIN IMMEDIATE")
# Reset all title selection state. The normal selector owns
# the final episode-title decision.
conn.execute(
"""
UPDATE job_titles
SET
selected = 0,
season_number = ?,
episode_number = NULL
WHERE job_id = ?
""",
(
season,
job_id,
),
)
conn.execute(
"""
UPDATE jobs
SET
title = ?,
status = 'INSPECTED',
destination = NULL,
error_code = NULL,
error_message = NULL,
notes = COALESCE(notes, '') ||
CASE
WHEN COALESCE(notes, '') = ''
THEN ''
ELSE '; '
END ||
?,
updated_at = ?
WHERE id = ?
""",
(
show,
(
"TV metadata confirmed; "
f"show={show}; season={season}; "
"released to normal selector"
),
now(),
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(
job_id,
now(),
"INFO",
"TV_METADATA_CONFIRMED",
(
f"show={show}; season={season}; "
"status=INSPECTED"
),
),
)
conn.commit()
except Exception:
conn.rollback()
raise
print()
print("TV metadata saved.")
print(
f"Job {job_id} released as "
f"{show} Season {season:02d}."
)
print(
"ARM will run the normal TV title selector "
"on its next cycle."
)
def movie_to_tv(conn, job_id):
job = get_job(conn, job_id)
if not job:
raise RuntimeError("Job not found.")
if job["media_type"] != "MOVIE":
raise RuntimeError(
f"Job {job_id} is {job['media_type']}, not MOVIE."
)
titles = get_titles(conn, job_id)
candidates = [
row
for row in titles
if is_tv_episode_candidate(row)
]
print()
print("TV EPISODE CANDIDATES")
print("-" * 78)
if not candidates:
print("No strong TV episode candidates found.")
return
for row in candidates:
print(
f"title {row['title_number']:<3} "
f"{row['duration']:<10} "
f"{row['size_display']:<10} "
f"{row['chapters']} chapters"
)
selected_movie = [
row
for row in titles
if row["selected"] == 1
]
print()
print("CURRENT MOVIE SELECTION")
print("-" * 78)
for row in selected_movie:
print(
f"title {row['title_number']} "
f"{row['duration']} "
f"{row['size_display']}"
)
show = input("\nShow name: ").strip()
if not show:
print("Cancelled: show name is required.")
return
season_text = input("Season number: ").strip()
try:
season = int(season_text)
except ValueError:
print("Cancelled: invalid season.")
return
if not 1 <= season <= 99:
print("Cancelled: season must be 1-99.")
return
default_prefix = job["disc_label"] or ""
print()
print(
"Override prefix controls which future disc labels "
"receive this classification."
)
print(
f"Press ENTER to use the full current label: "
f"{default_prefix}"
)
prefix = input("Override prefix: ").strip()
if not prefix:
prefix = default_prefix
print()
print("CORRECTION PREVIEW")
print("-" * 78)
print(f"Job : {job_id}")
print(f"Current type : MOVIE")
print(f"Correct type : TV")
print(f"Show : {show}")
print(f"Season : {season}")
print(f"Override : {prefix} = {show}|{season}")
print(f"Episode titles: {len(candidates)}")
if job["status"] == "LIBRARY_MOVED":
print()
print(
"The old movie has already reached Jellyfin."
)
print(
"Individual TV episodes were not ripped by this job."
)
print(
"A rerip is required. This operation WILL NOT delete "
"the existing movie yet."
)
if not yes_no("\nSave this TV classification override"):
print("Cancelled.")
return
add_tv_override(prefix, show, season)
conn.execute(
"""
UPDATE jobs
SET
notes = COALESCE(notes, '') ||
CASE
WHEN COALESCE(notes, '') = ''
THEN ''
ELSE '; '
END ||
?,
updated_at = ?
WHERE id = ?
""",
(
(
f"Correction prepared: rerip as TV; "
f"show={show}; season={season}; "
f"override_prefix={prefix}"
),
now(),
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(
job_id,
now(),
"INFO",
"JOB_CORRECTION_PREPARED",
(
f"MOVIE->TV; show={show}; "
f"season={season}; prefix={prefix}"
),
),
)
conn.commit()
print()
print("Correction prepared.")
print(
"Reinsert the disc. ARM should now classify it as TV."
)
print(
"After the corrected TV job succeeds, use Delete on "
f"old Job {job_id} to remove its obsolete movie."
)
def remove_path(path):
if not path:
return
p = Path(path)
if not p.exists():
return
if p.is_dir():
shutil.rmtree(p)
else:
p.unlink()
def delete_job(conn, job_id):
job = get_job(conn, job_id)
if not job:
raise RuntimeError("Job not found.")
if job["status"] not in SAFE_MUTATION_STATES:
raise RuntimeError(
f"Job {job_id} is {job['status']}.\n"
"Refusing to delete a job that may still be active."
)
files = get_files(conn, job_id)
print()
print("DELETE PREVIEW")
print("-" * 78)
owned_paths = set()
for row in files:
for column in (
"rip_file",
"transcode_file",
"library_file",
):
value = row[column]
if value:
owned_paths.add(value)
# Movie jobs often store the library result only in jobs.destination.
destination = job["destination"]
if (
job["status"] == "LIBRARY_MOVED"
and destination
and job["media_type"] == "MOVIE"
):
owned_paths.add(destination)
for value in sorted(owned_paths):
print(value)
if job["media_type"] == "MOVIE" and destination:
destination_path = Path(destination)
if destination_path.suffix:
parent = destination_path.parent
if parent.name == destination_path.stem:
owned_paths.add(str(parent))
print()
print(
f"Database rows for Job {job_id} will also be removed."
)
if not yes_no("Delete this job and its owned media"):
print("Cancelled.")
return
backup_db(f"delete-job-{job_id}")
# Delete the most-specific paths first.
for value in sorted(
owned_paths,
key=lambda item: len(Path(item).parts),
reverse=True,
):
p = Path(value)
if p.exists():
print(f"Removing: {p}")
remove_path(p)
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"DELETE FROM job_files WHERE job_id = ?",
(job_id,),
)
conn.execute(
"DELETE FROM job_titles WHERE job_id = ?",
(job_id,),
)
conn.execute(
"DELETE FROM events WHERE job_id = ?",
(job_id,),
)
conn.execute(
"DELETE FROM jobs WHERE id = ?",
(job_id,),
)
conn.commit()
print(f"Job {job_id} removed.")
def renumber_tv(conn, job_id):
job = get_job(conn, job_id)
if not job:
raise RuntimeError("Job not found.")
if job["media_type"] != "TV":
raise RuntimeError(
f"Job {job_id} is not a TV job."
)
if job["status"] not in SAFE_MUTATION_STATES:
raise RuntimeError(
f"Job {job_id} is {job['status']}.\n"
"Refusing to renumber a job that may still be active."
)
selected = conn.execute(
"""
SELECT *
FROM job_titles
WHERE job_id = ?
AND selected = 1
ORDER BY title_number
""",
(job_id,),
).fetchall()
if not selected:
raise RuntimeError(
"No selected TV titles exist for this job."
)
season = selected[0]["season_number"]
if season is None:
raise RuntimeError(
"Selected titles do not have a season number."
)
print()
print("CURRENT EPISODE MAP")
print("-" * 78)
for row in selected:
print(
f"title {row['title_number']:<3} "
f"S{int(season):02d}"
f"E{int(row['episode_number']):02d} "
f"{row['duration']}"
)
text = input(
"\nNew starting episode number: "
).strip()
try:
start = int(text)
except ValueError:
print("Cancelled: invalid episode number.")
return
if start < 1:
print("Cancelled.")
return
mapping = {}
for offset, row in enumerate(selected):
mapping[row["episode_number"]] = start + offset
print()
print("NEW MAP")
print("-" * 78)
for old_ep, new_ep in mapping.items():
print(
f"S{int(season):02d}E{int(old_ep):02d}"
f" -> "
f"S{int(season):02d}E{int(new_ep):02d}"
)
if not yes_no("\nApply this renumbering"):
print("Cancelled.")
return
# Initial version deliberately handles library-complete jobs only
# when each episode has a unique library_file.
files = get_files(conn, job_id)
by_episode = {
row["episode_number"]: row
for row in files
if row["episode_number"] is not None
}
for old_ep in mapping:
if old_ep not in by_episode:
raise RuntimeError(
f"No job_file for E{int(old_ep):02d}."
)
backup_db(f"renumber-job-{job_id}")
# Stage library-file renames to collision-proof temporary names.
staged = []
for old_ep, new_ep in mapping.items():
row = by_episode[old_ep]
library_file = row["library_file"]
if not library_file:
continue
src = Path(library_file)
if not src.exists():
raise RuntimeError(
f"Library file missing: {src}"
)
suffix = src.suffix
final = (
src.parent
/ (
f"{src.name.split(' - S')[0]} - "
f"S{int(season):02d}"
f"E{int(new_ep):02d}{suffix}"
)
)
temp = (
src.parent
/ f".arm-renumber-job-{job_id}-{row['id']}{suffix}"
)
if temp.exists():
raise RuntimeError(
f"Temporary path already exists: {temp}"
)
src.rename(temp)
staged.append(
(row, old_ep, new_ep, temp, final)
)
# Refuse final collisions before committing.
for _, _, _, _, final in staged:
if final.exists():
# Put staged files back before aborting.
for row, old_ep, _, temp, _ in staged:
original = Path(row["library_file"])
if temp.exists():
temp.rename(original)
raise RuntimeError(
f"Destination already exists: {final}"
)
try:
conn.execute("BEGIN IMMEDIATE")
# Update titles by stable title ID, never by old episode number.
for offset, title_row in enumerate(selected):
new_ep = start + offset
conn.execute(
"""
UPDATE job_titles
SET episode_number = ?
WHERE id = ?
""",
(
new_ep,
title_row["id"],
),
)
for row, _, new_ep, temp, final in staged:
temp.rename(final)
conn.execute(
"""
UPDATE job_files
SET
episode_number = ?,
library_file = ?,
updated_at = ?
WHERE id = ?
""",
(
new_ep,
str(final),
now(),
row["id"],
),
)
# Non-library rows still need their episode number corrected.
staged_ids = {
item[0]["id"]
for item in staged
}
ordered_files = sorted(
[
row
for row in files
if row["episode_number"] is not None
],
key=lambda row: row["episode_number"],
)
for offset, row in enumerate(ordered_files):
if row["id"] in staged_ids:
continue
conn.execute(
"""
UPDATE job_files
SET
episode_number = ?,
updated_at = ?
WHERE id = ?
""",
(
start + offset,
now(),
row["id"],
),
)
conn.execute(
"""
UPDATE jobs
SET updated_at = ?
WHERE id = ?
""",
(
now(),
job_id,
),
)
conn.commit()
except Exception:
conn.rollback()
raise
print()
print(f"Job {job_id} renumbered.")
def jellyfin_refresh():
"""
Ask Jellyfin to rescan its libraries.
Returns True on success and False on failure. A refresh failure
does not roll back an otherwise successful ARM correction.
"""
config = configparser.ConfigParser()
config.read(CONFIG)
if not config.has_section("jellyfin"):
print("Jellyfin refresh skipped: no [jellyfin] config.")
return False
url = config["jellyfin"].get("url", "").strip().rstrip("/")
key = config["jellyfin"].get("api_key", "").strip()
if not url or not key:
print(
"Jellyfin refresh skipped: url/api_key not configured."
)
return False
request = urllib.request.Request(
url + "/Library/Refresh",
method="POST",
headers={
"X-Emby-Token": key,
"Accept": "application/json",
},
)
try:
with urllib.request.urlopen(
request,
timeout=10,
) as response:
response.read()
except Exception as exc:
print(
f"WARNING: Jellyfin refresh request failed: {exc}"
)
return False
print("Jellyfin library refresh requested.")
return True
def cleanup_old_jellyfin_tv_tree(show_dir):
"""
Remove stale Jellyfin-generated artifacts after TV media has
been moved to a corrected show path.
Known generated artifacts:
- *.nfo
- *-thumb.jpg
- *.trickplay directories
Refuses recursive removal when other real/unknown files remain.
"""
show_dir = Path(show_dir)
if not show_dir.exists():
return True
print()
print("Cleaning old Jellyfin metadata:")
print(f" {show_dir}")
# Remove trickplay trees first.
trickplay_dirs = sorted(
[
p
for p in show_dir.rglob("*.trickplay")
if p.is_dir()
],
key=lambda p: len(p.parts),
reverse=True,
)
for directory in trickplay_dirs:
print(f" remove trickplay: {directory}")
shutil.rmtree(directory)
# Remove known per-episode generated metadata.
generated_files = []
for p in show_dir.rglob("*"):
if not p.is_file():
continue
name = p.name.lower()
if name.endswith(".nfo"):
generated_files.append(p)
continue
if name.endswith("-thumb.jpg"):
generated_files.append(p)
continue
for generated in generated_files:
print(f" remove metadata: {generated}")
generated.unlink()
# Determine whether anything real/unknown remains.
remaining_files = [
p
for p in show_dir.rglob("*")
if p.is_file()
]
if remaining_files:
print()
print(
"Old show directory still contains files."
)
print(
"It will NOT be recursively removed:"
)
for remaining in remaining_files:
print(f" KEEP: {remaining}")
return False
# Remove now-empty directories from deepest to shallowest.
directories = sorted(
[
p
for p in show_dir.rglob("*")
if p.is_dir()
],
key=lambda p: len(p.parts),
reverse=True,
)
for directory in directories:
try:
directory.rmdir()
except OSError:
pass
try:
show_dir.rmdir()
print(f"Removed obsolete show directory: {show_dir}")
return True
except OSError:
print(
f"Old show directory remains non-empty: {show_dir}"
)
return False
def correct_tv_show(conn, job_id):
"""
Correct the canonical show name for a completed TV job and,
when needed, renumber that job's episodes.
Moves existing Jellyfin library files, updates jobs.title,
jobs.destination, job_titles, and job_files, and can save a
tv_overrides prefix for future discs.
"""
job = get_job(conn, job_id)
if not job:
raise RuntimeError("Job not found.")
if job["media_type"] != "TV":
raise RuntimeError(
f"Job {job_id} is not a TV job."
)
if job["status"] != "LIBRARY_MOVED":
raise RuntimeError(
f"Job {job_id} is {job['status']}.\n"
"Correct TV show currently requires LIBRARY_MOVED."
)
selected = conn.execute(
"""
SELECT *
FROM job_titles
WHERE job_id = ?
AND selected = 1
AND season_number IS NOT NULL
AND episode_number IS NOT NULL
ORDER BY title_number
""",
(job_id,),
).fetchall()
if not selected:
raise RuntimeError(
"No selected TV episode titles found."
)
seasons = {
int(row["season_number"])
for row in selected
}
if len(seasons) != 1:
raise RuntimeError(
"Selected titles contain multiple seasons."
)
season = seasons.pop()
files = get_files(conn, job_id)
episode_files = [
row
for row in files
if row["episode_number"] is not None
]
if len(episode_files) != len(selected):
raise RuntimeError(
"Selected-title count does not match job_file count."
)
# Determine the currently used show directory.
current_show = None
if job["destination"]:
destination = Path(job["destination"])
if destination.parent.name == "TV Shows":
current_show = destination.name
if not current_show:
library_rows = [
row
for row in episode_files
if row["library_file"]
]
if library_rows:
p = Path(library_rows[0]["library_file"])
# .../TV Shows/<show>/Season XX/file.mp4
if len(p.parents) >= 2:
current_show = p.parent.parent.name
current_show = current_show or job["title"] or job["disc_label"]
print()
print("CORRECT TV SHOW")
print("-" * 78)
print(f"Job : {job_id}")
print(f"Disc label : {job['disc_label']}")
print(f"Current show: {current_show}")
print(f"Season : {season:02d}")
new_show = input(
f"\nNew show name [{current_show}]: "
).strip()
if not new_show:
new_show = current_show
default_start = min(
int(row["episode_number"])
for row in selected
)
start_text = input(
f"Starting episode [{default_start}]: "
).strip()
if start_text:
try:
start_episode = int(start_text)
except ValueError:
print("Cancelled: invalid episode number.")
return
else:
start_episode = default_start
if start_episode < 1:
print("Cancelled: episode must be >= 1.")
return
# Stable mapping by title_number/job_title_id.
title_mapping = []
for offset, title_row in enumerate(selected):
new_episode = start_episode + offset
title_mapping.append(
(
title_row,
new_episode,
)
)
# Match job_files to job_titles primarily by job_title_id.
file_by_title_id = {
row["job_title_id"]: row
for row in episode_files
if row["job_title_id"] is not None
}
file_by_old_episode = {
int(row["episode_number"]): row
for row in episode_files
}
planned = []
media_root = Path("/mnt/jellyfin")
new_show_dir = media_root / "TV Shows" / new_show
new_season_dir = (
new_show_dir
/ f"Season {season:02d}"
)
print()
print("CORRECTION PREVIEW")
print("-" * 78)
for title_row, new_episode in title_mapping:
old_episode = int(title_row["episode_number"])
file_row = file_by_title_id.get(
title_row["id"]
)
if file_row is None:
file_row = file_by_old_episode.get(
old_episode
)
if file_row is None:
raise RuntimeError(
f"No job_file found for old E{old_episode:02d}."
)
if not file_row["library_file"]:
raise RuntimeError(
f"job_file {file_row['id']} has no library_file."
)
source = Path(file_row["library_file"])
if not source.exists():
raise RuntimeError(
f"Library file missing:\n{source}"
)
final = (
new_season_dir
/ (
f"{new_show} - "
f"S{season:02d}"
f"E{new_episode:02d}"
f"{source.suffix}"
)
)
planned.append(
(
title_row,
file_row,
old_episode,
new_episode,
source,
final,
)
)
print(
f"S{season:02d}E{old_episode:02d} "
f"-> "
f"S{season:02d}E{new_episode:02d}"
)
print(f" FROM: {source}")
print(f" TO: {final}")
# Refuse collisions with files not owned by this job.
source_paths = {
source.resolve()
for _, _, _, _, source, _ in planned
}
for _, _, _, _, _, final in planned:
if (
final.exists()
and final.resolve() not in source_paths
):
raise RuntimeError(
f"Destination already exists:\n{final}"
)
print()
print(f"jobs.title -> {new_show}")
print(f"jobs.destination -> {new_show_dir}")
print()
save_override = yes_no(
"Save/update a TV classification override for future discs"
)
override_prefix = None
if save_override:
default_prefix = job["disc_label"] or ""
print()
print(
"Enter the common label prefix. "
"For Example Show Season 11 use EXAMPLE_SHOW_S11_"
)
override_prefix = input(
f"Override prefix [{default_prefix}]: "
).strip()
if not override_prefix:
override_prefix = default_prefix
print(
f"Override: {override_prefix} = "
f"{new_show}|{season}"
)
if not yes_no("\nApply this correction"):
print("Cancelled.")
return
backup_db(f"correct-tv-show-job-{job_id}")
# Save config override before touching media. add_tv_override()
# performs its own config backup and refuses conflicting entries.
if save_override:
add_tv_override(
override_prefix,
new_show,
season,
)
new_season_dir.mkdir(
parents=True,
exist_ok=True,
)
# Stage every source to a unique temporary file first so
# renumbering cannot collide with itself.
staged = []
try:
for (
title_row,
file_row,
old_episode,
new_episode,
source,
final,
) in planned:
temp = (
source.parent
/ (
f".arm-correct-job-{job_id}-"
f"{file_row['id']}"
f"{source.suffix}"
)
)
if temp.exists():
raise RuntimeError(
f"Temporary file already exists:\n{temp}"
)
source.rename(temp)
staged.append(
(
title_row,
file_row,
old_episode,
new_episode,
source,
temp,
final,
)
)
# Re-check destinations after staging.
for item in staged:
final = item[-1]
if final.exists():
raise RuntimeError(
f"Destination appeared during correction:\n{final}"
)
conn.execute("BEGIN IMMEDIATE")
for (
title_row,
file_row,
old_episode,
new_episode,
source,
temp,
final,
) in staged:
final.parent.mkdir(
parents=True,
exist_ok=True,
)
temp.rename(final)
conn.execute(
"""
UPDATE job_titles
SET
season_number = ?,
episode_number = ?
WHERE id = ?
""",
(
season,
new_episode,
title_row["id"],
),
)
conn.execute(
"""
UPDATE job_files
SET
season_number = ?,
episode_number = ?,
library_file = ?,
updated_at = ?
WHERE id = ?
""",
(
season,
new_episode,
str(final),
now(),
file_row["id"],
),
)
conn.execute(
"""
UPDATE jobs
SET
title = ?,
destination = ?,
error_code = NULL,
error_message = NULL,
updated_at = ?
WHERE id = ?
""",
(
new_show,
str(new_show_dir),
now(),
job_id,
),
)
conn.execute(
"""
INSERT INTO events(
job_id,
timestamp,
level,
event,
message
)
VALUES (?, ?, ?, ?, ?)
""",
(
job_id,
now(),
"INFO",
"TV_SHOW_CORRECTED",
(
f"show={new_show}; "
f"season={season}; "
f"start_episode={start_episode}"
),
),
)
conn.commit()
except Exception:
conn.rollback()
# Best-effort restoration of any staged/moved files.
for (
_,
_,
_,
_,
original,
temp,
final,
) in reversed(staged):
try:
if final.exists():
original.parent.mkdir(
parents=True,
exist_ok=True,
)
final.rename(original)
elif temp.exists():
original.parent.mkdir(
parents=True,
exist_ok=True,
)
temp.rename(original)
except Exception:
pass
raise
# --------------------------------------------------------
# Clean the obsolete Jellyfin show identity.
# --------------------------------------------------------
old_show_dirs = set()
for _, _, _, _, source, _ in planned:
# source:
# .../TV Shows/<old show>/Season XX/file.mp4
old_show_dirs.add(source.parent.parent)
for old_show_dir in old_show_dirs:
# Never clean the canonical destination itself.
try:
same_directory = (
old_show_dir.resolve()
== new_show_dir.resolve()
)
except FileNotFoundError:
same_directory = False
if same_directory:
continue
cleanup_old_jellyfin_tv_tree(
old_show_dir
)
# Ask Jellyfin to forget stale entries and discover the
# corrected paths immediately.
jellyfin_refresh()
print()
print(
f"Job {job_id} corrected to "
f"{new_show} Season {season:02d}, "
f"starting at E{start_episode:02d}."
)
def menu(job_id):
conn = connect()
try:
if not print_job(conn, job_id):
return
print()
print("=" * 78)
print("JOB CORRECTOR")
print("=" * 78)
print()
print(" 1 Movie -> TV / prepare rerip")
print(" 2 Inspect job")
print(" 3 Renumber TV episodes")
print(" 4 Correct TV show / episode start")
print(" 5 Delete completed job + owned media")
print(" 6 Resolve TV metadata / release job")
print(" Q Cancel")
print()
choice = input("Choice: ").strip().lower()
if choice == "1":
movie_to_tv(conn, job_id)
elif choice == "2":
print_job(conn, job_id)
print()
input("Press ENTER when finished reviewing this job...")
elif choice == "3":
renumber_tv(conn, job_id)
elif choice == "4":
correct_tv_show(conn, job_id)
elif choice == "5":
delete_job(conn, job_id)
elif choice == "6":
resolve_tv_metadata(conn, job_id)
elif choice == "q":
return
else:
print("Unknown selection.")
finally:
conn.close()
def main():
delete_mode = False
if len(sys.argv) == 3 and sys.argv[1] == "--delete":
delete_mode = True
job_arg = sys.argv[2]
elif len(sys.argv) == 2:
job_arg = sys.argv[1]
else:
print(
"Usage:\n"
" sudo job_corrector.py JOB_ID\n"
" sudo job_corrector.py --delete JOB_ID"
)
raise SystemExit(2)
try:
job_id = int(job_arg)
except ValueError:
print("JOB_ID must be an integer.")
raise SystemExit(2)
try:
if delete_mode:
conn = connect()
try:
if not print_job(conn, job_id):
return
delete_job(conn, job_id)
finally:
conn.close()
else:
menu(job_id)
except KeyboardInterrupt:
print("\nCancelled.")
except Exception as exc:
print()
print(f"ERROR: {exc}")
raise SystemExit(1)
if __name__ == "__main__":
main()