diff options
Diffstat (limited to 'desktop')
| -rwxr-xr-x | desktop/assistant/fileops | 225 | ||||
| -rwxr-xr-x | desktop/assistant/test_fileops.py | 141 |
2 files changed, 366 insertions, 0 deletions
diff --git a/desktop/assistant/fileops b/desktop/assistant/fileops new file mode 100755 index 0000000..4bfaee0 --- /dev/null +++ b/desktop/assistant/fileops @@ -0,0 +1,225 @@ +#!/usr/bin/python3 +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 2 as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General +# Public License for more details. +"""fileops <op> <outdir> <file...> + +One local operation on files, for the assistant canvas's Do row. Prints one +JSON line, {"ok": true, "out": [paths]} or {"ok": false, "error": "..."}. +Each result is written under a temporary name and renamed into place, to +the first free name; SIGTERM kills the running tool and removes what was +half written. +""" + +import json +import os +import shutil +import signal +import subprocess +import sys +import tarfile +import zipfile + +current = None # the running tool, killed with its group on SIGTERM +temps = [] # partial results to remove if we do not finish + +ARCHIVE_EXTS = (".tar.gz", ".tar.xz", ".tar.bz2", ".tar.zst", ".tgz", ".txz", + ".zip", ".tar", ".7z", ".rar", ".gz", ".xz", ".bz2", ".zst") + + +class Fail(Exception): + pass + + +def finish(result, code): + print(json.dumps(result), flush=True) + sys.exit(code) + + +def stop(*_): + if current is not None: + try: + os.killpg(current.pid, signal.SIGKILL) + except ProcessLookupError: + pass + for t in temps: + shutil.rmtree(t, ignore_errors=True) if os.path.isdir(t) else _unlink(t) + finish({"ok": False, "error": "stopped"}, 1) + + +def _unlink(path): + try: + os.unlink(path) + except FileNotFoundError: + pass + + +def run(argv): + # Own process group, so a stop takes ffmpeg and its children too. + global current + if shutil.which(argv[0]) is None: + hint = ": link it into ~/bin" if argv[0] == "trackcrop" else "" + raise Fail(f"{argv[0]} not found{hint}") + current = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, start_new_session=True) + out, err = current.communicate() + rc, current = current.returncode, None + if rc: + last = err.strip().splitlines()[-1:] or [f"exit status {rc}"] + raise Fail(f"{argv[0]}: {last[0]}"[:300]) + return out + + +def split(name): + # ("report", ".pdf"), keeping a compound suffix whole: "a.tar.gz". + low = name.lower() + for ext in (".tar.gz", ".tar.xz", ".tar.bz2", ".tar.zst"): + if low.endswith(ext): + return name[:-len(ext)], name[-len(ext):] + stem, ext = os.path.splitext(name) + return stem, ext + + +def stem(path): + name = os.path.basename(path) + for ext in ARCHIVE_EXTS: + if name.lower().endswith(ext) and len(name) > len(ext): + return name[:-len(ext)] + return os.path.splitext(name)[0] + + +def free(outdir, name): + path = os.path.join(outdir, name) + base, ext = split(name) + n = 1 + while os.path.lexists(path): + path = os.path.join(outdir, f"{base} ({n}){ext}") + n += 1 + return path + + +def make(outdir, name, build, folder=False): + # build(tmp) writes the result at tmp; it keeps the suffix, which is how + # ffmpeg and magick choose the format. + tmp = os.path.join(outdir, f".fileops-{os.getpid()}-{len(temps)}{split(name)[1]}") + temps.append(tmp) + if folder: + os.mkdir(tmp) + build(tmp) + final = free(outdir, name) + os.rename(tmp, final) + temps.remove(tmp) + return final + + +def audio_codec(path): + return run(["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", + "stream=codec_name", "-of", "csv=p=0", path]).strip() + + +# Copied as it is when the track already is one of these. +COPY = {"aac": ".m4a", "mp3": ".mp3", "opus": ".opus"} +GIF = "fps=12,scale=480:-1:flags=lanczos,split[a][b];[a]palettegen[p];[b][p]paletteuse" +# This ffmpeg has no libx264 and no AAC encoder: H.264 is the GPU's, through +# VAAPI, with MP3 audio, which MP4 carries. +VAAPI = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload", + "-c:v", "h264_vaapi", "-qp", "23"] + + +def each(op, outdir, f): + s, ext = stem(f), os.path.splitext(f)[1] + ff = ["ffmpeg", "-v", "error", "-y", "-i", f] + if op == "pdf-images": + return make(outdir, f"{s} (pages)", + lambda t: run(["pdftoppm", "-r", "150", "-png", f, os.path.join(t, "page")]), + folder=True) + if op == "audio": + codec = audio_codec(f) + if not codec: + raise Fail(f"{os.path.basename(f)} has no audio track") + if codec in COPY: + return make(outdir, s + COPY[codec], lambda t: run(ff + ["-vn", "-c:a", "copy", t])) + return make(outdir, s + ".mp3", lambda t: run(ff + ["-vn", "-c:a", "libmp3lame", "-q:a", "2", t])) + if op == "mp4": + return make(outdir, s + ".mp4", + lambda t: run(ff + VAAPI + ["-c:a", "libmp3lame", "-q:a", "2", + "-movflags", "+faststart", t])) + if op == "webm": + return make(outdir, s + ".webm", + lambda t: run(ff + ["-c:v", "libvpx-vp9", "-b:v", "0", "-crf", "33", + "-c:a", "libopus", t])) + if op == "gif": + return make(outdir, s + ".gif", lambda t: run(ff + ["-vf", GIF, t])) + if op == "trackcrop": + return make(outdir, f"{s} (tracked).mp4", lambda t: run(["trackcrop", f, t])) + if op in ("png", "jpeg", "webp"): + return make(outdir, f"{s}.{'jpg' if op == 'jpeg' else op}", lambda t: run(["magick", f, t])) + if op == "half": + return make(outdir, f"{s} (50%){ext}", lambda t: run(["magick", f, "-resize", "50%", t])) + if op == "extract": + return make(outdir, s, lambda t: run(["bsdtar", "-xf", f, "-C", t]), folder=True) + raise Fail(f"unknown operation {op}") + + +def compress(op, outdir, files): + ext = {"zip": ".zip", "tgz": ".tar.gz", "txz": ".tar.xz", "tbz2": ".tar.bz2"}[op] + name = (stem(files[0]) if len(files) == 1 else "archive") + ext + + def build(t): + if op == "zip": + with zipfile.ZipFile(t, "w", zipfile.ZIP_DEFLATED) as z: + for f in files: + if os.path.isdir(f): + for root, _, names in os.walk(f): + for n in names: + p = os.path.join(root, n) + z.write(p, os.path.relpath(p, os.path.dirname(f))) + else: + z.write(f, os.path.basename(f)) + else: + with tarfile.open(t, "w:" + {"tgz": "gz", "txz": "xz", "tbz2": "bz2"}[op]) as tar: + for f in files: + tar.add(f, arcname=os.path.basename(f)) + return make(outdir, name, build) + + +def main(argv): + if len(argv) < 3: + raise Fail("usage: fileops <op> <outdir> <file...>") + op, outdir, files = argv[0], argv[1], argv[2:] + if not os.path.isabs(outdir): + raise Fail("the output folder must be an absolute path") + for f in files: + if not os.path.exists(f): + raise Fail(f"cannot read {os.path.basename(f)}") + os.makedirs(outdir, exist_ok=True) + if op == "pdf-merge": + if len(files) < 2: + raise Fail("merging needs two or more PDFs") + return [make(outdir, "merged.pdf", + lambda t: run(["qpdf", "--empty", "--pages", *files, "--", t]))] + if op in ("zip", "tgz", "txz", "tbz2"): + return [compress(op, outdir, files)] + return [each(op, outdir, f) for f in files] + + +if __name__ == "__main__": + signal.signal(signal.SIGTERM, stop) + try: + out = main(sys.argv[1:]) + except Fail as e: + for t in temps: + shutil.rmtree(t, ignore_errors=True) if os.path.isdir(t) else _unlink(t) + finish({"ok": False, "error": str(e)}, 1) + except Exception as e: # a bug: say so rather than hang the card + for t in temps: + shutil.rmtree(t, ignore_errors=True) if os.path.isdir(t) else _unlink(t) + finish({"ok": False, "error": f"{type(e).__name__}: {e}"}, 1) + finish({"ok": True, "out": out}, 0) diff --git a/desktop/assistant/test_fileops.py b/desktop/assistant/test_fileops.py new file mode 100755 index 0000000..850d1d7 --- /dev/null +++ b/desktop/assistant/test_fileops.py @@ -0,0 +1,141 @@ +#!/usr/bin/python3 +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU General Public License version 2 as published +# by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General +# Public License for more details. +"""Run: /usr/bin/python3 desktop/assistant/test_fileops.py + +Builds throwaway inputs in a temp folder and runs every operation but +trackcrop (it opens a window) into a temp output folder. +""" + +import json +import os +import signal +import subprocess +import sys +import tarfile +import tempfile +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +FILEOPS = os.path.join(HERE, "fileops") + + +def fileops(op, outdir, *files): + p = subprocess.run([sys.executable, FILEOPS, op, outdir, *files], + capture_output=True, text=True, timeout=120) + return json.loads(p.stdout.strip().splitlines()[-1]), p.returncode + + +def mime(path): + return subprocess.run(["file", "-b", "--mime-type", path], capture_output=True, + text=True).stdout.strip() + + +def sh(*argv): + subprocess.run(argv, check=True, capture_output=True) + + +with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as out: + pdf_a, pdf_b = os.path.join(src, "a.pdf"), os.path.join(src, "b.pdf") + png = os.path.join(src, "pic.png") + # This ffmpeg has no libx264 and no AAC encoder: inputs are VP9 + Opus + # in WebM and MPEG-4 + PCM in Matroska. + clip = os.path.join(src, "clip.webm") + raw = os.path.join(src, "raw.mkv") + tarball = os.path.join(src, "bundle.tar") + sh("magick", "-size", "200x100", "xc:white", pdf_a) + sh("magick", "-size", "200x100", "xc:gray", pdf_b) + sh("magick", "-size", "64x32", "xc:red", png) + sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=1:size=160x120:rate=10", + "-f", "lavfi", "-i", "sine=duration=1", "-c:v", "libvpx-vp9", "-c:a", "libopus", "-shortest", clip) + sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=1:size=160x120:rate=10", + "-f", "lavfi", "-i", "sine=duration=1", "-c:v", "mpeg4", "-c:a", "pcm_s16le", "-shortest", raw) + with open(os.path.join(src, "note.txt"), "w") as f: + f.write("hello\n") + with tarfile.open(tarball, "w") as t: + t.add(os.path.join(src, "note.txt"), arcname="note.txt") + + def one(op, *files): + r, rc = fileops(op, out, *files) + assert r["ok"] and rc == 0, (op, r) + return r["out"] + + # Each result, by the type file(1) sees in it. + [m] = one("pdf-merge", pdf_a, pdf_b) + assert mime(m) == "application/pdf" and os.path.basename(m) == "merged.pdf" + assert subprocess.run(["qpdf", "--show-npages", m], capture_output=True, + text=True).stdout.strip() == "2" + [d] = one("pdf-images", pdf_a) + assert os.path.basename(d) == "a (pages)" and [mime(os.path.join(d, x)) for x in os.listdir(d)] == ["image/png"] + [a] = one("audio", clip) # Opus: copied as it is + assert a.endswith("clip.opus") and mime(a) == "audio/ogg", mime(a) + [a2] = one("audio", raw) # PCM: encoded to MP3 + assert a2.endswith("raw.mp3") and mime(a2) == "audio/mpeg", mime(a2) + [w] = one("webm", raw) + assert mime(w) == "video/webm" and w.endswith("raw.webm") + [v] = one("mp4", raw) # h264_vaapi + MP3 + assert mime(v) == "video/mp4" and v.endswith("raw.mp4") + [g] = one("gif", clip) + assert mime(g) == "image/gif" + [j] = one("jpeg", png) + assert mime(j) == "image/jpeg" + [p] = one("png", j) + assert mime(p) == "image/png" and os.path.basename(p) == "pic.png" + [wp] = one("webp", png) + assert mime(wp) == "image/webp" + [h] = one("half", png) + assert os.path.basename(h) == "pic (50%).png" + assert subprocess.run(["magick", "identify", "-format", "%wx%h", h], capture_output=True, + text=True).stdout == "32x16" + [x] = one("extract", tarball) + assert os.path.basename(x) == "bundle" and os.listdir(x) == ["note.txt"] + for op, want, ext in (("zip", "application/zip", ".zip"), ("tgz", "application/gzip", ".tar.gz"), + ("txz", "application/x-xz", ".tar.xz"), ("tbz2", "application/x-bzip2", ".tar.bz2")): + [z] = one(op, png, pdf_a) + assert mime(z) == want and os.path.basename(z) == "archive" + ext, (op, z) + [z1] = one("zip", png) # one file: named after it + assert os.path.basename(z1) == "pic.zip" + + # One result per file, and taken names get a number. + rs = one("png", png, png) + assert [os.path.basename(r) for r in rs] == ["pic (1).png", "pic (2).png"], rs + + # A failure leaves nothing behind. + before = set(os.listdir(out)) + fake = os.path.join(src, "fake.mp4") + with open(fake, "w") as f: + f.write("not a video\n") + r, rc = fileops("gif", out, fake) + assert not r["ok"] and rc == 1 and r["error"], r + assert set(os.listdir(out)) == before, set(os.listdir(out)) - before + for bad in (("nope", out, png), ("png", "relative/dir", png), ("png", out, src + "/missing.png")): + r, rc = fileops(*bad) + assert not r["ok"] and rc == 1, bad + r, _ = fileops("audio", out, g) + assert not r["ok"] and "audio" in r["error"], r # a GIF has no audio track + + # SIGTERM mid-encode (the canvas's Stop): no ffmpeg left, no partial file. + long_clip = os.path.join(src, "long.mp4") + sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=20:size=1280x720:rate=30", + "-c:v", "mpeg4", long_clip) + with tempfile.TemporaryDirectory() as slow: + proc = subprocess.Popen([sys.executable, FILEOPS, "webm", slow, long_clip], + stdout=subprocess.PIPE, text=True) + time.sleep(1.5) + proc.send_signal(signal.SIGTERM) + r = json.loads(proc.communicate(timeout=10)[0].strip().splitlines()[-1]) + assert not r["ok"] and r["error"] == "stopped", r + time.sleep(0.5) + left = subprocess.run(["pgrep", "-f", slow], capture_output=True, text=True).stdout + assert left == "", left + assert os.listdir(slow) == [], os.listdir(slow) + +print("ok") |