The following script uploads files from the first of the month via S3
import os
import datetime
import re
import subprocess
import argparse
def get_newest_in_first_day(path, s3_target_path):
"""
This function scans a given directory for files and directories that were created on the first day of the current month.
It then uploads the newest file and directory to a specified S3 target path using the s3cmd tool.
Args:
path (str): The path of the directory to scan.
s3_target_path (str): The S3 path to upload the file and directory to.
Returns:
Tuple[str, str]: The names of the newest file and directory, if they exist.
"""
first_day_files = []
first_day_dirs = []
today = datetime.datetime.now()
# Calculate the first day of the current month
first_day_of_month = today.replace(day=1)
pattern = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z(\.pbm\.json)?$")
for name in os.listdir(path):
# Skip names that do not match the pattern
if not pattern.match(name):
continue
full_path = os.path.join(path, name)
timestamp_str = name.split('T')[0] # Extract the date part from the name
timestamp = datetime.datetime.strptime(timestamp_str, "%Y-%m-%d")
# Check if it's the first day of the month
if timestamp.date() == first_day_of_month.date():
if os.path.isfile(full_path):
first_day_files.append((timestamp, name))
elif os.path.isdir(full_path):
first_day_dirs.append((timestamp, name))
# Sort by timestamp (newest first) and get the first name
first_day_files.sort(reverse=True)
first_day_dirs.sort(reverse=True)
newest_file = first_day_files[0][1] if first_day_files else None
newest_dir = first_day_dirs[0][1] if first_day_dirs else None
# If both the file and directory exist, upload them
if newest_file and newest_dir:
s3_config_path = "./s3config.txt"
subprocess.call(["s3cmd", "put", "--config", s3_config_path, os.path.join(path, newest_file), s3_target_path])
subprocess.call(["s3cmd", "put", "--recursive", "--config", s3_config_path, os.path.join(path, newest_dir), s3_target_path])
print(f"Uploaded file: {newest_file}")
print(f"Uploaded directory: {newest_dir}")
else:
print("Nothing was uploaded.")
return newest_file, newest_dir
def main():
parser = argparse.ArgumentParser(description='Process some paths.')
parser.add_argument('path', type=str, help='The path of the directory to scan')
parser.add_argument('s3_target_path', type=str, help='The S3 path to upload the file and directory to')
args = parser.parse_args()
get_newest_in_first_day(args.path, args.s3_target_path)
if __name__ == "__main__":
main()
With Notification Email
The following version sends an email with the results
import os
import datetime
import re
import subprocess
import argparse
import smtplib
from email.mime.text import MIMEText
def send_email(subject, content):
msg = MIMEText(content)
msg['Subject'] = subject
msg['From'] = 'youremail@yourdomain.com'
msg['To'] = 'receiver@yourdomain.com'
# setup the SMTP server
server = smtplib.SMTP('your-smtp-server.com', 587) # use your smtp server
server.starttls()
server.send_message(msg)
server.quit()
def get_newest_in_first_day(path, s3_target_path):
# [The function body remains the same]
if newest_file and newest_dir:
# [The same upload code]
print(f"Uploaded file: {newest_file}")
print(f"Uploaded directory: {newest_dir}")
send_email('Upload Successful', f'Uploaded file: {newest_file}\nUploaded directory: {newest_dir}')
else:
print("Nothing was uploaded.")
send_email('Nothing was Uploaded', 'No new files or directories found on the first day of this month.')
return newest_file, newest_dir
def main():
# [Same as before]
if __name__ == "__main__":
main()
There is no error handling for email, so add this if needed.