S3 와 Lambda Trigger 를 이용한 파이프라인과 tmp dir

import json
import os
import zipfile
import shutil
import time
import boto3

S3_CLIENT = boto3.client("s3")
LAMBDA_CLIENT = boto3.client("lambda")

ZIP_EXT = ".zip"
TMP_ZIP_PATH = "/tmp/result.zip"
EXTRACT_FOLDER_PATH = "/tmp/extracted"
RESULT_FOLDER = "/result/"
ALERT_API_FUNCTION = "alert-api"  # Lambda_2 호출
ALERT_S3_FUNCTION = "alert-s3"  # Lambda_2 호출

# tmp 초기화 코드
def clear_tmp_directory():
    tmp_dir = '/tmp'
    for filename in os.listdir(tmp_dir):
        file_path = os.path.join(tmp_dir, filename)
        try:
            if os.path.isfile(file_path) or os.path.islink(file_path):
                os.unlink(file_path)
            elif os.path.isdir(file_path):
                shutil.rmtree(file_path)
        except Exception as e:
            print("실패하는 경우 처리")

def lambda_handler(event, context):
    invoke_slack(event=event,context=context
    start_time = time.time()

    bucket, key = get_bucket_and_key(event)

    print(f"파일 업로드: {key} bucket: {bucket}")

    clear_tmp_directory() # 폴더 초기화
    if ZIP_EXT in key:
        process_zip_file(bucket, key)

        end_time = time.time()
        elapsed_time = end_time - start_time
        print(f"실행에 걸린 시간: {elapsed_time} 초")

        invoke_slack(contents=contents)
        return {"statusCode": 200, "body": "성공 응답"}

    invoke_slack(contents="실행과정 로그내용")

    raise Exception(exception_info)

def get_bucket_and_key(event):
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = event["Records"][0]["s3"]["object"]["key"]
    return bucket, key

def process_zip_file(bucket, key):
    download_path = download_zip_file(bucket, key)
    unzip_files(download_path)
    upload_files(bucket, key)
    invoke_lambda_function(key)


def download_zip_file(bucket, key):
    S3_CLIENT.download_file(bucket, key, TMP_ZIP_PATH)
    print("다운로드 완료 및 실행준비 단계, 이 단계에서 tmp 를 이용")
    return TMP_ZIP_PATH

def unzip_files(zip_file_path):
    os.makedirs(EXTRACT_FOLDER_PATH, exist_ok=True)
    with zipfile.ZipFile(zip_file_path, "r") as zip_ref:
        zip_ref.extractall(EXTRACT_FOLDER_PATH)

def upload_files(bucket, key):
    upload_path = "/".join(key.split("/")[:-1])
    for root, dirs, files in os.walk(EXTRACT_FOLDER_PATH):
        for file in files:
            upload_file_to_s3(bucket, file, root, upload_path)

def upload_file_to_s3(bucket, file, root, upload_path):
    local_path = os.path.join(root, file)
    s3_key = os.path.relpath(local_path, EXTRACT_FOLDER_PATH)
    s3_upload_path = upload_path + RESULT_FOLDER + s3_key # 업로드되는 경로 지정 upload_path/s3_key


    if s3_key.split(".")[-1] == "txt":
        S3_CLIENT.upload_file(local_path,bucket,s3_upload_path,ExtraArgs={"ContentType":"text/plain; charset=utf-8"})
    else:
        S3_CLIENT.upload_file(local_path, bucket, s3_upload_path)

def invoke_lambda_function(key):
    print("별도 로직")

def invoke_slack(event=None, context=None,contents=None):
    print("별도 로직")