Implementacja pliku zip / tar Git przed zatwierdzeniem i po pobraniu
Regularnie używam narzędzia (Amesim), które pakuje swoje pliki w nieskompresowany plik tar. W przypadku wersjonowania zwykle nazwałem pliki jako file1_Rev01.ame i iterowałem ze zmianami. Działa to, gdy jestem jedynym użytkownikiem, ale ostatnio częściej udostępniam pliki / modele. Próba udostępnienia tych modeli jest bolesna, często zawierają one wyniki, które są dość duże (gbs danych) i śledzenie zmian między wersjami, jeśli jest to trudne, chyba że rygorystyczne dodawanie tekstu do modelu przy każdej zmianie. (Amesim to narzędzie takie jak Simulink.)
Czytałem o hakach git i filtrach git, ale nie jestem pewien, co zrobić, aby lepiej zarządzać wersjami archiwum tar.
Powiedzmy, że mam plik „my_file.tar” i składa się on z a.txt, b.model, c.data i d.results.
Ze strony aplikacji wystawiłbym plik „my_file.tar” i wysłał zatwierdzenie „Aktualizacje modelu”. Bez zmian w git, śledzi zmiany w pliku binarnym. To jest nieczytelne i zajmuje dużo miejsca. Jeśli uwzględniono wyniki, plik jest dość duży. Klonowanie repozytorium będzie trudne, jeśli wyniki są stale przechowywane.
W pierwszej próbie próbowałem użyć haczyków przed zatwierdzeniem i po zakupie.
Po zatwierdzeniu mój punkt przechwytujący przed zatwierdzeniem usuwa „my_file.tar” do katalogu „my_file_tar”. Usuwa plik * .results, który pochodzi z uruchomienia modelu. Nie ma potrzeby śledzenia tego i oszczędza znaczną przestrzeń (GB).
Kiedy wyciągnę model, po zakończeniu zakupu wyszukuje wszystkie foldery z _tar i taruje je, zmieniając ich nazwę na my_file.tar.
Teraz ogólnie to działa. Ale jak mam obsłużyć plik my_file.tar i nieskompresowany folder? Jeśli automatycznie usunę nieskompresowany folder po wyewidencjonowaniu, git stwierdza, że mam istotne zmiany do śledzenia. Czy za każdym razem muszę dodawać / usuwać folder do .gitignore? Ponadto plik tar nigdy nie pokaże, że jest śledzony, ponieważ usunąłem go w kodzie przed zatwierdzeniem. Co mogę zrobić, aby uporządkować ten proces? Jak mam sobie z tym radzić inaczej?
Bibliografia:
- Pliki ZIP Git
- Rozmażyć i wyczyścić
- Dokumentacja Git Office
- Zippey
- XLTrail
W przypadku tego kodu plik .ame jest plikiem tar.
wstępne zatwierdzenie
#!/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)
i po wymeldowaniu
#!/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)
Odpowiedzi
Kod w tej odpowiedzi implementuje filtr git w przeciwieństwie do punktu zaczepienia przed zatwierdzeniem i zaczepu po sprawdzeniu w pytaniu. Zaletą filtra jest to, że obsługuje on tylko jeden plik. Dodatkowe pliki nie muszą być oddzielnie śledzone i zatwierdzane / pobierane. Raczej, podobnie jak Zippey, tworzy nieskompresowany strumień danych i usuwa niepotrzebne pliki po drodze.
Uwaga: Nie używaj instrukcji print, ponieważ miesza to ze strumieniem stdout w filtrze git. To była bolesna lekcja.
Uwaga: zakończenia CRLF i LF stanowią problem. Podczas dekodowania z pierwszego git pull musiałem wyczyścić zakończenia linii, ponieważ Sourcetree / Git przekonwertował na format Windows.
Dyskusja rozwiązania:
Ponieważ plik, z którym pracuję, jest nieskompresowanym plikiem tar, rozwiązanie Zippey nie miało bezpośredniego zastosowania. Zippey jest przeznaczony tylko dla plików zip. Zamiast tego zaimplementowałem technikę Zippeya z plikami tar.
Po zatwierdzeniu stosowany jest czysty filtr, który „koduje” plik tar. Funkcja kodowania pobiera każdy plik i rejestruje długość danych, nieprzetworzoną długość danych, jeśli są binarne, tryb przechowywania (ascii lub binarny) oraz nazwę pliku.
Skrypt kodujący przesyła strumieniowo wszystkie pliki do jednego pliku o tej samej nazwie w nieskompresowanym formacie. Pliki binarne są zakodowane w base64 w jednej linii, dzięki czemu różnice są łatwiejsze do odczytania.
Podczas kodowania unika się plików o określonych rozszerzeniach (takich jak pliki wyników).
Podczas ściągania filtr smug dekompresuje plik, wykorzystując cztery metatagi do odczytywania informacji. Każdy plik jest przetwarzany i dodawany do obiektu pliku tar, a na końcu wypisywany jest plik tar.
Podobnie jak Zippey, ze świeżego klonu repozytorium pobierany jest zakodowany plik, który jest nieczytelny dla mojego narzędzia. Dlatego Clone Setup szuka moich plików * .ame, które są zakodowane i dekoduje je, a także ustawia odpowiednie filtry git.
Ponieważ pracuję zarówno na maszynach z systemem Linux, jak i Windows, a git ma tendencję do dodawania CRLF przy pobieraniu, skrypty upewniają się, że usuną CRLF przed kodowaniem i usuwają CRLF z zakodowanych plików przed dekodowaniem.
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")