Initial commit.

[epls] / mkm3u

#! /usr/bin/python3
### -*- mode: python; coding: utf-8 -*-

from contextlib import contextmanager
import os as OS
import re as RX
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 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()

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
  def url(me, title = None, chapter = None):
    if title is None:
      if me.TITLEP: raise ExpectedError("missing title number")
      if 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 chapter is None:
      suffix = "#%d" % title
    elif not me.CHAPTERP:
      raise ExpectedError("can't specify chapter with `%s'" % me.fn)
    else:
      suffix = "#%d:%d-%d:%d" % (title, chapter, title, chapter)
    return me.PREFIX + ROOT + urlencode(me.fn) + suffix

class VideoDisc (Source):
  PREFIX = "dvd://"
  TITLEP = CHAPTERP = True

class VideoEpisodes (VideoDisc):
  def __init__(me, fn, season, eps, *args, **kw):
    super().__init__(fn, *args, **kw)
    me.season = season
    me.eps = eps

class VideoSeason (object):
  def __init__(me, i, title):
    me.i = i
    me.title = title
    me.episodes = {}
  def add_disc(me, fn, eps):
    d = VideoEpisodes(fn, me, eps)
    for i in eps:
      if i in me.episodes:
        raise ExpectedError("season %d episode %d already taken" % (me.i, i))
      me.episodes[i] = d
    return d

def some_group(m, *gg):
  for g in gg:
    s = m.group(g)
    if s is not None: return s
  return None

class VideoDir (object):

  _R_ISO_PRE = RX.compile(r"""
        ^
        (?: S (?P<si> \d+) (?: \. \ (?P<st> .*)—)? (?: D (?P<sdi> \d+))? |
            (?P<di> \d+))
        \. \  #
        (?P<eps> .*)
        \. iso $
  """, RX.X)

  _R_ISO_EP = RX.compile(r"""
        ^ E (?P<ei> \d+) (?: – (?P<ej> \d+))? $
  """, RX.X)

  def __init__(me, dir):
    fns = OS.listdir(OS.path.join(ROOT, dir))
    fns.sort()
    season, last_j = None, 0
    seasons = {}
    for fn in fns:
      path = OS.path.join(dir, fn)
      if not fn.endswith(".iso"): continue
      m = me._R_ISO_PRE.match(fn)
      if not m: continue

      i = filter(m.group("si"), int, 1)
      stitle = m.group("st")
      if season is None or i != season.i:
        check(season is None or i == season.i + 1,
              "season %d /= %d" % (i, season is None and -1 or season.i + 1))
        check(i not in seasons, "season %d already seen" % i)
        seasons[i] = season = VideoSeason(i, stitle)
        last_j = 0
      else:
        check(stitle == season.title,
              "season title `%s' /= `%s'" % (stitle, season.title))
      j = filter(some_group(m, "sdi", "di"), int)
      if j is not None:
        check(j == last_j + 1,
              "season %d disc %d /= %d" % (season.i, j, last_j + 1))

      eps = set()
      bad = False
      for eprange in m.group("eps").split(", "):
        mm = me._R_ISO_EP.match(eprange)
        if mm is None: bad = True; continue
        start = filter(mm.group("ei"), int)
        end = filter(mm.group("ej"), int, start)
        for k in range(start, end + 1): eps.add(k)
      if bad and eps:
        raise ExpectedError("bad ep list in `%s'", fn)
      season.add_disc(path, eps)
      last_j = j
    me.seasons = seasons

class AudioDisc (Source):
  #PREFIX = "file://"
  TITLEP = CHAPTERP = False

class AudioEpisode (Source):
  #PREFIX = "file://"
  TITLEP = CHAPTERP = False
  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):
    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 = episode.source.url(episode.tno, i)

class Episode (object):
  def __init__(me, season, i, neps, title, src, tno = None):
    me.season = season
    me.i, me.neps, me.title = i, neps, title
    me.chapters = []
    me.source, me.tno = src, tno
    me.url = src.url(tno)
  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 Season (object):
  def __init__(me, playlist, title, i, implicitp = False):
    me.playlist = playlist
    me.title, me.i = title, i
    me.implicitp = implicitp
    me.episodes = []
  def add_episode(me, j, neps, title, src, tno):
    ep = Episode(me, j, neps, title, src, tno)
    me.episodes.append(ep)
    return ep
  def _eplabel(me, i, neps, title):
    if neps == 1: epname = me.playlist.epname; epn = "%d" % i
    elif neps == 2: epname = me.playlist.epnames; epn = "%d, %d" % (i, i + 1)
    else: epname = me.playlist.epnames; epn = "%d–%d" % (i, i + neps - 1)
    if title is None:
      if me.implicitp: return "%s %s" % (epname, epn)
      elif me.title is None: return "%s %d.%s" % (epname, me.i, epn)
      else: return "%s—%s %s" % (me.title, epname, epn)
    else:
      if me.implicitp: return "%s. %s" % (epn, title)
      elif me.title is None: return "%d.%s. %s" % (me.i, epn, title)
      else: return "%s—%s. %s" % (me.title, epn, title)

class MovieSeason (object):
  def __init__(me, playlist):
    me.playlist = playlist
    me.i = -1
    me.implicitp = False
    me.episodes = []
  def add_episode(me, j, neps, title, src, tno):
    if title is None: raise ExpectedError("movie must have a title")
    ep = Episode(me, j, neps, title, src, tno)
    me.episodes.append(ep)
    return ep
  def _eplabel(me, i, epn, title):
    return title

class Playlist (object):

  def __init__(me):
    me.seasons = []
    me.epname, me.epnames = "Episode", "Episodes"

  def add_season(me, title, i, implicitp = False):
    season = Season(me, title, i, implicitp)
    me.seasons.append(season)
    return season

  def add_movies(me):
    season = MovieSeason(me)
    me.seasons.append(season)
    return season

  def write(me, f):
    f.write("#EXTM3U\n")
    for season in me.seasons:
      f.write("\n")
      for i, ep in enumerate(season.episodes, 1):
        if not ep.chapters:
          f.write("#EXTINF:0,,%s\n%s\n" % (ep.label(), ep.url))
        else:
          for ch in ep.chapters:
            f.write("#EXTINF:0,,%s: %s\n%s\n" %
                    (ep.label(), ch.title, ch.url))

UNSET = ["UNSET"]

def parse_list(fn):
  playlist = Playlist()
  season, episode, chapter, ep_i = None, None, None, 1
  vds = {}
  ads = iso = None
  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("!"):
          ww = Words(line[1:])
          cmd = ww.nextword()
          check(cmd is not None, "missing command")

          if cmd == "season":
            v = ww.nextword();
            check(v is not None, "missing season number")
            if v == "-":
              check(v.rest() is None, "trailing junk")
              season = playlist.add_movies()
            else:
              i = getint(v)
              title = ww.rest()
              season = playlist.add_season(title, i, implicitp = False)
            episode = chapter = None
            ep_i = 1

          elif cmd == "movie":
            check(ww.rest() is None, "trailing junk")
            season = playlist.add_movies()
            episode = chapter = None
            ep_i = 1

          elif cmd == "epname":
            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:]
            playlist.epname, playlist.epnames = name, names

          elif cmd == "epno":
            i = ww.rest()
            check(i is not None, "missing episode number")
            ep_i = getint(i)

          elif cmd == "iso":
            fn = ww.rest(); check(fn is not None, "missing filename")
            if fn == "-": iso = None
            else:
              check(OS.path.exists(OS.path.join(ROOT, fn)),
                    "iso file `%s' not found" % fn)
              iso = VideoDisc(fn)

          elif cmd == "vdir":
            name = ww.nextword(); check(name is not None, "missing name")
            fn = ww.rest(); check(fn is not None, "missing directory")
            if fn == "-":
              try: del vds[name]
              except KeyError: pass
            else:
              vds[name] = VideoDir(fn)

          elif cmd == "adir":
            fn = ww.rest(); check(fn is not None, "missing directory")
            if fn == "-": ads = None
            else: ads = AudioDir(fn)

          elif cmd == "end":
            break

          else:
            raise ExpectedError("unknown command `%s'" % cmd)

        else:

          if not line[0].isspace():
            ww = Words(line)
            conf = ww.nextword()

            check(conf is not None, "missing config")
            i, vdname, neps, fake_epi = UNSET, "-", 1, ep_i
            for c in conf.split(","):
              if c.isdigit(): i = int(c)
              elif c == "-": i = None
              else:
                eq = c.find("="); check(eq >= 0, "bad assignment `%s'" % c)
                k, v = c[:eq], c[eq + 1:]
                if k == "vd": vdname = v
                elif k == "n": neps = getint(v)
                elif k == "ep": fake_epi = getint(v)
                else: raise ExpectedError("unknown setting `%s'" % k)

            title = ww.rest()
            check(i is not UNSET, "no title number")
            if season is None:
              season = playlist.add_season(None, 1, implicitp = True)

            if i is None:
              check(ads, "no title, but no audio directory")
              check(season.implicitp, "audio source, but explicit season")
              src = ads.episodes[ep_i]

            elif iso:
              src = iso

            else:
              check(vdname in vds, "title, but no iso or video directory")
              src = vds[vdname].seasons[season.i].episodes[ep_i]

            episode = season.add_episode(fake_epi, neps, title, src, i)
            chapter = None
            ep_i += neps

          else:
            ww = Words(line)
            title = ww.rest()
            check(episode is not None, "no current episode")
            check(episode.source.CHAPTERP,
                  "episode source doesn't allow chapters")
            if chapter is None: j = 1
            else: j += 1
            chapter = episode.add_chapter(title, j)

  return playlist

ROOT = "/mnt/dvd/archive/"

try:
  for f in SYS.argv[1:]:
    parse_list(f).write(SYS.stdout)
except (ExpectedError, IOError, OSError) as e:
  LOC.report(e)
  SYS.exit(2)