Skip to content

Instantly share code, notes, and snippets.

@billfitzgerald
Last active July 15, 2024 20:04
Show Gist options
  • Save billfitzgerald/a6920c582049e346c75f0a61939c61b8 to your computer and use it in GitHub Desktop.
Save billfitzgerald/a6920c582049e346c75f0a61939c61b8 to your computer and use it in GitHub Desktop.
Short script to move files generated by an automated process. For example, a Raspberry Pi can mount a remote directory vis sshfs and then periodically copy files to a different remote location.
import glob
import os
import time
import shutil
import schedule
import sys
# this script is released into the Public Domain
# use freely, and at your own risk :)
source_dir = './source_files/'
destination_dir = './moved_files/'
repeat_cycle = 3 # number of time units (minutes or seconds, based on test_run)
# True to repeat cycles in seconds for testing;
# False to repeat cycles in minutes in actual use
test_run = True # True or False
# number of files to leave intact in source_dir
# useful when harvesting files created by another operation
file_buffer = 3
def check_dir(dir):
if os.path.isdir(dir) == True:
results = ''
else:
results = f'{dir} does not exist.'
return results
def move_the_files():
# Get list of all files in source directory
list_of_files = filter(os.path.isfile, glob.glob(source_dir + '*') )
# Sort files based on last modification time in ascending order
list_of_files = sorted( list_of_files, key = os.path.getmtime)
if len(list_of_files) > file_buffer + 1:
list_of_files = list_of_files[:-file_buffer]
for file_path in list_of_files:
timestamp_str = time.strftime('%m/%d/%Y :: %H:%M:%S', time.gmtime(os.path.getmtime(file_path)))
file_name = os.path.basename(file_path)
destination = destination_dir + file_name
shutil.move(file_path, destination)
print(f'Moved {file_name}')
else:
print('No files to move.\n')
results = check_dir(source_dir)
results = results + '\n' + check_dir(destination_dir)
if len(results) > 3:
sys.exit(results)
else:
pass
if test_run == True:
schedule.every(repeat_cycle).seconds.do(move_the_files)
elif test_run == False:
schedule.every(repeat_cycle).minutes.do(move_the_files)
else:
sys.exit(f'test_run must be set to True or False.')
while True:
schedule.run_pending()
time.sleep(1)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment