serverAppConfigs/scripts/media-census.py

#!/usr/bin/env python3
r"""Read-only census of a movie library — the inputs a relayout plan is built from.

  media-census.py enumerate [--root DIR] [--out TSV]
  media-census.py probe     [--enum TSV] [--out TSV] [--jobs N]
  media-census.py plex      [--root DIR] [--db PATH] [--out TSV]

Run it on the machine that mounts the library. Nothing here writes inside the
library and nothing is deleted; the only outputs are the three TSVs below, and
scripts/media-plan.py reads them instead of touching media.

  enum    bytes <TAB> ext <TAB> path                  every file under the root
  probed  bytes <TAB> path <TAB> dur <TAB> vcodec <TAB> WxH <TAB> fps <TAB> vbit
          <TAB> pix <TAB> audio <TAB> subs <TAB> spare <TAB> format
          one row per video-container file; `audio` is "codec:lang+codec:lang",
          `subs` the subtitle-stream count, `spare` the audio streams carrying
          neither a default nor a forced tag
  plex    file <TAB> leaf_title <TAB> … (see PLEX_COLS)
          the titles Plex already matched — the authority over any filename guess

`probe` is the slow step (every container is opened for its header, ~10 min for
4,800 files on a spindle array at 6 workers) so it is `nice`d: a census beside a
live service must not starve it. `enumerate` is a stat pass, so re-censusing
before a plan is cheap — and a plan is only ever as good as these three files.
"""
import argparse
import csv
import json
import os
import re
import sqlite3
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor

# Containers ffprobe can read. .iso/.img are deliberately NOT here: a disc image is
# a mounted volume, not a stream, and the planner treats it as a movie file instead.
PROBE_EXT = {".mkv", ".mp4", ".avi", ".m4v", ".mov", ".ts", ".m2ts", ".mpg",
             ".mpeg", ".webm", ".wmv", ".divx", ".ogm", ".bdmv", ".vob"}
# Trash and thumbnail caches are not library content, and this repo's trash policy
# means .AGENTTRASH can sit inside a movie folder at any time.
SKIP_DIR = re.compile(r"(^|/)(\.AGENTTRASH|\.thumbnails|\.Trash-\d+|@eaDir|lost\+found)(/|$)")
PLEX_DB = ("/var/lib/plexmediaserver/Library/Application Support/Plex Media Server"
           "/Plug-in Support/Databases/com.plexapp.plugins.library.db")
PLEX_COLS = ["file", "leaf_title", "leaf_orig", "leaf_edition", "leaf_year",
             "parent_title", "parent_orig", "parent_year", "parent_edition",
             "container", "vcodec", "acodec", "width", "height"]
PLEX_SQL = """
select p.file, coalesce(mi.title,''), coalesce(mi.original_title,''),
       coalesce(mi.edition_title,''), coalesce(mi.year,0),
       coalesce(par.title,''), coalesce(par.original_title,''),
       coalesce(par.year,0), coalesce(par.edition_title,''),
       m.container, m.video_codec, m.audio_codec, m.width, m.height
from media_parts p
join media_items m on m.id = p.media_item_id
join metadata_items mi on mi.id = m.metadata_item_id
left join metadata_items par on par.id = mi.parent_id
where p.file like ? and p.deleted_at is null and mi.deleted_at is null
"""


def enumerate_lib(root, out):
    """One row per real file under root: size, extension, path."""
    rows = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames
                       if not SKIP_DIR.search(os.path.join(dirpath, d))]
        for fn in filenames:
            p = os.path.join(dirpath, fn)
            if SKIP_DIR.search(p):
                continue
            try:
                st = os.lstat(p)
            except OSError:
                continue
            if not os.path.isfile(p) or os.path.islink(p):
                continue                # links and sockets are not library content
            rows.append((st.st_size, os.path.splitext(fn)[1].lower(), p))
    with open(out, "w") as fh:
        for size, ext, p in rows:
            fh.write(f"{size}\t{ext}\t{p}\n")
    vid = [r for r in rows if r[1] in PROBE_EXT]
    print(f"files={len(rows)} bytes={sum(r[0] for r in rows)} "
          f"probes={len(vid)} probe_bytes={sum(r[0] for r in vid)} -> {out}")


def probe_one(path):
    """ffprobe header read -> the probed TSV row (never the whole file)."""
    try:
        os.nice(19)                     # a census must never outrank the media server
    except OSError:
        pass
    try:
        size = os.stat(path).st_size
    except OSError:
        return None
    cmd = ["ffprobe", "-v", "error", "-print_format", "json",
           "-show_format", "-show_streams", path]
    try:
        done = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
        data = json.loads(done.stdout or "{}")
    except (subprocess.TimeoutExpired, subprocess.SubprocessError,
            json.JSONDecodeError, OSError):
        data = {}
    st = data.get("streams", [])
    fmt = data.get("format", {})
    v = next((s for s in st if s.get("codec_type") == "video"), None)
    aud = [s for s in st if s.get("codec_type") == "audio"]
    subs = [s for s in st if s.get("codec_type") == "subtitle"]
    fps = "-"
    if v and v.get("r_frame_rate"):
        try:
            n, d = (int(x) for x in v["r_frame_rate"].split("/", 1))
            fps = f"{round(n / d * 100) / 100:g}" if d else "0"
        except ValueError:
            pass

    def num(x):
        try:
            return round(float(x))
        except (TypeError, ValueError):
            return 0

    return [size, path, num(fmt.get("duration")),
            (v or {}).get("codec_name", "-"),
            f"{v['width']}x{v['height']}" if v else "-",
            fps, num(fmt.get("bit_rate")), (v or {}).get("pix_fmt", "-"),
            "+".join(f"{a.get('codec_name', '?')}:{a.get('language', '?')}" for a in aud),
            len(subs),
            sum(1 for a in aud if not re.search(
                r"default|forced", json.dumps(a.get("tags", {})), re.I)),
            fmt.get("format_name", "-") or "-"]


def probe(enum, out, jobs):
    """Probe every video-container row of the enum TSV, in parallel."""
    paths = []
    with open(enum) as fh:
        for line in fh:
            f = line.rstrip("\n").split("\t", 2)
            if len(f) == 3 and f[1] in PROBE_EXT:
                paths.append(f[2])
    print(f"probing {len(paths)} files with {jobs} workers", flush=True)
    done = unreadable = 0
    with open(out, "w") as fh, ThreadPoolExecutor(max_workers=jobs) as pool:
        for row in pool.map(probe_one, paths):
            if row is None:
                continue
            if row[3] == "-":
                unreadable += 1         # header unreadable: the row still gets written
            done += 1
            fh.write("\t".join(str(x) for x in row) + "\n")
            if done % 250 == 0:
                print(f"  {done}/{len(paths)}", flush=True)
    print(f"probed={done} unreadable={unreadable} -> {out}")


def plex(root, db, out):
    """Plex's matched titles for files under root, from its DB, read-only."""
    uri = f"file:{db}?mode=ro&immutable=1"
    try:
        con = sqlite3.connect(uri, uri=True)
    except sqlite3.Error as e:
        sys.exit(f"cannot open the Plex database read-only ({db}): {e}\n"
                 "Plex runs rootless here: run this as the container's owner, never "
                 "with sudo — rootless storage is per-user, so root sees nothing.")
    with open(out, "w", newline="") as fh:
        w = csv.writer(fh, delimiter="\t")
        w.writerow(PLEX_COLS)
        n = 0
        for r in con.execute(PLEX_SQL, (root + "/%",)):
            w.writerow([str(x).replace("\t", " ").replace("\n", " ") for x in r])
            n += 1
    print(f"plex-matched={n} -> {out}")


def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    sub = ap.add_subparsers(dest="cmd", required=True)
    root = {"default": os.environ.get("MEDIA_ROOT", "/mnt/entropy/Movies"),
            "help": "library root"}
    e = sub.add_parser("enumerate", help="stat every file under the root")
    e.add_argument("--root", **root)
    e.add_argument("--out", default="/tmp/media-enum.tsv")
    p = sub.add_parser("probe", help="ffprobe every video container")
    p.add_argument("--enum", default="/tmp/media-enum.tsv")
    p.add_argument("--out", default="/tmp/media-probed.tsv")
    p.add_argument("--jobs", type=int, default=6)
    x = sub.add_parser("plex", help="export Plex's matched titles for the root")
    x.add_argument("--root", **root)
    x.add_argument("--db", default=PLEX_DB)
    x.add_argument("--out", default="/tmp/plex-movies.tsv")
    a = ap.parse_args()
    if a.cmd == "enumerate":
        enumerate_lib(a.root, a.out)
    elif a.cmd == "probe":
        probe(a.enum, a.out, a.jobs)
    else:
        plex(a.root, a.db, a.out)


if __name__ == "__main__":
    main()