#! /usr/bin/python3 ### -*- mode: python; coding: utf-8 -*- from contextlib import contextmanager import optparse as OP import os as OS import re as RX import subprocess as SP import sys as SYS class ExpectedError (Exception): pass @contextmanager def location(loc): global LOC old, LOC = LOC, loc yield loc LOC = old def filter(value, func = None, dflt = None): if value is None: return dflt elif func is None: return value else: return func(value) def check(cond, msg): if not cond: raise ExpectedError(msg) def lookup(dict, key, msg): try: return dict[key] except KeyError: raise ExpectedError(msg) def forget(dict, key): try: del dict[key] except KeyError: pass def getint(s): if not s.isdigit(): raise ExpectedError("bad integer `%s'" % s) return int(s) class Words (object): def __init__(me, s): me._s = s me._i, me._n = 0, len(s) def _wordstart(me): s, i, n = me._s, me._i, me._n while i < n: if not s[i].isspace(): return i i += 1 return -1 def nextword(me): s, n = me._s, me._n begin = i = me._wordstart() if begin < 0: return None while i < n and not s[i].isspace(): i += 1 me._i = i return s[begin:i] def rest(me): s, n = me._s, me._n begin = me._wordstart() if begin < 0: return None else: return s[begin:].rstrip() def program_output(*args, **kw): try: return SP.check_output(*args, **kw) except SP.CalledProcessError as e: raise ExpectedError("program `%s' failed with code %d" % (e.cmd, e.returncode)) URL_SAFE_P = 256*[False] for ch in \ b"ABCDEFGHIJKLMNOPQRSTUVWXYZ" \ b"abcdefghijklmnopqrstuvwxyz" \ b"0123456789" b"!$%-.,/": URL_SAFE_P[ch] = True def urlencode(s): return "".join((URL_SAFE_P[ch] and chr(ch) or "%%%02x" % ch for ch in s.encode("UTF-8"))) PROG = OS.path.basename(SYS.argv[0]) class BaseLocation (object): def report(me, exc): SYS.stderr.write("%s: %s%s\n" % (PROG, me._loc(), exc)) class DummyLocation (BaseLocation): def _loc(me): return "" class FileLocation (BaseLocation): def __init__(me, fn, lno = 1): me._fn, me._lno = fn, lno def _loc(me): return "%s:%d: " % (me._fn, me._lno) def stepline(me): me._lno += 1 LOC = DummyLocation() class Source (object): PREFIX = "" TITLEP = CHAPTERP = False def __init__(me, fn): me.fn = fn me.neps = None me.used_titles = dict() me.used_chapters = set() me.nuses = 0 def _duration(me, title, start_chapter, end_chapter): return -1 def url_and_duration(me, title = None, start_chapter = None, end_chapter = None): if title == "-": if me.TITLEP: raise ExpectedError("missing title number") if start_chapter is not None or end_chapter is not None: raise ExpectedError("can't specify chapter without title") suffix = "" elif not me.TITLEP: raise ExpectedError("can't specify title with `%s'" % me.fn) elif start_chapter is None: if end_chapter is not None: raise ExpectedError("can't specify end chapter without start chapter") suffix = "#%d" % title elif not me.CHAPTERP: raise ExpectedError("can't specify chapter with `%s'" % me.fn) elif end_chapter is None: suffix = "#%d:%d" % (title, start_chapter) else: suffix = "#%d:%d-%d:%d" % (title, start_chapter, title, end_chapter - 1) duration = me._duration(title, start_chapter, end_chapter) if end_chapter is not None: keys = [(title, ch) for ch in range(start_chapter, end_chapter)] set = me.used_chapters else: keys, set = [title], me.used_titles for k in keys: if k in set: if title == "-": raise ExpectedError("`%s' already used" % me.fn) elif end_chapter is None: raise ExpectedError("`%s' title %d already used" % (me.fn, title)) else: raise ExpectedError("`%s' title %d chapter %d already used" % (me.fn, title, k[1])) if end_chapter is not None: for ch in range(start_chapter, end_chapter): me.used_chapters.add((title, ch)) return me.PREFIX + ROOT + urlencode(me.fn) + suffix, duration class VideoDisc (Source): PREFIX = "dvd://" TITLEP = CHAPTERP = True def __init__(me, fn, *args, **kw): super().__init__(fn, *args, **kw) me.neps = 0 def _duration(me, title, start_chapter, end_chapter): path = OS.path.join(ROOT, me.fn) ntitle = int(program_output(["dvd-info", path, "titles"])) if not 1 <= title <= ntitle: raise ExpectedError("bad title %d for `%s': must be in 1 .. %d" % (title, me.fn, ntitle)) if start_chapter is None: durq = "duration:%d" % title else: nch = int(program_output(["dvd-info", path, "chapters:%d" % title])) if end_chapter is None: end_chapter = nch else: end_chapter -= 1 if not 1 <= start_chapter <= end_chapter <= nch: raise ExpectedError("bad chapter range %d .. %d for `%s' title %d: " "must be in 1 .. %d" % (start_chapter, end_chapter, me.fn, title, nch)) durq = "duration:%d.%d-%d" % (title, start_chapter, end_chapter) duration = int(program_output(["dvd-info", path, durq])) return duration class VideoSeason (object): def __init__(me, i, title): me.i = i me.title = title me.episodes = {} def set_episode_disc(me, i, disc): if i in me.episodes: raise ExpectedError("season %d episode %d already taken" % (me.i, i)) me.episodes[i] = disc; disc.neps += 1 def match_group(m, *groups, dflt = None, mustp = False): for g in groups: try: s = m.group(g) except IndexError: continue if s is not None: return s if mustp: raise ValueError("no match found") else: return dflt class VideoDir (object): _R_ISO_PRE = list(map(lambda pats: list(map(lambda pat: RX.compile("^" + pat + r"\.iso$", RX.X), pats)), [[r""" S (?P \d+) \. \ (?P .*) — (?: D \d+ \. \ )? (?P .*) """, r""" S (?P \d+) (?: D \d+)? \. \ (?P .*) """, r""" S (?P \d+) \. \ (?P E \d+ .*) """, r""" S (?P \d+) (?P E \d+) \. \ .* """], [r""" (?P \d+) [A-Z]? \. \ (?P .*) — (?P .*) """], [r""" \d+ \. \ (?P [ES] \d+ .*) """], [r""" (?P \d+ ) \. \ .* """]])) _R_ISO_EP = RX.compile(r""" ^ (?: S (?P \d+) \ )? E (?P \d+) (?: – (?P \d+))? $ """, RX.X) def __init__(me, dir): me.dir = dir fns = OS.listdir(OS.path.join(ROOT, dir)) fns.sort() season = None seasons = {} styles = me._R_ISO_PRE for fn in fns: path = OS.path.join(dir, fn) if not fn.endswith(".iso"): continue #print(";; `%s'" % path, file = SYS.stderr) for sty in styles: for r in sty: m = r.match(fn) if m: styles = [sty]; break else: continue break else: #print(";;\tignored (regex mismatch)", file = SYS.stderr) continue si = filter(match_group(m, "si"), int) stitle = match_group(m, "stitle") check(si is not None or stitle is None, "explicit season title without number in `%s'" % fn) if si is not None: if season is None or si != season.i: check(season is None or si == season.i + 1, "season %d /= %d" % (si, season is None and -1 or season.i + 1)) check(si not in seasons, "season %d already seen" % si) seasons[si] = season = VideoSeason(si, stitle) else: check(stitle == season.title, "season title `%s' /= `%s'" % (stitle, season.title)) disc = VideoDisc(path) ts = season any, bad = False, False epnum = match_group(m, "epnum") if epnum is not None: eplist = ["E" + epnum] else: eplist = match_group(m, "epex", mustp = True).split(", ") for eprange in eplist: mm = me._R_ISO_EP.match(eprange) if mm is None: #print(";;\t`%s'?" % eprange, file = SYS.stderr) bad = True; continue if not any: any = True i = filter(mm.group("si"), int) if i is not None: try: ts = seasons[i] except KeyError: ts = seasons[i] = VideoSeason(i, None) if ts is None: ts = season = seasons[1] = VideoSeason(1, None) start = filter(mm.group("ei"), int) end = filter(mm.group("ej"), int, start) for k in range(start, end + 1): ts.set_episode_disc(k, disc) #print(";;\tepisode %d.%d" % (ts.i, k), file = SYS.stderr) if not any: #print(";;\tignored", file = SYS.stderr) pass elif bad: raise ExpectedError("bad ep list in `%s'", fn) me.seasons = seasons class AudioDisc (Source): PREFIX = "file://" TITLEP = CHAPTERP = False def _duration(me, title, start_chapter, end_chaptwr): out = program_output(["metaflac", "--show-total-samples", "--show-sample-rate", OS.path.join(ROOT, me.fn)]) nsamples, hz = map(float, out.split()) return int(nsamples/hz) class AudioEpisode (AudioDisc): def __init__(me, fn, i, *args, **kw): super().__init__(fn, *args, **kw) me.i = i class AudioDir (object): _R_FLAC = RX.compile(r""" ^ E (\d+) (?: \. \ (.*))? \. flac $ """, RX.X) def __init__(me, dir): me.dir = dir fns = OS.listdir(OS.path.join(ROOT, dir)) fns.sort() episodes = {} last_i = 0 for fn in fns: path = OS.path.join(dir, fn) if not fn.endswith(".flac"): continue m = me._R_FLAC.match(fn) if not m: continue i = filter(m.group(1), int) etitle = m.group(2) check(i == last_i + 1, "episode %d /= %d" % (i, last_i + 1)) episodes[i] = AudioEpisode(path, i) last_i = i me.episodes = episodes class Chapter (object): def __init__(me, episode, title, i): me.title, me.i = title, i me.url, me.duration = \ episode.source.url_and_duration(episode.tno, i, i + 1) class Episode (object): def __init__(me, season, i, neps, title, src, tno = None, startch = None, endch = None): me.season = season me.i, me.neps, me.title = i, neps, title me.chapters = [] me.source, me.tno = src, tno me.url, me.duration = src.url_and_duration(tno, startch, endch) def add_chapter(me, title, j): ch = Chapter(me, title, j) me.chapters.append(ch) return ch def label(me): return me.season._eplabel(me.i, me.neps, me.title) class BaseSeason (object): def __init__(me, series, implicitp = False): me.series = series me.episodes = [] me.implicitp = implicitp me.ep_i, episodes = 1, [] def add_episode(me, j, neps, title, src, tno, startch, endch): ep = Episode(me, j, neps, title, src, tno, startch, endch) me.episodes.append(ep) src.nuses += neps; me.ep_i += neps return ep def _epnames(me, i, neps): playlist = me.series.playlist if neps == 1: return playlist.epname, ["%d" % i] elif neps == 2: return playlist.epnames, ["%d" % i, "%d" % (i + 1)] else: return playlist.epnames, ["%d–%d" % (i, i + neps - 1)] class Season (BaseSeason): def __init__(me, series, title, i, *args, **kw): super().__init__(series, *args, **kw) me.title, me.i = title, i def _eplabel(me, i, neps, title): epname, epn = me._epnames(i, neps) if title is None: if me.implicitp: label = "%s %s" % (epname, ", ".join(epn)) elif me.title is None: label = "%s %s" % \ (epname, ", ".join("%d.%s" % (me.i, e) for e in epn)) else: label = "%s—%s %s" % (me.title, epname, ", ".join(epn)) else: if me.implicitp: label = "%s. %s" % (", ".join(epn), title) elif me.title is None: label = "%s. %s" % \ (", ".join("%d.%s" % (me.i, e) for e in epn), title) else: label = "%s—%s. %s" % (me.title, ", ".join(epn), title) return label class MovieSeason (BaseSeason): def __init__(me, series, title, *args, **kw): super().__init__(series, *args, **kw) me.title = title me.i = None def add_episode(me, j, neps, title, src, tno, startch, endch): if me.title is None and title is None: raise ExpectedError("movie or movie season must have a title") return super().add_episode(j, neps, title, src, tno, startch, endch) def _eplabel(me, i, neps, title): if me.title is None: label = title elif title is None: epname, epn = me._epnames(i, neps) label = "%s—%s %s" % (me.title, epname, ", ".join(epn)) else: label = "%s—%s" % (me.title, title) return label class Series (object): def __init__(me, playlist, name, title = None, wantedp = True): me.playlist = playlist me.name, me.title = name, title me.cur_season = None me.wantedp = wantedp def _add_season(me, season): me.cur_season = season def add_season(me, title, i, implicitp = False): me._add_season(Season(me, title, i, implicitp)) def add_movies(me, title = None): me._add_season(MovieSeason(me, title)) def ensure_season(me): if me.cur_season is None: me.add_season(None, 1, implicitp = True) return me.cur_season def end_season(me): me.cur_season = None class Playlist (object): def __init__(me): me.seasons = [] me.episodes = [] me.epname, me.epnames = "Episode", "Episodes" me.nseries = 0 def add_episode(me, episode): me.episodes.append(episode) def done_season(me): if me.episodes: me.seasons.append(me.episodes) me.episodes = [] def write(me, f): f.write("#EXTM3U\n") for season in me.seasons: f.write("\n") for ep in season: label = ep.label() if me.nseries > 1: label = ep.season.series.title + " " + label if not ep.chapters: f.write("#EXTINF:%d,,%s\n%s\n" % (ep.duration, label, ep.url)) else: for ch in ep.chapters: f.write("#EXTINF:%d,,%s: %s\n%s\n" % (ch.duration, label, ch.title, ch.url)) DEFAULT_EXPVAR = 0.05 R_DURMULT = RX.compile(r""" ^ (\d+ (?: \. \d+)?) x $ """, RX.X) R_DUR = RX.compile(r""" ^ (?: (?: (\d+) :)? (\d+) :)? (\d+) (?: / (\d+ (?: \. \d+)?) \%)? $ """, RX.X) def parse_duration(s, base = None, basevar = DEFAULT_EXPVAR): if base is not None: m = R_DURMULT.match(s) if m is not None: return base*float(m.group(1)), basevar m = R_DUR.match(s) if not m: raise ExpectedError("invalid duration spec `%s'" % s) hr, min, sec = map(lambda g: filter(m.group(g), int, 0), [1, 2, 3]) var = filter(m.group(4), lambda x: float(x)/100.0) if var is None: var = DEFAULT_EXPVAR return 3600*hr + 60*min + sec, var def format_duration(d): if d >= 3600: return "%d:%02d:%02d" % (d//3600, (d//60)%60, d%60) elif d >= 60: return "%d:%02d" % (d//60, d%60) else: return "%d s" % d MODE_UNSET = 0 MODE_SINGLE = 1 MODE_MULTI = 2 class EpisodeListParser (object): def __init__(me, series_wanted = None, chapters_wanted_p = False): me._pl = Playlist() me._cur_episode = me._cur_chapter = None me._series = {}; me._vdirs = {}; me._audirs = {}; me._isos = {} me._series_wanted = series_wanted me._chaptersp = chapters_wanted_p me._explen, me._expvar = None, DEFAULT_EXPVAR if series_wanted is None: me._mode = MODE_UNSET else: me._mode = MODE_MULTI def _bad_keyval(me, cmd, k, v): raise ExpectedError("invalid `!%s' option `%s'" % (cmd, v if k is None else k)) def _keyvals(me, opts): if opts is not None: for kv in opts.split(","): try: sep = kv.index("=") except ValueError: yield None, kv else: yield kv[:sep], kv[sep + 1:] def _set_mode(me, mode): if me._mode == MODE_UNSET: me._mode = mode elif me._mode != mode: raise ExpectedError("inconsistent single-/multi-series usage") def _get_series(me, name): if name is None: me._set_mode(MODE_SINGLE) try: series = me._series[None] except KeyError: series = me._series[None] = Series(me._pl, None) me._pl.nseries += 1 else: me._set_mode(MODE_MULTI) series = lookup(me._series, name, "unknown series `%s'" % name) return series def _opts_series(me, cmd, opts): name = None for k, v in me._keyvals(opts): if k is None: name = v else: me._bad_keyval(cmd, k, v) return me._get_series(name) def _auto_epsrc(me, series): dir = lookup(me._vdirs, series.name, "no active video directory") season = series.ensure_season() check(season.i is not None, "must use explicit iso for movie seasons") vseason = lookup(dir.seasons, season.i, "season %d not found in video dir `%s'" % (season.i, dir.dir)) src = lookup(vseason.episodes, season.ep_i, "episode %d.%d not found in video dir `%s'" % (season.i, season.ep_i, dir.dir)) return src def _process_cmd(me, ww): cmd = ww.nextword(); check(cmd is not None, "missing command") try: sep = cmd.index(":") except ValueError: opts = None else: cmd, opts = cmd[:sep], cmd[sep + 1:] if cmd == "series": name = None for k, v in me._keyvals(opts): if k is None: name = v else: me._bad_keyval(cmd, k, v) check(name is not None, "missing series name") check(name not in me._series, "series `%s' already defined" % name) title = ww.rest(); check(title is not None, "missing title") me._set_mode(MODE_MULTI) me._series[name] = series = Series(me._pl, name, title, me._series_wanted is None or name in me._series_wanted) if series.wantedp: me._pl.nseries += 1 elif cmd == "season": series = me._opts_series(cmd, opts) w = ww.nextword(); check(w is not None, "missing season number") if w == "-": if not series.wantedp: return series.add_movies(ww.rest()) else: title = ww.rest(); i = getint(w) if not series.wantedp: return series.add_season(ww.rest(), getint(w), implicitp = False) me._cur_episode = me._cur_chapter = None me._pl.done_season() elif cmd == "explen": w = ww.rest(); check(w is not None, "missing duration spec") d, v = parse_duration(w) me._explen = d if v is not None: me._expvar = v elif cmd == "epname": for k, v in me._keyvals(opts): me._bad_keyval("epname", k, v) name = ww.rest(); check(name is not None, "missing episode name") try: sep = name.index(",") except ValueError: names = name + "s" else: name, names = name[:sep], name[sep + 1:] me._pl.epname, me._pl.epnames = name, names elif cmd == "epno": series = me._opts_series(cmd, opts) w = ww.rest(); check(w is not None, "missing episode number") epi = getint(w) if not series.wantedp: return series.ensure_season().ep_i = epi elif cmd == "iso": series = me._opts_series(cmd, opts) fn = ww.rest(); check(fn is not None, "missing filename") if not series.wantedp: return if fn == "-": forget(me._isos, series.name) else: check(OS.path.exists(OS.path.join(ROOT, fn)), "iso file `%s' not found" % fn) me._isos[series.name] = VideoDisc(fn) elif cmd == "vdir": series = me._opts_series(cmd, opts) dir = ww.rest(); check(dir is not None, "missing directory") if not series.wantedp: return if dir == "-": forget(me._vdirs, series.name) else: me._vdirs[series.name] = VideoDir(dir) elif cmd == "adir": series = me._opts_series(cmd, opts) dir = ww.rest(); check(dir is not None, "missing directory") if not series.wantedp: return if dir == "-": forget(me._audirs, series.name) else: me._audirs[series.name] = AudioDir(dir) elif cmd == "displaced": series = me._opts_series(cmd, opts) w = ww.rest(); check(w is not None, "missing count"); n = getint(w) src = me._auto_epsrc(series) src.nuses += n else: raise ExpectedError("unknown command `%s'" % cmd) def _process_episode(me, ww): opts = ww.nextword(); check(opts is not None, "missing title/options") ti = None; sname = None; neps = 1; epi = None; loch = hich = None explen, expvar, explicitlen = me._explen, me._expvar, False for k, v in me._keyvals(opts): if k is None: if v.isdigit(): ti = int(v) elif v == "-": ti = "-" else: sname = v elif k == "s": sname = v elif k == "n": neps = getint(v) elif k == "ep": epi = getint(v) elif k == "l": if v == "-": me._explen, me._expvar = None, DEFAULT_EXPVAR else: explen, expvar = parse_duration(v, explen, expvar) explicitlen = True elif k == "ch": try: sep = v.index("-") except ValueError: loch, hich = getint(v), None else: loch, hich = getint(v[:sep]), getint(v[sep + 1:]) + 1 else: raise ExpectedError("unknown episode option `%s'" % k) check(ti is not None, "missing title number") series = me._get_series(sname) me._cur_chapter = None title = ww.rest() if not series.wantedp: return season = series.ensure_season() if epi is None: epi = season.ep_i if ti == "-": check(season.implicitp, "audio source, but explicit season") dir = lookup(me._audirs, series.name, "no title, and no audio directory") src = lookup(dir.episodes, season.ep_i, "episode %d not found in audio dir `%s'" % (epi, dir.dir)) else: try: src = me._isos[series.name] except KeyError: src = me._auto_epsrc(series) episode = season.add_episode(epi, neps, title, src, ti, loch, hich) if episode.duration != -1 and explen is not None: if not explicitlen: explen *= neps if not explen*(1 - expvar) <= episode.duration <= explen*(1 + expvar): if season.i is None: epid = "episode %d" % epi else: epid = "episode %d.%d" % (season.i, epi) raise ExpectedError \ ("%s duration %s %g%% > %g%% from expected %s" % (epid, format_duration(episode.duration), abs(100*(episode.duration - explen)/explen), 100*expvar, format_duration(explen))) me._pl.add_episode(episode) me._cur_episode = episode def _process_chapter(me, ww): check(me._cur_episode is not None, "no current episode") check(me._cur_episode.source.CHAPTERP, "episode source doesn't allow chapters") if me._chaptersp: if me._cur_chapter is None: i = 1 else: i = me._cur_chapter.i + 1 me._cur_chapter = me._cur_episode.add_chapter(ww.rest(), i) def parse_file(me, fn): with location(FileLocation(fn, 0)) as floc: with open(fn, "r") as f: for line in f: floc.stepline() sline = line.lstrip() if sline == "" or sline.startswith(";"): continue if line.startswith("!"): me._process_cmd(Words(line[1:])) elif not line[0].isspace(): me._process_episode(Words(line)) else: me._process_chapter(Words(line)) me._pl.done_season() def done(me): discs = set() for name, vdir in me._vdirs.items(): if not me._series[name].wantedp: continue for s in vdir.seasons.values(): for d in s.episodes.values(): discs.add(d) for adir in me._audirs.values(): for d in adir.episodes.values(): discs.add(d) for d in sorted(discs, key = lambda d: d.fn): if d.neps is not None and d.neps != d.nuses: raise ExpectedError("disc `%s' has %d episodes, used %d times" % (d.fn, d.neps, d.nuses)) return me._pl ROOT = "/mnt/dvd/archive/" op = OP.OptionParser \ (usage = "%prog [-c] [-s SERIES] EPLS", description = "Generate M3U playlists from an episode list.") op.add_option("-c", "--chapters", dest = "chaptersp", action = "store_true", default = False, help = "Output individual chapter names") op.add_option("-s", "--series", dest = "series", type = "str", default = None, help = "Output only the listed SERIES (comma-separated)") opts, argv = op.parse_args() if len(argv) != 1: op.print_usage(file = SYS.stderr); SYS.exit(2) if opts.series is None: series_wanted = None else: series_wanted = set() for name in opts.series.split(","): series_wanted.add(name) try: ep = EpisodeListParser(series_wanted, opts.chaptersp) ep.parse_file(argv[0]) pl = ep.done() pl.write(SYS.stdout) except (ExpectedError, IOError, OSError) as e: LOC.report(e) SYS.exit(2)