Git zip / tar 파일 구현 사전 커밋 및 사후 체크 아웃 후크

Dec 22 2020

저는 압축되지 않은 tar 파일로 파일을 패키징하는 도구 (Amesim)를 정기적으로 사용합니다. 버전 관리를 위해 일반적으로 파일 이름을 file1_Rev01.ame으로 지정하고 변경 사항을 반복합니다. 이것은 내가 유일한 사용자 일 때 작동하지만 최근에는 파일 / 모델을 더 정기적으로 공유하고 있습니다. 이러한 모델을 공유하려는 시도는 고통스럽고 매우 큰 결과 (데이터 GB)를 포함하고 변경 될 때마다 모델 내에 텍스트를 엄격하게 추가하지 않는 한 어려운 경우 버전 간의 변경 사항을 추적합니다. (Amesim은 Simulink와 같은 도구입니다.)

git hooks 및 git filters에 대해 읽어 봤지만 tarball의 버전 관리를 더 잘 관리하려면 어떻게해야할지 모르겠습니다.

"my_file.tar"파일이 있고 a.txt, b.model, c.data 및 d.results로 구성되어 있다고 가정 해 보겠습니다.

애플리케이션 측면에서 "my_file.tar"를 준비하고 "모델에 대한 업데이트"커밋을 제출합니다. git을 변경하지 않고 바이너리 파일의 변경 사항을 추적합니다. 이것은 읽을 수 없으며 상당한 공간을 소비합니다. 결과가 포함 된 경우 파일이 상당히 큽니다. 결과가 지속적으로 저장되면 리포지토리를 복제하는 것이 어려울 수 있습니다.

첫 번째 시도에서는 사전 커밋 및 사후 체크 아웃 후크를 사용하려고했습니다.

커밋시, 내 사전 커밋 후크는 "my_file.tar"를 "my_file_tar"디렉토리로 풉니 다. 모델 실행에서 오는 * .results 파일을 제거합니다. 이를 추적 할 필요가 없으며 상당한 공간 (gbs)을 절약 할 수 있습니다.

모델을 가져 오면 체크 아웃 후 _tar가 포함 된 폴더를 검색하고 tar로 이름을 바꾸고 my_file.tar로 이름을 바꿉니다.

이제 일반적으로 작동합니다. 그러나 my_file.tar 및 압축되지 않은 폴더는 어떻게 처리해야합니까? 체크 아웃 후 압축되지 않은 폴더를 자동 삭제하면 git은 추적해야 할 중요한 변경 사항이 있다고 말합니다. 매번 .gitignore에 폴더를 추가 / 제거해야합니까? 또한 tar 파일은 사전 커밋 코드에서 제거했기 때문에 추적되었음을 표시하지 않습니다. 이 프로세스를 정리하려면 어떻게해야합니까? 이것을 어떻게 다르게 처리해야합니까?

참조 :

  • Git Zip 파일
  • 얼룩 및 청소
  • Git Office 문서
  • Zippey
  • XLTrail

이 코드에서 .ame은 tar 파일입니다.

사전 커밋

#!/usr/bin/env python

import argparse
import os
import tarfile
import zipfile
import subprocess

def parse_args():
    pass

def log_file(log_item):
    cwd = os.getcwd()
    file = open("MyFile.txt", "a") # Open file in append mode
    file.write(log_item + '\n')
    return 1
    
def get_staged_ame_files():
    '''Request a list of staged files from git and return a list of *.ame files

    This function opens a subprocess with git, requests a list of names in the git staged list. It will return a list of strings.
    '''
    out = subprocess.Popen(['git', 'diff', '--staged', '--name-only'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    stdout, stderr = out.communicate()
    # Separate output by newlines
    # staged_files = stdout.split(b'\n') # split as bytes
    
    # filter for files with .ame 
    staged_files = stdout.decode('utf-8').split('\n') # split as strings
    # Create list of *just* amesim files
    staged_ame_files = []
    for entry in staged_files:
        if entry.endswith(".ame"):
            staged_ame_files.append(entry)
    
    if not staged_ame_files:
        return None
    else:
        return staged_ame_files

def extract_ame_files(file_list):
    folder_list = []
    for list_item in file_list:
        # If file exists, extract it. Else continue.
        if os.path.isfile(list_item):
            tar = tarfile.open(list_item, "r:")
            folder_name = list_item[0:-4] + "_ame"
            folder_list.append(folder_name)
            tar.extractall(path = folder_name)
            tar.close()
            log_file(folder_name)
        else:
            print("File {} does not exist.".format(list_item))
            
    return folder_list
    

def cleanup_ame_ignored_files(folder_list):
    '''Removes unecessary files from the folder. 
    
    '''
    for folder in folder_list:
        file_list = os.listdir(folder)
        for file in file_list:
            if item.endswith(".results"):
                os.remove(item)
            if item.endswith(".exe"):
                os.remove(item)
    return 1


def git_add_ame_folders(folders):
    # Add *_ame folders to git stage
    for folder in folders:
        out = subprocess.Popen(['git', 'add', folder + '/'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
        # The -u will capture removed files?
        out = subprocess.Popen(['git', 'add', '-u', folder + '/'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
        
        log_file(stdout.decode('utf-8'))
    return 1
    
def remove_ame_from_staging(file_list):
    # Loop through any staged ame files.
    for file in file_list:
        out = subprocess.Popen(['git', 'rm', '--cached', file], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
    return 1

def main(args=None):
    # if file name is *.ame
    # extract *.ame as a tar of the same name into a folder of the same name + _ame
    # delete .results file
    # don't commit .ame file 
    
    # Search for files we want to process in the staged list
    # These will only be *.ame files.
    staged_ame_files = get_staged_ame_files()
    if not staged_ame_files:
        # If its empty, there's nothing to do. End the function.
        return 0
    
    # We're not empty, lets extract each one.
    folder_list = extract_ame_files(staged_ame_files)
    
    # Delete all .results files in each extracted folder  path
    
    # Stage all files in each folder path 
    git_add_ame_folders(folder_list)
    
    # Unstage the .ame file
    remove_ame_from_staging(staged_ame_files)
    return 1

if __name__ == "__main__":
    args = parse_args()
    main(args)

및 체크 아웃 후

#!/usr/bin/env python

import argparse
import os
import tarfile
import zipfile
import subprocess
import shutil
#from shutil import rmtree # Delete directory trees

def parse_args():
    pass

def log_file(log_item):
    cwd = os.getcwd()
    file = open("MyFile2.txt", "a") # Open file in append mode
    file.write(log_item + '\n')
    return 1
    
def compress_ame_files(folder_list):
    for list_item in folder_list:
        log_file("We're on item {}".format(list_item))
        file_name = list_item[0:-4] + ".ame"
        log_file("Tar file name {}".format(file_name))
        # Delete the file if it exists first.
        os.remove(file_name)
        with tarfile.open(file_name, "w:") as tar:
            tar.add(list_item, arcname=os.path.basename('../'))
    return 1
    

def cleanup_ame_ignored_files(folder_list):
    '''Removes unecessary files from the folder. 
    
    '''
    for folder in folder_list:
        file_list = os.listdir(folder)
        for file in file_list:
            if item.endswith(".results"):
                os.remove(item)
            if item.endswith(".exe"):
                os.remove(item)
    return 1


def git_add_ame_folders(folders):
    # Add *_ame folders to git stage
    for folder in folders:
        out = subprocess.Popen(['git', 'add', folder + '/'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
        # The -u will capture removed files?
        out = subprocess.Popen(['git', 'add', '-u', folder + '/'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
        
        #log_file(stdout.decode('utf-8'))
    return 1
    
def remove_ame_from_staging(file_list):
    # Loop through any staged ame files.
    for file in file_list:
        out = subprocess.Popen(['git', 'rm', '--cached', file], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        stdout, stderr = out.communicate()
    return 1

def fast_scandir(dirname):
    # https://stackoverflow.com/questions/973473/getting-a-list-of-all-subdirectories-in-the-current-directory?rq=1
    subfolders= [f.path for f in os.scandir(dirname) if f.is_dir()]
    for dirname in list(subfolders):
        subfolders.extend(fast_scandir(dirname))
    return subfolders

def delete_ame_folders(folders):
    for folder in folders:
        try:
            shutil.rmtree(folder)
        except OSError as e:
            print("Error: %s : %s" % (dir_path, e.strerror))
    return 1
    
#def main(args=None):
def main(lines):
    print("Post checkout running.")
    # find folders with the name _ame
    #log_file("We're running.")
    folder_list = []
    for folder in fast_scandir(os.getcwd()):
        if folder.endswith("_ame"):
            #log_file("Found folder {}.".format(folder))
            folder_list.append(os.path.join(os.getcwd(), folder))
    # tar each folder up and rename with .ame
    compress_ame_files(folder_list)
    
    # Delete the folders
    #delete_ame_folders(folder_list)

    return 1

if __name__ == "__main__":
    args = parse_args()
    main(args)

답변

3 Rukie Dec 31 2020 at 00:46

이 답변의 코드는 질문의 사전 커밋 후크 및 사후 체크 아웃 후크와는 반대로 git 필터를 구현합니다. 필터의 장점은 하나의 파일 만 조작한다는 것입니다. 추가 파일을 별도로 추적하고 커밋 / 풀링 할 필요가 없습니다. 오히려 Zippey와 마찬가지로 압축되지 않은 데이터 스트림을 생성하고 그 과정에서 불필요한 파일을 제거합니다.

참고 : git 필터에서 stdout 스트림을 엉망으로 만들기 때문에 print 문을 사용하지 마십시오. 이것은 고통스러운 교훈이었습니다.

참고 : CRLF 및 LF 엔딩이 문제입니다. 첫 번째 git pull에서 디코딩 할 때 Sourcetree / Git이 Windows 형식으로 변환 되었기 때문에 줄 끝을 정리해야했습니다.

솔루션 논의 :

내가 작업중인 파일이 압축되지 않은 tar이기 때문에 Zippey 솔루션은 직접 적용되지 않았습니다. Zippey는 zip 파일 전용입니다. 대신 tar 파일로 Zippey의 기술을 구현했습니다.

커밋시 tar 파일을 '인코딩'하는 깨끗한 파일러가 적용됩니다. 인코딩 기능은 각 파일을 가져와 데이터 길이, 바이너리 인 경우 데이터의 원시 길이, 저장 모드 (ascii 또는 바이너리) 및 파일 이름을 기록합니다.

인코딩 스크립트는 모든 파일을 압축되지 않은 형식으로 동일한 이름의 단일 파일로 스트리밍합니다. 바이너리 파일은 base64로 한 줄로 인코딩되어 diff를 더 쉽게 읽을 수 있습니다.

인코딩하는 동안 특정 확장자의 파일 (예 : 결과 파일)이 사용되지 않습니다.

풀 때 얼룩 필터는 4 개의 메타 태그를 사용하여 정보를 읽어 파일을 압축 해제합니다. 각 파일이 처리되어 tar 파일 객체에 추가되고 마지막에 tar 파일이 기록됩니다.

Zippey와 마찬가지로 저장소의 새 복제본에서 인코딩 된 파일이 내 도구에서 읽을 수 없습니다. 따라서 Clone Setup은 인코딩 된 * .ame 파일을 찾아 디코딩하고 적절한 git 필터를 설정합니다.

내가 Linux와 Windows 시스템에서 작업하고 git은 체크 아웃시 CRLF를 추가하는 경향이 있으므로 스크립트는 인코딩 전에 CRLF를 제거하고 디코딩하기 전에 인코딩 된 파일에서 CRLF를 제거합니다.

amefilter.py

import tarfile
import sys
import io
import base64
import string
import tempfile
import os.path

DEBUG_AME_FILTER = False
NAME = 'Amesim_Git'
ENCODING = 'UTF-8'

W_EOL = b'\r\n'
U_EOL = b'\n'

# decompress these defined files
AME_EXTENSIONS = ['.amegp', '.cir', '.sad', '.units', '.views', '.xml']
ASCII_EXTENSIONS = ['.txt', '.py']
# Do not include these files in tracking. 
EXCLUDE = ['.results']

def debug(msg):
    '''Print debug message'''
    if DEBUG_AME_FILTER:
        sys.stderr.write('{0}: debug: {1}\n'.format(NAME, msg))

def error(msg):
    '''Print error message'''
    sys.stderr.write('{0}: error: {1}\n'.format(NAME, msg))

def init():
    '''Initialize writing; set binary mode for windows'''
    debug("Running on {}".format(sys.platform))
    if sys.platform.startswith('win'):
        import msvcrt
        debug("Enable Windows binary workaround")
        msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
        msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)

def encode(input, output):
    '''Encode into special VCS friendly format from input to output'''
    debug("ENCODE was called")
    # Create a temporary file based off of the input AME file
    # This lets tarfile access a binary file object
    tfp = tempfile.TemporaryFile(mode='w+b')
    # Write contents into temporary file
    tfp.write(input.read())
    tfp.seek(0)  # Make sure tarfile reads from the start, otherwise object is empty
    tar = tarfile.open(fileobj=tfp, mode='r:')
    # Loop through objects within tar file
    for name in tar.getnames():
        # Get the file name of each object.
        tarinfo = tar.getmember(name)
        if tarinfo.isdir():
            continue # Skip folders, not sure how to handle encode/decode yet.
        data = tar.extractfile(name).read()
        
        # List of ASCII files to decode and version control
        text_extensions = list(set(AME_EXTENSIONS).union(set(ASCII_EXTENSIONS)))
        
        # Isolate extension.
        extension = os.path.splitext(name)[1][1:].strip().lower()
        # Amesim may store batched simulations as *.results.1, *.results.2, remove numeric endings and identify the real ending.
        if extension.isnumeric():
            root_name = os.path.splitext(name)[0][0:]
            real_extension = os.path.splitext(root_name)[1][1:].strip().lower()
            if real_extension in EXCLUDE:
                continue  # Skip excluded extensions
            
        if extension in EXCLUDE:
            continue  # Skip excluded extensions.
            
        # Encode the defined extensions in UTF-8
        try:
            # Check if text data
            data.decode(ENCODING)
            data = data.replace(W_EOL, U_EOL)  # Fix line endings
            try:
                strdata = map(chr, data)
            except TypeError:
                strdata = data
            if extension not in text_extensions and not all(c in string.printable for c in strdata):
                # File is not ascii, append binary file.
                raise UnicodeDecodeError(ENCODING, "".encode(ENCODING), 0, 1, "Artificial exception")

            # Encode
            debug("Appending text file '{}'".format(name))
            mode = 'A'  # ASCII Mode
            output.write("{}|{}|{}|{}\n".format(len(data), len(data), mode, name).encode(ENCODING))
            output.write(data)
            output.write("\n".encode(ENCODING)) # Separation from next meta line
        except UnicodeDecodeError:
            # Binary data
            debug("Appending binary file '{}'".format(name))
            mode = 'B'  # Binary Mode
            raw_len = len(data)
            data = base64.b64encode(data)
            output.write("{}|{}|{}|{}\n".format(len(data), raw_len, mode, name).encode(ENCODING))
            output.write(data)
            output.write("\n".encode(ENCODING))  # Separation from next meta line
    tar.close()

def decode(input, output):
    '''Decode from special VCS friendly format from input to output'''
    debug("DECODE was called")
    tfp = tempfile.TemporaryFile(mode='w+b')
    tar = tarfile.open(fileobj=tfp, mode='w:')
    #input = io.open(input, 'rb')
    while True:
        meta = input.readline().decode(ENCODING)
        if not meta:
            break
        #print(meta)
        (data_len, raw_len, mode, name) = [t(s) for (t, s) in zip((int, int, str, str), meta.split('|'))]
        #print('Data length:{}'.format(data_len))
        #print('Mode: {}'.format(mode))
        #print('Name: {}'.format(name))
        if mode == 'A':
            #print('Appending ascii data')
            debug("Appending text file '{}'".format(name))
            #https://stackoverflow.com/questions/740820/python-write-string-directly-to-tarfile
            info = tarfile.TarInfo(name=name.rstrip())
            info.size = raw_len
            raw_data = input.read(data_len)
            binary_data = io.BytesIO(raw_data)
            # Add each file object to our tarball
            tar.addfile(tarinfo=info, fileobj=binary_data)
            input.read(1) # Skip last '\n'
        elif mode == 'B':
            #print('Appending binary data')
            debug("Appending binary file '{}'".format(name.rstrip()))

            info = tarfile.TarInfo(name=name.rstrip())
            info.size = raw_len
            raw_data = input.read(data_len)
            decoded_data = base64.b64decode(raw_data)
            binary_data = io.BytesIO(decoded_data)
            tar.addfile(tarinfo=info, fileobj=binary_data)
            input.read(1) # Skip last '\n'
        else:
            # Should never reach here
            tar.close()
            tfp.close()
            error('Illegal mode "{}"'.format(mode))
            sys.exit(1)

    # Flush all writes
    tar.close()

    # Write output
    tfp.seek(0) # Go to the start of our temporary file
    output.write(tfp.read())
    tfp.close()

def main():
    '''Main program'''
    #import codecs
    #sys.stdout = codecs.getwriter('utf8')(sys.stdout)
    init()
    input = io.open(sys.stdin.fileno(), 'rb')
    output = io.open(sys.stdout.fileno(), 'wb')
    if len(sys.argv) < 2 or sys.argv[1] == '-' or sys.argv[1] == '--help':
        # This is wrong
        sys.stdout.write("{}\nTo encode: 'python ame_filter.py e'\nTo decode: 'python ame_filter.py d'\nAll files read from stdin and printed to stdout\n".format(NAME))
    elif sys.argv[1] == 'e':
        encode(input, output)
    elif sys.argv[1] == 'd':
        decode(input, output)
    else:
        error("Illegal argument '{}'. Try --help for more information".format(sys.argv[1]))
        sys.exit(1)

        
if __name__ == '__main__':
    main()

Clone_Setup.py

#!/usr/bin/env python
'''
Clone_Setup.py initializes the git environment. 
Each time a new instance of the repository is generated, these commands must 
be run.

'''
import os
import sys
import io
import subprocess
import ame_filter as amef
import tempfile
import shutil

# replacement strings
W_EOL = b'\r\n'
U_EOL = b'\n'

def setup_git():
    os.system("git config filter.ame_filter.smudge \"./ame_filter.py d\"")
    os.system("git config filter.ame_filter.clean \"./ame_filter.py e\"")
    
    '''
        Create .gitattributes programmatically. 
        Add these lines if they do not exist
    '''
    items = ["*.ame filter=ame_filter", "*.ame diff"]
    try:
        with open(".gitattributes", "x") as f:
            for item in items:
                f.write(item + "\n")
    except:
        with open(".gitattributes", "r+") as f:
            for item in items:
                f.seek(0)
                line_found = any(item in line for line in f)
                if not line_found:
                    f.seek(0, os.SEEK_END)
                    f.write("\n" + item)
    
    '''
        Create .gitignore programmatically. 
        Add these lines if they do not exist.
    '''
    items = ["*.gra",
             "*.res",
             "*.req",
             "*.pyc",
             "*.results",
             "*.results.*"
             ]
    
    try:
        with open(".gitignore", "x") as f:
            for item in items:
                f.write(item + "\n")
    except:
        with open(".gitignore", "r+") as f:
            for item in items:
                f.seek(0)
                line_found = any(item in line for line in f)
                if not line_found:
                    f.seek(0, os.SEEK_END)
                    f.write("\n" + item)
        

''' Search for AME files '''
def find_ame_files():
    out = subprocess.Popen(['git', 'ls-files'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    stdout, stderr = out.communicate()
    # Separate output by newlines
    # filter for files with .ame 
    git_files = stdout.decode('utf-8').split('\n') # split as strings
    # Create list of *just* amesim files
    ame_files = [entry for entry in git_files if entry.endswith(".ame")]
    ''' #  Equivalent code block
    ame_files = []
    for entry in git_files:
        if entry.endswith(".ame"):
            ame_files.append(entry)
    '''
    return ame_files

def decode_ame_files(ame_files):
    for file in ame_files:
        input = io.open(file, 'rb')
        tfp = tempfile.TemporaryFile(mode='w+b')
        # Write contents into temporary file
        tfp.write(input.read().replace(W_EOL, U_EOL))
        tfp.seek(0)
        input.close()
        output = io.open(file+'~', 'wb')
        try:
            amef.decode(tfp, output)
            output.close()
            shutil.move(file+'~', file)
        except:
            print("File is already decoded. Returning to normal.")
            output.close()
        finally:
            os.remove(file+'~')
            
            

def main():
    '''Main program'''
    print("Setting up git.")
    setup_git()
    print("Finding ame files.")
    ame_files = find_ame_files()
    print(ame_files)
    print("Decoding ame files.")
    decode_ame_files(ame_files)
    
        
if __name__ == '__main__':
    main()
    # Keep console open to view messages on windows machines.
    if os.name == 'nt':
        input("Press enter to exit")