我经常使用一个工具（Amesim）将其文件打包在未压缩的 tar 文件中。对于版本控制，我通常将文件命名为 file1_Rev01.ame，并迭代更改。当我是唯一的用户时，这是有效的，但最近我更频繁地共享文件/模型。尝试共享这些模型是痛苦的，它们通常包含相当大的结果（GB 数据），并且如果很难跟踪版本之间的更改，除非在每次更改时在模型中严格添加文本。 （Amesim 是一个类似于 Simulink 的工具。）
我一直在阅读 git hooks 和 git 过滤器，但我不确定如何才能更好地管理 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 的文件夹并将其压缩，将它们重命名为 my_file.tar。
现在一般来说这是有效的。但是，我应该如何处理 my_file.tar 和未压缩的文件夹？如果我在签出后自动删除未压缩的文件夹，git 会指出我有重大更改需要跟踪。我是否需要每次都将文件夹添加/删除到 .gitignore ？此外，tar 文件永远不会显示它已被跟踪，因为我在预提交代码中删除了它。我可以做什么来清理这个过程？我应该如何以不同的方式处理这个问题？
参考资料：
对于此代码，.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)
此答案中的代码实现了 git 过滤器，而不是问题中的预提交挂钩和结账后挂钩。过滤器的优点是它只操作一个文件。不需要单独跟踪和提交/拉取其他文件。相反，像 Zippey 一样，它创建未压缩的数据流并删除不必要的文件。
注意：不要使用 print 语句，因为它会扰乱 git 过滤器中的 stdout 流。这是一个惨痛的教训。
注意：CRLF 和 LF 结尾是一个问题。当从第一个 git pull 解码时，我必须清理行结尾，因为 Sourcetree/Git 转换为 Windows 格式。
解决方案讨论：
由于我正在使用的文件是未压缩的 tar，因此 Zippey 解决方案不能直接应用。 Zippey 仅适用于 zip 文件。我用 tar 文件实现了 Zippey 的技术。
提交时，应用一个干净的文件管理器来“编码”tar 文件。编码函数获取每个文件并记录数据的长度、数据的原始长度（如果是二进制）、存储模式（ascii 或二进制）以及文件名。
编码脚本将所有文件以未压缩格式流式传输到同名的单个文件中。二进制文件采用 base64 编码成一行，使差异更易于阅读。
在编码过程中，避免使用特定扩展名的文件（如结果文件）。
拉取时，污迹过滤器通过利用四个元标记读取信息来解压缩文件。每个文件都会被处理并添加到一个 tar 文件对象，最后写出一个 tar 文件。
像 Zippey 一样，在存储库的新克隆上，会提取一个编码文件，而我的工具无法读取该文件。因此，克隆安装程序会查找我的已编码的 *.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()
克隆_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")