[AWS] s3에 이미지 업로드

1. 이미지 추출
2. 이미지 업로드
3. 이미지 클라이언트에게 전달

 

이전 포스팅에서 만들었던 s3를 활용해 자취방 이미지들을 s3에 업로드해서 사용해보도록 하겠습니다.

 

1. 이미지 추출

먼저 pdf에서 이미지를 추출해보도록하겠습니다. 전에 upstage의 document parser를 사용해봤었는데 이미지만 추출하는게 아니라 텍스트,이미지 등을 종류별로 추출해서 마크다운, html 형식으로 변환해주다보니 이미지 추출쪽에서는 성능이 부족한 상황을 경험했습니다. 따라서 이번엔 이미지 추출만을 목적으로 하는 라이브러리를 사용해서 성능에 중점을 맞추고 진행해보았습니다.

 

1 ) PyMuPDF

PyMuPDF 는 PDF에 포함된 원본 이미지 파일을 별도의 변환 없이 추출합니다. 변환 과정이 없기 때문에 속도와 품질 모두 우수합니다. 

 

import fitz  # PyMuPDF
import os

def extract_images_from_pdf(pdf_path, output_dir):
    # PDF 열기
    pdf_document = fitz.open(pdf_path)
    os.makedirs(output_dir, exist_ok=True)
    image_paths = []

    for page_num in range(len(pdf_document)):
        page = pdf_document[page_num]
        images = page.get_images(full=True)  # 이미지 정보 가져오기

        for img_index, img in enumerate(images):
            xref = img[0]  # 이미지 xref 값
            base_image = pdf_document.extract_image(xref)  # 이미지 추출
            image_bytes = base_image["image"]
            image_ext = base_image["ext"]  # 이미지 확장자 (png, jpeg 등)
            image_path = os.path.join(output_dir, f"{page_num + 1}_{img_index + 1}.{image_ext}")

            # 이미지 파일 저장
            with open(image_path, "wb") as image_file:
                image_file.write(image_bytes)
            image_paths.append(image_path)
    
    pdf_document.close()
    return image_paths

# PDF 경로와 이미지 저장 디렉토리
pdf_path = './data/pdf_data/back_gate.pdf'
output_dir = './data/images/back_gate'

# 이미지 추출
images = extract_images_from_pdf(pdf_path, output_dir)
print(f"Extracted images: {images}")

 

테스트용으로 후문에 위치한 자취방들을 대상으로 이미지 추출을 진행해주었습니다. 저장할 이미지명은 "(page_num)_(img_index)" 형식으로 지정했습니다. 후에 page_num으로 이미지를 구분해서 s3에 page_num별로 폴더를 만드려고 위와 같은 형식으로 이미지를 저장했습니다.

 

 

상당히 괜찮은 화질로 이미지가 추출된걸 확인하고 이제 s3에 이미지를 저장하는 작업을 진행하였습니다.

 

2. 이미지 업로드

import boto3
import os
import re

# S3 클라이언트 생성
s3 = boto3.client('s3')
BUCKET_NAME = '{버킷명}'
BASE_FOLDER = 'rooms/'  # S3 내 기본 폴더

def upload_images_to_s3(folder_path):
    # 폴더 내 파일 가져오기
    files = os.listdir(folder_path)

    # 정규식으로 파일 이름에서 room_id와 이미지 번호 추출
    pattern = r"(\d+)_(\d+)\.(\w+)"  # ex: 1_1.jpg, 2_2.png

    for file_name in files:
        match = re.match(pattern, file_name)
        if match:
            room_id, img_num, extension = match.groups()
            room_folder = f"{BASE_FOLDER}{room_id}/"  # S3의 room_id 폴더 경로
            s3_key = f"{room_folder}{file_name}"  # S3에 업로드할 전체 경로

            # 로컬 파일 경로
            local_path = os.path.join(folder_path, file_name)

            # S3에 파일 업로드
            s3.upload_file(local_path, BUCKET_NAME, s3_key)
            print(f"Uploaded {file_name} to s3://{BUCKET_NAME}/{s3_key}")

# 로컬 이미지 폴더 경로
local_folder_path = './data/images/back_gate'

# 이미지 업로드 실행
upload_images_to_s3(local_folder_path)

 

앞에서 말했던 대로 이미지명에서 room_id로 사용할 이름을 정규표현식으로 추출해서 s3에 room_id 폴더를 만들고 내부에 이미지들을 저장해주었습니다.

 

 

 

정상적으로 이미지가 업로드 된 모습입니다. 이제 업로드된 이미지가 잘 가져와지는지 테스트 해보았습니다.

 

import boto3

# S3 클라이언트 생성
s3 = boto3.client('s3')
BUCKET_NAME = '{버킷명}'
BASE_FOLDER = 'rooms/'  # S3 내 기본 폴더 경로

def list_images_in_room(bucket_name, room_id):
    folder_prefix = f"{BASE_FOLDER}{room_id}/"  # room_id에 해당하는 폴더 경로
    images = []

    # S3에서 특정 폴더의 객체 리스트 가져오기
    response = s3.list_objects_v2(Bucket=bucket_name, Prefix=folder_prefix)

    # 객체 리스트에서 파일 경로만 추출
    if 'Contents' in response:
        for obj in response['Contents']:
            images.append(obj['Key'])  # S3 객체 키 (파일 경로)

    return images

# 예시: room_id가 1인 경우
room_id = '1'
images = list_images_in_room(BUCKET_NAME, room_id)
print(f"Images in room {room_id}: {images}")

 

1번 room_id에 해당하는 이미지들을 가져오는 샘플 코드를 작성해서 실행해보았고 아래 에러가 발생했습니다.

 

 

이 에러는 AWS IAM 정책이 S3 버킷에 대해 s3:ListBucket 액션을 허용하지 않아서 발생하는 문제입니다. 현재 사용 중인 IAM 사용자 또는 역할에게 s3:ListBucket 권한이 필요한 것 이었습니다. 이를 해결해주기 위해 s3:ListBucket 권한을 부여해주도록 하겠습니다.

 

 

"AmazonS3FullAccess" 를 추가해서 s3 에 대한 모든 요청을 허용하도록 하였습니다.

 

 

코드를 다시 실행해보니 정상적으로 이미지가 가져와지는 모습을 볼 수 있었습니다.

 

3. 이미지 클라이언트에게 전달

지정한 room_id에 맞는 이미지들을 추출할 수 있지만 위의 형태로는 클라이언트에게 이미지를 보여줄 수 없습니다. 클라이언트에서 room_id로 요청이 들어오면 해당 이미지들에 대한 객체 URL을 json format으로 바꿔서 클라이언트에게 전달해 주도록 하겠습니다.

객체 URL은 "https://<bucket-name>.s3.<region>.amazonaws.com/<object-key>" 의 형태를 지니고 있습니다. 

 

import boto3
import json

# S3 클라이언트 생성
s3 = boto3.client('s3')
BUCKET_NAME = '{버킷명}'
BASE_FOLDER = 'rooms/'  # S3 내 기본 폴더 경로
REGION = 'ap-northeast-2'  # S3 버킷이 위치한 AWS 리전

def list_images_in_room(bucket_name, room_id):
    folder_prefix = f"{BASE_FOLDER}{room_id}/"  # room_id에 해당하는 폴더 경로
    images = []

    # S3에서 특정 폴더의 객체 리스트 가져오기
    response = s3.list_objects_v2(Bucket=bucket_name, Prefix=folder_prefix)

    # 객체 리스트에서 파일 경로만 추출
    if 'Contents' in response:
        for obj in response['Contents']:
            if not obj['Key'].endswith('/'):  # 폴더 자체는 제외
                images.append(obj['Key'])  # S3 객체 키 (파일 경로)

    return images

def generate_image_urls(bucket_name, region, keys):
    urls = [
        f"https://{bucket_name}.s3.{region}.amazonaws.com/{key}"
        for key in keys
    ]
    return urls

# 예시: room_id가 1인 경우
room_id = '1'
keys = list_images_in_room(BUCKET_NAME, room_id)

# S3 객체 키를 URL로 변환
image_urls = generate_image_urls(BUCKET_NAME, REGION, keys) if keys else []

# JSON 응답 생성
response_data = {
    "room_id": room_id,
    "images": image_urls
}
response_json = json.dumps(response_data, indent=4)  # JSON 문자열로 변환 (보기 좋게 들여쓰기 포함)

# 클라이언트에게 전송 (예: 출력)
print(response_json)

 

따라서 위와 같이 코드를 작성해서 image가 있는 room_id 요청에 대해서는 적절한 이미지들을 json 으로 전달하고, 이미지가 없는 경우 비어있는 images를 전달하게끔 코드를 작성하였습니다.