0 A Simple Tool for Splitting and Rejoining Data with External Commands.md

A Simple Tool for Splitting and Rejoining Data with External Commands

This Python script provides a small command-line utility that can split a stream of bytes (from standard input) into many chunk files, and concatenate those chunk files back to a single stream (written to standard output). It’s designed to be flexible by letting you define how files are created, saved, and read using your own shell commands.

What the Program Does

• split: reads data from stdin in fixed-size pieces and writes each piece to a new file. It can group files into numbered directories.
• concat: reads those files back in order and writes their contents to stdout.

Both operations delegate the actual filesystem work (mkdir, write, read) to external commands you specify. This makes the tool adaptable to many environments (local disk, special devices, custom commands, etc.).

Key Ideas in the Design

• Templates for names You supply Python format templates for directory and file names, e.g. d{:05d} or f{:05d}. The script fills in the sequence numbers (0, 1, 2, …).

• External command templates You pass shell command templates for:

  • making directories (--mkdir),
  • saving data from stdin (--save),
  • reading data to stdout (--cat).

  Placeholders:

  • {dn} → the directory name
  • {fn} → the file name

• Chunking and grouping You can choose the maximum bytes per file (default: 1 MiB) and the maximum files per directory (default: 1000). When the file count hits the limit, the tool moves to the next directory.

How It Works Internally

• The MyClass object stores templates and wraps three helpers:

  • _do_mkdir(dir_name) runs the mkdir command (if provided).
  • _do_save(dir_name, file_name, data) runs the save command and writes the chunk to the process’s stdin.
  • _do_cat(dir_name, file_name, stdout) runs the read command and pipes its output to the tool’s stdout.

• Split loop Repeatedly read up to max_bytes from stdin. For each non-empty read:

  1. Format the next file name.
  2. Call the save command.
  3. Increment counters; roll over to a new directory when needed.

• Concat loop Iterate file indices in order. For each file:

  1. Format the file name.
  2. Run the cat command and stream it to stdout.
  3. If the command returns non-zero (treated as “no more files”), stop.
  4. Roll over to the next directory after files_per_dir files.

Command-Line Interface

The tool has two subcommands: split and concat.

Split

python3 tool.py split \
  --dir "d{:05d}" \
  --file "f{:05d}" \
  --mkdir "mkdir -p {dn}" \
  --save "dd of={dn}/{fn}" \
  --max-bytes 1048576 \
  --files-per-dir 1000

• Reads from stdin, writes chunk files like d00000/f00000, d00000/f00001, …
• You can swap dd for tee {dn}/{fn} >/dev/null or any writer that reads from stdin.

Example usage:

cat bigfile.bin | python3 tool.py split --mkdir "mkdir -p {dn}" --save "dd of={dn}/{fn}"

Concat

python3 tool.py concat \
  --dir "d{:05d}" \
  --file "f{:05d}" \
  --cat "dd if={dn}/{fn}" \
  --files-per-dir 1000 > rejoined.bin

• Reads files back in order, starting at d00000/f00000, and writes to stdout.

Example usage:

python3 tool.py concat --cat "cat {dn}/{fn}" > restored.bin

Customization Tips

• Different storage backends: Point --save and --cat to commands that interact with special paths, devices, or network tools.
• Name schemes: Use any Python format spec like chunks_{0:04d} or part_{0:06d}.bin to match your conventions.
• Directory strategy: Adjust --files-per-dir to avoid huge directories on filesystems that slow down with many entries.

Safety and Caveats

• Return codes matter: The tool treats any non-zero exit code during concat as “no more files.” Ensure your --cat command behaves accordingly.
• Defaults review: The default --mkdir in the script you pasted is mkdir -p {dn}/{fn} which tries to create a directory path including the file name. You’ll likely want mkdir -p {dn} instead.
• Atomicity and retries: There’s no retry logic; failed commands raise errors.
• No integrity checks: Consider adding checksums or verification if data correctness is critical.

When to Use This Tool

• Splitting huge files for storage/transfer limits.
• Streaming pipelines where chunking and reassembly must integrate with existing shell tools.
• Environments that require custom IO commands (e.g., dd, device files, or specialized CLI clients).

In short, this script offers a small, adaptable “glue” layer: you decide the file layout and the exact commands, and it handles the counting, chunking, and ordering.

main.py

#!/usr/bin/env python3
import sys
import subprocess
import shlex
import argparse
from typing import Optional
import time

MAX_BYTES_PER_FILE = 128 * 1024 * 1024
FILES_PER_DIR = 1000


class MyClass:
    def __init__(self):
        # Templates are set when subcommands are executed
        self.dir_name_template: Optional[str] = None
        self.file_name_template: Optional[str] = None
        self.mkdir_command_template = None
        self.save_command_template = None
        self.cat_command_template = None

    def _do_mkdir(self, dir_name: str):
        if not self.mkdir_command_template:
            # Do nothing if no mkdir command is specified (optional)
            return
        args = tuple(arg.format(dn=dir_name) for arg in self.mkdir_command_template)
        proc = subprocess.Popen(args)
        ret = proc.wait()
        if ret != 0:
            raise subprocess.CalledProcessError(ret, proc.args)

    def _do_save(self, dir_name: str, file_name: str, data: bytes):
        args = tuple(arg.format(fn=file_name, dn=dir_name) for arg in self.save_command_template)
        for try_num in range(5):
            proc = subprocess.Popen(args, stdin=subprocess.PIPE)
            assert proc.stdin is not None
            try:
                proc.stdin.write(data)
            except BrokenPipeError as e:
                sys.stderr.write(f"warn: {e=}\n")
                proc.kill()
                proc.wait()
                time.sleep(60)
                continue
            proc.stdin.close()
            ret = proc.wait()
            if ret != 0:
                sys.stderr.write(f"warn: {proc.args=} {ret=} sleep and retry\n")
                time.sleep(60)
                continue
                # raise subprocess.CalledProcessError(ret, proc.args)
            elif ret == 0:
                break

    def _do_cat(self, dir_name: str, file_name: str, stdout=None):
        args = tuple(arg.format(fn=file_name, dn=dir_name) for arg in self.cat_command_template)
        for try_num in range(5):
            proc = subprocess.Popen(args, stdout=subprocess.PIPE)
            data = proc.stdout.read()
            ret = proc.wait()
            if ret == 0:
                stdout.write(data)
                return
            else:
                sys.stderr.write(f"warn: {proc.args=} {ret=} sleep and retry\n")
                time.sleep(60)
                continue

        raise subprocess.CalledProcessError(ret, proc.args)

    # ---- Subcommand implementations ----
    def do_split(self, dir_tpl: str, file_tpl: str, mkdir_cmd: str, save_cmd: str,
                 max_bytes: int = MAX_BYTES_PER_FILE, files_per_dir: int = FILES_PER_DIR):
        self.dir_name_template = dir_tpl
        self.file_name_template = file_tpl
        self.mkdir_command_template = shlex.split(mkdir_cmd) if mkdir_cmd else None
        self.save_command_template = shlex.split(save_cmd)

        file_index = 0
        dir_index = 0

        dir_name = self.dir_name_template.format(dir_index)
        self._do_mkdir(dir_name)

        while True:
            data = sys.stdin.buffer.read(max_bytes)
            if not data:
                break

            file_name = self.file_name_template.format(file_index)
            self._do_save(dir_name, file_name, data)

            file_index += 1
            if file_index == files_per_dir:
                file_index = 0
                dir_index += 1
                dir_name = self.dir_name_template.format(dir_index)
                self._do_mkdir(dir_name)

    def do_concat(self, dir_tpl: str, file_tpl: str, cat_cmd: str,
                  files_per_dir: int = FILES_PER_DIR):
        self.dir_name_template = dir_tpl
        self.file_name_template = file_tpl
        self.cat_command_template = shlex.split(cat_cmd)

        file_index = 0
        dir_index = 0
        dir_name = self.dir_name_template.format(dir_index)

        while True:
            file_name = self.file_name_template.format(file_index)
            self._do_cat(dir_name, file_name, stdout=sys.stdout.buffer)

            file_index += 1
            if file_index == files_per_dir:
                file_index = 0
                dir_index += 1
                dir_name = self.dir_name_template.format(dir_index)


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Split stdin into chunk files or concatenate chunk files to stdout using external commands."
    )
    sub = p.add_subparsers(dest="command", required=True)

    # split
    ps = sub.add_parser("split", help="Split stdin and save")
    ps.add_argument("--dir", "-d", help="Format template for directory name (e.g., chunks_{0:04d})", dest="dir_tpl", default="d{:05d}")
    ps.add_argument("--file", "-f", help="Format template for file name (e.g., part_{0:06d}.bin)", dest="file_tpl", default="f{:05d}")
    ps.add_argument("--mkdir", help="Directory creation command (supports format: {dn}), e.g., 'mkdir -p {dn}'", dest="mkdir_cmd", default="mkdir -p {dn}/{fn}")
    ps.add_argument("--save", help="Save command (reads from stdin, supports format: {dn} {fn}), e.g., 'tee {dn}/{fn} >/dev/null'", dest="save_cmd", default="dd of={dn}/{fn}")
    ps.add_argument("--max-bytes", type=int, default=MAX_BYTES_PER_FILE, help=f"Maximum bytes per chunk (default {MAX_BYTES_PER_FILE})")
    ps.add_argument("--files-per-dir", type=int, default=FILES_PER_DIR, help=f"Maximum number of files per directory (default {FILES_PER_DIR})")

    # concat
    pc = sub.add_parser("concat", help="Concatenate split files to stdout")
    pc.add_argument("--dir", "-d", help="Format template for directory name (e.g., chunks_{0:04d})", dest="dir_tpl", default="d{:05d}")
    pc.add_argument("--file", "-f", help="Format template for file name (e.g., part_{0:06d}.bin)", dest="file_tpl", default="f{:05d}")
    pc.add_argument("--cat", help="Output command (writes to stdout, supports format: {dn} {fn}), e.g., 'cat {dn}/{fn}'", dest="cat_cmd", default="dd if={dn}/{fn}")
    pc.add_argument("--files-per-dir", type=int, default=FILES_PER_DIR, help=f"Maximum number of files per directory (default {FILES_PER_DIR})")

    return p


def main():
    parser = build_parser()
    args = parser.parse_args()
    worker = MyClass()

    if args.command == "split":
        worker.do_split(
            dir_tpl=args.dir_tpl,
            file_tpl=args.file_tpl,
            mkdir_cmd=args.mkdir_cmd,
            save_cmd=args.save_cmd,
            max_bytes=args.max_bytes,
            files_per_dir=args.files_per_dir,
        )
    elif args.command == "concat":
        worker.do_concat(
            dir_tpl=args.dir_tpl,
            file_tpl=args.file_tpl,
            cat_cmd=args.cat_cmd,
            files_per_dir=args.files_per_dir,
        )
    else:
        parser.error("unknown command")


if __name__ == "__main__":
    main()