大きなサイズのファイルをGitで上手く管理する方法について考える・続

    conf = parse_config(config_file)
    rclone_config_label = conf["storebin"]["rclone_config_label"]

    remote_location = f"{rclone_config_label}:{object_storage_key}"

        cmd = f"rclone deletefile --verbose {rclone_config_label}:{object_storage_key}"

#! /usr/bin/env python3
import configparser
import hashlib
import os
import re
import shlex
import shutil
import subprocess
import sys

from collections.abc import Mapping
from dataclasses import dataclass

__author__ = "TakesxiSximada"
__version__ = "3"


POINTER_FILE_FORMAT_VERSION = "3"

pointer_file_format = """# LFS POINTER
version https://git-lfs.github.com/spec/v1
oid {pointer.hash_type}:{pointer.oid}
size {pointer.size}

# storebin3 customize
# NOTE: extension is a part of the object storage key. DO NOT EDIT.
storebin_version {pointer_file_format_version}
"""

@dataclass(frozen=True)
class StorebinPointer:
    hash_type: str
    oid: str
    size: int
    extension: str = ""

class StorebinPointerParseError(Exception):
    pass

def build_pointer_file_content(pointer: StorebinPointer) -> str:
    content = pointer_file_format.format(
        pointer=pointer,
        pointer_file_format_version=POINTER_FILE_FORMAT_VERSION,
    )
    if pointer.extension:
        content += f"extension {pointer.extension}\n"
    return content

def parse_pointer_file_content(data: str) -> StorebinPointer:
    versions = re.findall(r"^storebin_version (\d+)$", data, re.M)
    if versions != [POINTER_FILE_FORMAT_VERSION]:
        raise StorebinPointerParseError("incompatible pointer file")

    oids = re.findall(r"^oid ([a-z0-9]+):([0-9a-f]{64})$", data, re.M)
    if len(oids) != 1 or oids[0][0] != "sha256":
        raise StorebinPointerParseError("bad oid")

    sizes = re.findall(r"^size (\d+)$", data, re.M)
    if len(sizes) != 1:
        raise StorebinPointerParseError("bad size")

    extension_lines = re.findall(r"^extension .*$", data, re.M)
    extensions = re.findall(r"^extension (\.[A-Za-z0-9]+)$", data, re.M)
    if len(extension_lines) != len(extensions) or len(extensions) > 1:
        raise StorebinPointerParseError("bad extension")

    return StorebinPointer(
        hash_type=oids[0][0],
        oid=oids[0][1],
        size=int(sizes[0]),
        extension=extensions[0] if extensions else "",
    )


def traverse_config_file(path):
    cur_dir = os.path.abspath(path)
    for i in range(100):
        conf_path = os.path.join(cur_dir, "storebin.conf")
        if os.path.exists(conf_path):
            return conf_path
        cur_dir = os.path.dirname(cur_dir)

def parse_config(path) -> configparser.ConfigParser:
    conf = configparser.ConfigParser()
    conf.read(path)
    return conf

def validate_config(conf: configparser.ConfigParser) -> configparser.ConfigParser:
    # 必須キーの存在確認。無ければKeyErrorを送出
    conf["storebin3"]["rclone_config_label"]
    conf["storebin3"]["object_storage_bucket_name"]
    conf["storebin3"]["object_storage_key_prefix"]
    return conf

def build_object_storage_key(conf: configparser.ConfigParser, oid: str, extension: str) -> str:
    prefix = conf["storebin3"]["object_storage_key_prefix"]
    return f"{prefix}/{oid[:2]}/{oid[2:4]}/{oid}{extension}"

def build_remote_location(conf: configparser.ConfigParser, object_storage_key: str) -> str:
    label = conf["storebin3"]["rclone_config_label"]
    bucket = conf["storebin3"]["object_storage_bucket_name"]
    return f"{label}:{bucket}/{object_storage_key}"

def os_system(line) -> int:
    if os.environ.get("STOREBIN_VERBOSE"):
        print("> ", line, file=sys.stderr)
    return os.waitstatus_to_exitcode(os.system(line))

def get_rclone_flags(environ: Mapping[str, str]) -> str:
    return environ.get("STOREBIN_RCLONE_FLAGS", "")

def print_error(message: str, **fields) -> None:
    print(f"error: {message}", file=sys.stderr)
    width = max((len(key) for key in fields), default=0)
    for key, value in fields.items():
        print(f"    {key.ljust(width)} : {value}", file=sys.stderr)

def rclone_object_size(remote_location: str) -> int | None:
    """リモートのオブジェクトのサイズを返す。存在しなければ None。"""

    argv = [
        "rclone",
        "lsf",
        *shlex.split(get_rclone_flags(os.environ)),
        "--format",
        "s",
        remote_location,
    ]
    if os.environ.get("STOREBIN_VERBOSE"):
        print(">", " ".join(argv), file=sys.stderr)

    proc = subprocess.run(argv, capture_output=True, text=True)
    if proc.returncode in (3, 4):   # directory not found / file not found
        return None

    if proc.returncode != 0:
        print_error(
            "rclone lsf failed",
            remote_location=remote_location,
            exit_code=proc.returncode,
            stderr=proc.stderr.strip(),
        )
        sys.exit(1)

    sizes = proc.stdout.split()
    if not sizes:                   # 空の出力 = 存在しない
        return None
    if len(sizes) != 1:             # 単一オブジェクトを指定したのに複数返る =  想定外
        print_error("unexpected lsf output", remote_location=remote_location,
                    output=proc.stdout.strip())
        sys.exit(1)
    return int(sizes[0])


sub_command = sys.argv[1]
if "put" == sub_command:
    target_file = sys.argv[2]

    pointer_file = f"{target_file}.storebin"
    if os.path.exists(pointer_file):
        print("already exists: ", pointer_file, file=sys.stderr)
        sys.exit(1)

    with open(target_file, "rb") as fp:
        hash_value_obj = hashlib.file_digest(fp, "sha256")
    oid = hash_value_obj.hexdigest()
    file_size = os.path.getsize(target_file)

    # 拡張子はそのまま使う
    extension = os.path.splitext(target_file)[1]
    if not re.fullmatch(r"\.[A-Za-z0-9]+", extension):
        extension = ""  # ただし半角英数じゃない拡張子は無視

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf", file=sys.stderr)
        sys.exit(1)

    conf = validate_config(parse_config(config_file))
    object_storage_key = build_object_storage_key(conf, oid, extension)
    remote_location = build_remote_location(conf, object_storage_key)

    pointer = StorebinPointer(
        hash_type="sha256",
        oid=oid,
        size=file_size,
        extension=extension,
    )
    pointer_file_content = build_pointer_file_content(pointer)
    cmd = f"rclone copyto {get_rclone_flags(os.environ)} {shlex.quote(target_file)} {shlex.quote(remote_location)}"
    rc = os_system(cmd)
    if rc != 0:
        print_error(
            "failed to put",
            target_file=target_file,
            pointer_file=pointer_file,
            oid=oid,
            size=file_size,
            extension=extension or "(none)",
            object_storage_key=object_storage_key,
            remote_location=remote_location,
            command=cmd,
            exit_code=rc,
        )
        sys.exit(rc)

    with open(pointer_file, "w+") as fp:
        fp.write(pointer_file_content)

elif "get" == sub_command:
    pointer_file = sys.argv[2]
    target_file = pointer_file.removesuffix(".storebin")
    if os.path.exists(target_file):
        print("already exists: ", target_file, file=sys.stderr)
        sys.exit(1)

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf", file=sys.stderr)
        sys.exit(1)

    conf = validate_config(parse_config(config_file))

    try:
        with open(pointer_file) as fp:
            pointer = parse_pointer_file_content(fp.read())
    except StorebinPointerParseError as err:
        print_error("failed to parse pointer file", pointer_file=pointer_file, reason=err)
        sys.exit(1)

    object_storage_key = build_object_storage_key(conf, pointer.oid, pointer.extension)
    remote_location = build_remote_location(conf, object_storage_key)

    cmd = f"rclone copyto {get_rclone_flags(os.environ)} --immutable {shlex.quote(remote_location)} {shlex.quote(target_file)}"
    rc = os_system(cmd)
    if rc != 0:
        print_error(
            "failed to get",
            pointer_file=pointer_file,
            target_file=target_file,
            oid=pointer.oid,
            size=pointer.size,
            extension=pointer.extension or "(none)",
            object_storage_key=object_storage_key,
            remote_location=remote_location,
            command=cmd,
            exit_code=rc,
        )
        sys.exit(rc)

    if not os.path.isfile(target_file):
        print_error(
            "file not transferred",
            pointer_file=pointer_file,
            target_file=target_file,
            oid=pointer.oid,
            object_storage_key=object_storage_key,
            remote_location=remote_location,
            command=cmd,
            exit_code=rc,
        )
        sys.exit(1)


    with open(target_file, "rb") as fp:  # ファイルちゃんとダウンロードされて作られたか検証する
        actual_oid = hashlib.file_digest(fp, "sha256").hexdigest()

    if actual_oid != pointer.oid:
        print_error(
            "checksum mismatch",
            pointer_file=pointer_file,
            target_file=target_file,
            expected=pointer.oid,
            actual=actual_oid,
        )
        sys.exit(1)

elif "delete" == sub_command:
    pointer_file = sys.argv[2]
    target_file = pointer_file.removesuffix(".storebin")

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf", file=sys.stderr)
        sys.exit(1)

    conf = validate_config(parse_config(config_file))

    try:
        with open(pointer_file) as fp:
            pointer = parse_pointer_file_content(fp.read())
    except StorebinPointerParseError as err:
        print_error("failed to parse pointer file", pointer_file=pointer_file, reason=err)
        sys.exit(1)

    object_storage_key = build_object_storage_key(conf, pointer.oid, pointer.extension)
    remote_location = build_remote_location(conf, object_storage_key)

    cmd = f"rclone deletefile {get_rclone_flags(os.environ)} {shlex.quote(remote_location)}"
    rc = os_system(cmd)
    if rc != 0:
        print_error(
            "failed to delete",
            pointer_file=pointer_file,
            oid=pointer.oid,
            size=pointer.size,
            extension=pointer.extension or "(none)",
            object_storage_key=object_storage_key,
            remote_location=remote_location,
            command=cmd,
            exit_code=rc,
        )
        sys.exit(rc)

    shutil.move(pointer_file,  pointer_file + ".deleted")

elif "verify" == sub_command:
    pointer_files = sys.argv[2:]
    if not pointer_files:
        print("no pointer file", file=sys.stderr)
        sys.exit(1)

    problem_count = 0
    seen = {}

    for pointer_file in pointer_files:
        if not pointer_file.endswith(".storebin"):
            print(f"not a pointer file: {pointer_file}")
            problem_count += 1
            continue

        config_file = traverse_config_file(pointer_file)
        if not config_file:
            print(f"no config file: {pointer_file}", file=sys.stderr)
            problem_count += 1
            continue

        try:
            conf = validate_config(parse_config(config_file))
        except Exception:
            print(f"bad config file: {config_file}", file=sys.stderr)
            problem_count += 1
            continue

        try:
            with open(pointer_file) as fp:
                pointer = parse_pointer_file_content(fp.read())
        except StorebinPointerParseError as err:
            print(f"parse error: {pointer_file}: {err}")
            problem_count += 1
            continue

        object_storage_key = build_object_storage_key(
            conf, pointer.oid, pointer.extension)
        remote_location = build_remote_location(conf, object_storage_key)
        seen.setdefault(remote_location, []).append(pointer_file)

        actual_size = rclone_object_size(remote_location)
        if actual_size is None:
            print(f"missing: {pointer_file}: {object_storage_key}")
            problem_count += 1
            continue

        if actual_size != pointer.size:
            print(f"size mismatch: {pointer_file}: expected={pointer.size} actual={actual_size}")
            problem_count += 1

    for remote_location, files in seen.items():
        if len(files) > 1:
            print(f"shared object: {', '.join(files)}")
            problem_count += 1

    if problem_count:
        sys.exit(1)

elif "selftest" == sub_command:
    test_file = f"storebin3-selftest-{os.getpid()}.bin"
    pointer_file = f"{test_file}.storebin"
    deleted_pointer_file = f"{pointer_file}.deleted"

    # 内容が毎回異なるので oid が一意になり、既存のオブジェクトに影響しない
    with open(test_file, "wb") as fp:
        for _ in range(16):
            fp.write(os.urandom(1024 * 1024))

    # multipart を必ず通す(5MiB チャンク × 4パート)
    os.environ["STOREBIN_RCLONE_FLAGS"] = (
        "--s3-upload-cutoff 5Mi --s3-chunk-size 5Mi --multi-thread-cutoff 5Mi")

    self_file = shlex.quote(sys.argv[0])
    steps = [
        ("put", f"{self_file} put {shlex.quote(test_file)}"),
        ("verify", f"{self_file} verify {shlex.quote(pointer_file)}"),
        ("remove local file", f"rm -f {shlex.quote(test_file)}"),
        ("get", f"{self_file} get {shlex.quote(pointer_file)}"),
        ("delete", f"{self_file} delete {shlex.quote(pointer_file)}"),
    ]

    for step, cmd in steps:
        rc = os_system(cmd)
        if rc != 0:
            print_error(
                "selftest failed",
                step=step,
                test_file=test_file,
                pointer_file=pointer_file,
                command=cmd,
                exit_code=rc,
                note="not cleaned up for investigation",
            )
            sys.exit(1)

    os.remove(test_file)
    os.remove(deleted_pointer_file)

else:
    print("not support subcommand")
    sys.exit(1)