mtp daisychaining

bin/mtag/vidchk.py

@@ -1,15 +1,19 @@
 #!/usr/bin/env python3
 
+import json
 import sys
 import subprocess as sp
 
 from copyparty.util import fsenc
-from copyparty.mtag import ffprobe
 
-
-"""
+_ = r"""
 inspects video files for errors and such
-usage: -mtp vidchk=t600,ay,bin/mtag/vidchk.py
+usage: -mtp vidchk=t600,ay,p,bin/mtag/vidchk.py
+
+t600: timeout 10min
+  ay: only process files which contain audio (including video with audio)
+   p: set priority 1 (lowest priority after initial ffprobe/mutagen for base tags),
+       makes copyparty feed base tags into this script as json
 """
 
 
@@ -19,17 +23,18 @@ FAST = True  # parse entire file at container level
 
 def main():
     fp = sys.argv[1]
-    md, _ = ffprobe(fp)
+    zb = sys.stdin.buffer.read()
+    zs = zb.decode("utf-8", "replace")
+    md = json.loads(zs)
 
     try:
-        w = int(md[".resw"][1])
-        h = int(md[".resh"][1])
+        w, h = [int(x) for x in md["res"].split("x")]
         if not w + h:
             raise Exception()
     except:
         return "could not determine resolution"
 
-    if min(w, h) < 720:
+    if min(w, h) < 1080:
         return "resolution too small"
 
     zs = (

copyparty/mtag.py

@@ -46,6 +46,7 @@ class MParser(object):
         self.force = False
         self.kill = "t"  # tree; all children recursively
         self.audio = "y"
+        self.pri = 0  # priority; higher = later
         self.ext = []
 
         while True:
@@ -83,6 +84,10 @@ class MParser(object):
                 self.ext.append(arg[1:])
                 continue
 
+            if arg.startswith("p"):
+                self.pri = int(arg[1:] or "1")
+                continue
+
             raise Exception()
 
 
@@ -487,7 +492,9 @@ class MTag(object):
         ret, md = ffprobe(abspath)
         return self.normalize_tags(ret, md)
 
-    def get_bin(self, parsers: dict[str, MParser], abspath: str) -> dict[str, Any]:
+    def get_bin(
+        self, parsers: dict[str, MParser], abspath: str, oth_tags: dict[str, Any]
+    ) -> dict[str, Any]:
         if not bos.path.isfile(abspath):
             return {}
 
@@ -498,7 +505,7 @@ class MTag(object):
         env["PYTHONPATH"] = pypath
 
         ret = {}
-        for tagname, parser in parsers.items():
+        for tagname, parser in sorted(parsers.items(), key=lambda x: (x[1].pri, x[0])):
             try:
                 cmd = [parser.bin, abspath]
                 if parser.bin.endswith(".py"):
@@ -506,6 +513,9 @@ class MTag(object):
 
                 args = {"env": env, "timeout": parser.timeout, "kill": parser.kill}
 
+                if parser.pri:
+                    args["sin"] = json.dumps(oth_tags).encode("utf-8", "replace")
+
                 if WINDOWS:
                     args["creationflags"] = 0x4000
                 else:

copyparty/up2k.py

@@ -67,12 +67,20 @@ class Dbw(object):
 
 
 class Mpqe(object):
-    def __init__(self, mtp: dict[str, MParser], entags: set[str], w: str, abspath: str):
+    def __init__(
+        self,
+        mtp: dict[str, MParser],
+        entags: set[str],
+        w: str,
+        abspath: str,
+        oth_tags: dict[str, Any],
+    ):
         # mtp empty = mtag
         self.mtp = mtp
         self.entags = entags
         self.w = w
         self.abspath = abspath
+        self.oth_tags = oth_tags
 
 
 class Up2k(object):
@@ -872,7 +880,7 @@ class Up2k(object):
                 if not mpool:
                     n_tags = self._tag_file(c3, entags, w, abspath)
                 else:
-                    mpool.put(Mpqe({}, entags, w, abspath))
+                    mpool.put(Mpqe({}, entags, w, abspath, {}))
                     # not registry cursor; do not self.mutex:
                     n_tags = len(self._flush_mpool(c3))
 
@@ -978,8 +986,8 @@ class Up2k(object):
                     abspath = os.path.join(ptop, rd, fn)
 
                     q = "select k from mt where w = ?"
-                    zq2 = cur.execute(q, (w,)).fetchall()
-                    have: dict[str, Union[str, float]] = {x[0]: 1 for x in zq2}
+                    zq = cur.execute(q, (w,)).fetchall()
+                    have: dict[str, Union[str, float]] = {x[0]: 1 for x in zq}
 
                     did_nothing = False
                     parsers = self._get_parsers(ptop, have, abspath)
@@ -988,7 +996,14 @@ class Up2k(object):
                         n_left -= 1
                         continue
 
-                    jobs.append(Mpqe(parsers, set(), w, abspath))
+                    if next((x for x in parsers.values() if x.pri), None):
+                        q = "select k, v from mt where w = ?"
+                        zq2 = cur.execute(q, (w,)).fetchall()
+                        oth_tags = {str(k): v for k, v in zq2}
+                    else:
+                        oth_tags = {}
+
+                    jobs.append(Mpqe(parsers, set(), w, abspath, oth_tags))
                     in_progress[w] = True
 
             with self.mutex:
@@ -1112,7 +1127,7 @@ class Up2k(object):
             return
 
         for _ in range(mpool.maxsize):
-            mpool.put(Mpqe({}, set(), "", ""))
+            mpool.put(Mpqe({}, set(), "", "", {}))
 
         mpool.join()
 
@@ -1128,7 +1143,7 @@ class Up2k(object):
                 if not qe.mtp:
                     tags = self.mtag.get(qe.abspath)
                 else:
-                    tags = self.mtag.get_bin(qe.mtp, qe.abspath)
+                    tags = self.mtag.get_bin(qe.mtp, qe.abspath, qe.oth_tags)
                     vtags = [
                         "\033[36m{} \033[33m{}".format(k, v) for k, v in tags.items()
                     ]
@@ -2383,7 +2398,7 @@ class Up2k(object):
                 ntags1 = len(tags)
                 parsers = self._get_parsers(ptop, tags, abspath)
                 if parsers:
-                    tags.update(self.mtag.get_bin(parsers, abspath))
+                    tags.update(self.mtag.get_bin(parsers, abspath, tags))
             except Exception as ex:
                 self._log_tag_err("", abspath, ex)
                 continue

copyparty/util.py

@@ -1566,12 +1566,17 @@ def runcmd(
     argv: Union[list[bytes], list[str]], timeout: Optional[int] = None, **ka: Any
 ) -> tuple[int, str, str]:
     kill = ka.pop("kill", "t")  # [t]ree [m]ain [n]one
+
+    sin = ka.pop("sin", None)
+    if sin:
+        ka["stdin"] = sp.PIPE
+
     p = sp.Popen(argv, stdout=sp.PIPE, stderr=sp.PIPE, **ka)
     if not timeout or PY2:
-        stdout, stderr = p.communicate()
+        stdout, stderr = p.communicate(sin)
     else:
         try:
-            stdout, stderr = p.communicate(timeout=timeout)
+            stdout, stderr = p.communicate(sin, timeout=timeout)
         except sp.TimeoutExpired:
             if kill == "n":
                 return -18, "", ""  # SIGCONT; leave it be