Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # pip3 install requests boto3 watchdog
- import time
- import logging
- import mimetypes
- from watchdog.observers import Observer
- from watchdog.events import LoggingEventHandler, FileCreatedEvent
- import boto3
- import os
- from botocore.exceptions import NoCredentialsError
- AWS_ACCESS_KEY_ID = ''
- AWS_SECRET_ACCESS_KEY = ''
- S3_BUCKET_NAME = 'sstv'
- SSTV_FOLDER = '/mmsstv/images/'
- def upload_to_aws(local_file, bucket, s3_file):
- s3 = boto3.client('s3', aws_access_key_id=AWS_ACCESS_KEY_ID,
- aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
- try:
- mime = mimetypes.guess_type(local_file)[0] or 'binary/octet-stream'
- s3.upload_file(local_file, bucket, s3_file, ExtraArgs={
- "ContentType": mime
- })
- logging.info(f"File Upload Successful ({s3_file})")
- return True
- except FileNotFoundError:
- logging.warning("File Upload Failed: File not found")
- return False
- except NoCredentialsError:
- logging.warning("File Upload Failed: No credentials")
- return False
- def upload_to_website(awsname):
- # Handle uploading file to your website
- pass
- class SSTVEvent(LoggingEventHandler):
- def dispatch(self, event):
- if isinstance(event, FileCreatedEvent):
- if not event.is_directory:
- logging.info(f"New SSTV image detected: {event.src_path}")
- awsname = f"sstv/{time.time()}.jpg"
- if upload_to_aws(event.src_path, S3_BUCKET_NAME, awsname):
- if upload_to_website(awsname):
- logging.info("Removing SSTV file from local disk")
- os.remove(event.src_path)
- logging.info("Done!")
- if __name__ == "__main__":
- logging.basicConfig(level=logging.INFO,
- format='%(asctime)s - %(message)s',
- datefmt='%Y-%m-%d %H:%M:%S')
- event_handler = SSTVEvent()
- observer = Observer()
- observer.schedule(event_handler, SSTV_FOLDER, recursive=False)
- observer.start()
- try:
- while True:
- time.sleep(1)
- except KeyboardInterrupt:
- observer.stop()
- observer.join()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement