#!/usr/bin/env -S uv run --script # # Copy the disc in the tray. # # A DVD is copied with dvdbackup into copydisk.dvd_output, and an empty # file named for the default Plex library is left in that directory so # disk.py can see it. An audio CD is ripped with abcde. The disc id is # recorded under copydisk.cd_output so the same disc is not ripped twice. # abcde chooses its own directory for the audio files. A data track is # reported and left on the disc; this script does not copy it. # # The drive, the mount point, and those two output roots come from # config.yaml. On success the output directory is printed as JSON unless # --quiet is given. Further fields in that object are still open. # import argparse import fcntl import json import os import subprocess import time from pathlib import Path import yaml from easydict import EasyDict as edict # noqa: N813 ROOT = Path(__file__).resolve().parent.parent CFG = edict(yaml.safe_load((ROOT / "config.yaml").read_text())) # linux/cdrom.h. The Python 2 CDROM module wrapped these same numbers. CDROM_DRIVE_STATUS = 0x5326 CDS_DISC_OK = 4 ABCDE = Path("/usr/bin/abcde") CD_DISCID = Path("/usr/bin/cd-discid") DU = Path("/usr/bin/du") DVD_BACKUP = Path("/usr/bin/dvdbackup") EJECT = Path("/usr/bin/eject") MOUNT = Path("/bin/mount") UDEVADM = Path("/usr/bin/udevadm") UMOUNT = Path("/bin/umount") # A DVD cannot hold this many kilobytes. A larger mount is a bad filesystem. # The marker file under copydisk.metadata is the memory of discs we have # seen and refused to copy. That directory can hold other notes later. TOO_BIG_KB = 10_000_000 def main(): args = parse_args() drive = Path(CFG.copydisk.drive) mount = Path(CFG.copydisk.mount) dvd_output = Path(CFG.copydisk.dvd_output) cd_output = Path(CFG.copydisk.cd_output) if not disc_present(drive): raise SystemExit(f"no disc in {drive}") props = udev_properties(drive) if is_dvd(props): result = copy_dvd(drive, mount, dvd_output, props) elif is_cd(props): result = copy_cd(drive, mount, cd_output, props) else: raise SystemExit(f"unhandled media in {drive}") if CFG.copydisk.eject and not result.get("data_tracks"): subprocess.check_call([str(EJECT), str(drive)]) if not args.quiet: print(json.dumps(result)) def parse_args(): parser = argparse.ArgumentParser(description="Copy the disc in the tray.") parser.add_argument("--quiet", action="store_true", help="do not print the result JSON") return parser.parse_args() def copy_dvd(drive, mount, dvd_output, props): category = CFG.plex.libraries[0] mount_disc(mount) try: label = props.get("ID_FS_LABEL") or "Unlabeled" title = dvd_title(drive) size_kb = int(subprocess.check_output([str(DU), "-sk", str(mount)], text=True).split()[0]) if size_kb >= TOO_BIG_KB: marker = note_too_big(dvd_output, label) raise SystemExit(f"{mount} is {size_kb} kbytes, which is past {TOO_BIG_KB}; noted {marker}") dest = dvd_output / title renamed = False if dest.exists(): dest = dvd_output / f"{title}.{os.getpid()}" renamed = True started = time.time() subprocess.check_call([ str(DVD_BACKUP), "-M", "-o", str(dvd_output), "-i", str(drive), "-n", dest.name, ]) # Empty marker file. disk.py reads this name as the Plex library. _ = (dest / category).write_text("") finally: if is_mounted(mount): subprocess.check_call([str(UMOUNT), str(mount)]) return edict( kind="dvd", output=str(dest), title=dest.name, label=label, renamed=renamed, minutes=round((time.time() - started) / 60, 1), ) def copy_cd(drive, mount, cd_output, props): if is_mounted(mount): subprocess.check_call([str(UMOUNT), str(mount)]) discid = subprocess.check_output([str(CD_DISCID), str(drive)], text=True).split()[0] seen = cd_output / "discid" / discid if seen.exists(): raise SystemExit(f"CD {discid} was already copied") n_audio = int(props.get("ID_CDROM_MEDIA_TRACK_COUNT_AUDIO") or 0) n_data = int(props.get("ID_CDROM_MEDIA_TRACK_COUNT_DATA") or 0) if n_audio == 0: raise SystemExit(f"CD {discid} has no audio tracks") started = time.time() cmd = [str(ABCDE), "-N"] cmd.append(f"1-{n_audio}") subprocess.check_call(cmd) seen.parent.mkdir(parents=True, exist_ok=True) _ = seen.write_text(f"{n_audio} audio\n{n_data} data\n") return edict( kind="cd", output=str(seen.parent), discid=discid, audio_tracks=n_audio, data_tracks=n_data, minutes=round((time.time() - started) / 60, 1), ) def disc_present(drive): fd = os.open(drive, os.O_RDONLY | os.O_NONBLOCK) try: status = fcntl.ioctl(fd, CDROM_DRIVE_STATUS) finally: os.close(fd) return status == CDS_DISC_OK def udev_properties(drive): text = subprocess.check_output( [str(UDEVADM), "info", "-q", "property", "-n", str(drive)], text=True, ) props = edict() for line in text.splitlines(): if "=" not in line: continue key, value = line.split("=", 1) props[key] = value return props def is_dvd(props): return any(props.get(name) == "1" for name in ( "ID_CDROM_DVD", "ID_CDROM_MEDIA_DVD", "ID_CDROM_MEDIA_DVD_PLUS_R", "ID_CDROM_MEDIA_DVD_PLUS_RW", )) def is_cd(props): return props.get("ID_CDROM_MEDIA_CD") == "1" or props.get("ID_CDROM_MEDIA_CD_R") == "1" def note_too_big(dvd_output, label): "Touch a marker so a bad DVD is remembered." name = time.strftime("BAD-%Y%m%d-%H%M%S") if label != "Unlabeled": name = f"{label}.{name}" marker = dvd_output / CFG.copydisk.metadata / name marker.parent.mkdir(parents=True, exist_ok=True) marker.touch() return marker def dvd_title(drive): info = subprocess.check_output([str(DVD_BACKUP), "-I", "-i", str(drive)], text=True) for line in info.splitlines(): if line.startswith("DVD-Video "): title = line.split('"')[1] if title: return title raise SystemExit(f"dvdbackup reported no title for {drive}") def mount_disc(mount): if is_mounted(mount): return subprocess.check_call([str(MOUNT), str(mount)]) def is_mounted(mount): return mount.stat().st_dev != mount.parent.stat().st_dev if __name__ == "__main__": main()