Git git zip / tar file di implementazione pre-commit e post-checkout hook
Uso regolarmente uno strumento (Amesim) che impacchetta i suoi file in un file tar non compresso. Per il controllo delle versioni, in genere ho denominato i file come file1_Rev01.ame e ho ripetuto le modifiche. Funziona quando sono l'unico utente, ma ultimamente condivido file / modelli più regolarmente. Cercare di condividere questi modelli è doloroso, spesso includono risultati piuttosto grandi (gbs di dati) e tenere traccia delle modifiche tra le versioni, se difficile, a meno che non si aggiunga rigorosamente del testo all'interno del modello ad ogni modifica. (Amesim è uno strumento come Simulink.)
Ho letto su git hook e filtri git, ma non sono sicuro di cosa fare per gestire meglio il controllo delle versioni di un tarball.
Diciamo che ho il file "my_file.tar" e consiste in a.txt, b.model, c.data e d.results.
Dal lato dell'applicazione, metterei in scena "my_file.tar" e invierei un commit "Aggiornamenti al modello". Senza modifiche a git, tiene traccia delle modifiche a un file binario. Non è leggibile e occupa uno spazio significativo. Se i risultati sono inclusi, il file è abbastanza grande. La clonazione del repository sarà difficile se i risultati vengono continuamente archiviati.
Per il mio primo tentativo, ho provato a utilizzare hook pre-commit e post-checkout.
Al momento del commit, il mio hook pre-commit decomprime "my_file.tar" in una directory "my_file_tar". Rimuove il file * .results che proviene dall'esecuzione del modello. Non è necessario tenerne traccia e consente di risparmiare spazio significativo (gbs).
Quando estraggo il modello, il post-checkout cercherà tutte le cartelle con _tar e le tarerà, rinominandole in my_file.tar.
Ora generalmente funziona. Ma come dovrei gestire my_file.tar e la cartella non compressa? Se elimino automaticamente la cartella non compressa dopo il check-out, git afferma che devo tenere traccia di modifiche significative. Devo aggiungere / rimuovere la cartella in .gitignore ogni volta? Inoltre, il file tar non mostrerà mai di essere tracciato, perché l'ho rimosso nel codice pre-commit. Cosa posso fare per ripulire questo processo? Come dovrei gestirlo diversamente?
Riferimenti:
- File Zip Git
- Sfuma e pulisci
- Git Office Docs
- Zippey
- XLTrail
Per questo codice, il .ame è un file tar.
pre-commit
#!/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)
e dopo il checkout
#!/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)
Risposte
Il codice in questa risposta implementa un filtro git invece di un hook pre-commit e un hook post-checkout nella domanda. Il vantaggio del filtro è che manipola solo un file. Non è necessario tenere traccia di file aggiuntivi e salvarli / estrarli separatamente. Piuttosto, come Zippey, crea un flusso di dati non compresso e rimuove i file non necessari lungo il percorso.
Nota: non utilizzare istruzioni print, poiché interferisce con il flusso stdout nel filtro git. Questa è stata una lezione dolorosa.
Nota: le terminazioni CRLF e LF sono un problema. Durante la decodifica dal primo pull git, ho dovuto pulire le terminazioni di riga perché Sourcetree / Git è stato convertito in formato Windows.
Discussione sulla soluzione:
Poiché il file con cui sto lavorando è un tar non compresso, la soluzione Zippey non è stata applicata direttamente. Zippey è solo per i file zip. Invece ho implementato la tecnica di Zippey con i file tar.
Al momento del commit, viene applicato un filer pulito che "codifica" il file tar. La funzione di codifica prende ogni file e registra la lunghezza dei dati, la lunghezza non elaborata dei dati se binari, la modalità di memorizzazione (ascii o binaria) e il nome del file.
Lo script di codifica trasmette tutti i file in un unico file con lo stesso nome in un formato non compresso. I file binari sono codificati in base64 in una singola riga, rendendo le differenze più facili da leggere.
Durante la codifica, i file di estensioni specifiche vengono evitati (come i file dei risultati).
Al momento del pull, un filtro smudge decomprime il file utilizzando i quattro meta tag per leggere le informazioni. Ogni file viene elaborato e aggiunto a un oggetto file tar, e alla fine viene scritto un file tar.
Come Zippey, su un nuovo clone del repository viene estratto un file codificato che è illeggibile per il mio strumento. Quindi Clone Setup cerca i miei file * .ame che sono codificati e li decodifica, oltre a impostare filtri git appropriati.
Dato che lavoro su macchine Linux e Windows e git ha la tendenza ad aggiungere CRLF al momento del checkout, gli script si assicurano di rimuovere CRLF prima della codifica e di rimuovere CRLF dai file codificati prima della decodifica.
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")