medias 0.6.1

TV media player on bare DRM/KMS: Vulkan UI plane over libmpv video, for local files, UPnP/DLNA servers and YouTube, controlled by HDMI-CEC remote, evdev or terminal
#!/usr/bin/env python3
"""Build a signed apt repository from dist/*.deb.

Layout (Debian archive, one suite):
  pool/main/m/medias/*.deb           every published version, never rewritten
  dists/<suite>/main/binary-<arch>/Packages{,.gz}
  dists/<suite>/{Release,InRelease,Release.gpg}
  medias.asc                         public signing key
  medias.sources                     deb822 source with the key embedded (needs --url)
Needs dpkg-scanpackages (dpkg) and gpg.
"""

import argparse
import email.utils
import filecmp
import gzip
import hashlib
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

from dist import cargo_meta

ROOT = Path(__file__).resolve().parent.parent
ARCHES = ("amd64", "arm64")
COMPONENT = "main"
# User ID of the key created by `cargo make repo-key`.
KEY_NAME = "medias apt repository"


def run(cmd, **kw):
    print("+", " ".join(map(str, cmd)), flush=True)
    return subprocess.run(cmd, check=True, **kw)


def deb_field(deb, field):
    return subprocess.run(["dpkg-deb", "--field", str(deb), field], check=True,
                          capture_output=True, text=True).stdout.strip()


def add_to_pool(debs, repo, replace):
    for deb in debs:
        package = deb_field(deb, "Package")
        pool = repo / "pool" / COMPONENT / package[0] / package
        pool.mkdir(parents=True, exist_ok=True)
        target = pool / deb.name
        if target.exists() and not filecmp.cmp(deb, target, shallow=False):
            if not replace:
                sys.exit(f"repo: {target.relative_to(repo)} exists with other contents; "
                         "bump the version or pass --replace")
            print(f"repo: replacing {target.relative_to(repo)}")
        shutil.copy2(deb, target)


def write_indexes(repo, suite):
    """Packages{,.gz} per architecture; returns their paths relative to dists/<suite>."""
    dist = repo / "dists" / suite
    files = []
    for arch in ARCHES:
        rel = Path(COMPONENT) / f"binary-{arch}"
        (dist / rel).mkdir(parents=True, exist_ok=True)
        packages = subprocess.run(
            ["dpkg-scanpackages", "--multiversion", "--arch", arch, "pool", "/dev/null"],
            cwd=repo, check=True, capture_output=True).stdout
        (dist / rel / "Packages").write_bytes(packages)
        (dist / rel / "Packages.gz").write_bytes(gzip.compress(packages, 9, mtime=0))
        files += [rel / "Packages", rel / "Packages.gz"]
    return files


def release_text(repo, suite, files, epoch):
    dist = repo / "dists" / suite
    lines = [
        "Origin: medias",
        "Label: medias",
        f"Suite: {suite}",
        f"Codename: {suite}",
        f"Date: {email.utils.formatdate(epoch, usegmt=True)}",
        f"Architectures: {' '.join(ARCHES)}",
        f"Components: {COMPONENT}",
        f"Description: medias for Debian {suite}",
    ]
    for name, algo in (("MD5Sum", hashlib.md5), ("SHA256", hashlib.sha256)):
        lines.append(f"{name}:")
        for rel in files:
            data = (dist / rel).read_bytes()
            lines.append(f" {algo(data).hexdigest()} {len(data):>16} {rel.as_posix()}")
    return "\n".join(lines) + "\n"


def sign(repo, suite, key):
    dist = repo / "dists" / suite
    gpg = ["gpg", "--yes", "--local-user", key, "--digest-algo", "SHA512"]
    run(gpg + ["--clearsign", "-o", str(dist / "InRelease"), str(dist / "Release")])
    run(gpg + ["--armor", "--detach-sign", "-o", str(dist / "Release.gpg"), str(dist / "Release")])
    public = subprocess.run(["gpg", "--armor", "--export", key], check=True,
                            capture_output=True, text=True).stdout
    if not public.strip():
        sys.exit(f"repo: gpg exported no public key for {key}")
    (repo / "medias.asc").write_text(public)
    return public


def sources_text(url, suite, public):
    """deb822 source; Signed-By embeds the armored key, empty lines become ' .'.
    Architectures stops apt from requesting foreign ones (armhf on Raspberry Pi OS)."""
    key = "\n".join(f" {line}" if line else " ." for line in public.strip().splitlines())
    return (f"Types: deb\nURIs: {url}\nSuites: {suite}\nComponents: {COMPONENT}\n"
            f"Architectures: {' '.join(ARCHES)}\nSigned-By:\n{key}\n")


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("debs", nargs="*", type=Path,
                        help="packages (default: dist/*.deb of the Cargo.toml version)")
    parser.add_argument("--repo", type=Path, default=ROOT / "dist" / "debian", help="repository directory")
    parser.add_argument("--suite", default="trixie")
    parser.add_argument("--key", default=os.environ.get("MEDIAS_REPO_KEY", KEY_NAME),
                        help=f"gpg signing key (default: $MEDIAS_REPO_KEY or '{KEY_NAME}')")
    parser.add_argument("--url", default=os.environ.get("MEDIAS_REPO_URL"),
                        help="public repository URL for medias.sources (default: $MEDIAS_REPO_URL)")
    parser.add_argument("--replace", action="store_true", help="overwrite changed pool files")
    args = parser.parse_args(argv)

    for tool in ("dpkg-deb", "dpkg-scanpackages", "gpg"):
        if shutil.which(tool) is None:
            sys.exit(f"repo: {tool} not found")
    if subprocess.run(["gpg", "--list-secret-keys", args.key], capture_output=True).returncode:
        sys.exit(f"repo: no secret key '{args.key}'; cargo make repo-key creates it")
    name, version, _ = cargo_meta()
    debs = args.debs or sorted((ROOT / "dist").glob(f"{name}_{version}-*_*.deb"))
    if not debs:
        sys.exit(f"repo: no dist/{name}_{version}-*.deb; run cargo make dist")

    add_to_pool(debs, args.repo, args.replace)
    files = write_indexes(args.repo, args.suite)
    (args.repo / "dists" / args.suite / "Release").write_text(
        release_text(args.repo, args.suite, files, time.time()))
    public = sign(args.repo, args.suite, args.key)
    if args.url:
        (args.repo / "medias.sources").write_text(sources_text(args.url.rstrip("/"), args.suite, public))
    else:
        print("repo: no --url/MEDIAS_REPO_URL; medias.sources not written")
    if args.repo.parent == ROOT / "dist":
        import index
        index.write(args.repo.parent, args.url)
    print(f"repo: {args.repo}")


if __name__ == "__main__":
    try:
        main()
    except subprocess.CalledProcessError as error:
        sys.exit(f"repo: {error.cmd[0]} failed (exit {error.returncode})")