mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
import boto3
|
|
import csv
|
|
from io import StringIO
|
|
import string
|
|
import secrets
|
|
import logging
|
|
|
|
|
|
def setup_logger(log_file=None, level=logging.INFO, overwrite_handler=False):
|
|
# Create a logger and set the logging level
|
|
logger = logging.getLogger()
|
|
logger.setLevel(level)
|
|
|
|
# if logger already has handlers, just return it
|
|
if logger.hasHandlers() and not overwrite_handler:
|
|
return logger
|
|
|
|
# Define the log message format
|
|
log_format = "%(asctime)s [%(levelname)s] %(message)s"
|
|
date_format = "%Y-%m-%d %H:%M:%S"
|
|
formatter = logging.Formatter(log_format, datefmt=date_format)
|
|
|
|
# Create a file handler and set the file path and format
|
|
if log_file:
|
|
file_handler = logging.FileHandler(log_file)
|
|
file_handler.setLevel(level)
|
|
file_handler.setFormatter(formatter)
|
|
logger.addHandler(file_handler)
|
|
|
|
# Create a console handler and set the format
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(level)
|
|
|
|
# Set the formatter for the handlers
|
|
console_handler.setFormatter(formatter)
|
|
|
|
# Add the handlers to the logger
|
|
logger.addHandler(console_handler)
|
|
|
|
return logger
|
|
|
|
|
|
def read_csv_from_s3(bucket_name, filepath):
|
|
s3 = boto3.client('s3')
|
|
|
|
# Get the object from s3
|
|
s3_object = s3.get_object(Bucket=bucket_name, Key=filepath)
|
|
|
|
# Read the CSV body from the s3 object
|
|
body = s3_object['Body'].read()
|
|
|
|
# Use StringIO to create a file-like object from the string
|
|
csv_data = StringIO(body.decode('utf-8'))
|
|
|
|
# Use csv library to read it into a list of dictionaries
|
|
reader = csv.DictReader(csv_data)
|
|
data = list(reader)
|
|
|
|
return data
|
|
|
|
|
|
def generate_api_key():
|
|
# Define the characters that will be used to generate the api key
|
|
characters = string.ascii_letters + string.digits
|
|
# Generate a 40 character long api key
|
|
api_key = ''.join(secrets.choice(characters) for _ in range(40))
|
|
return api_key
|