AWS S3 Bucket
AWS S3 bucket의 이미지를 Roboflow에 업로드
마지막 업데이트
도움이 되었나요?
도움이 되었나요?
def generate_presigned_url(bucket_name: str, object_name: str, region: str = 'us-east-2') -> str:
"""S3 객체의 사전 서명된 URL을 생성합니다."""
s3 = boto3.client('s3', region_name=region, config=Config(signature_version='s3v4'))
url = s3.generate_presigned_url('get_object',
Params={'Bucket': bucket_name, 'Key': object_name},
ExpiresIn=3600)
return urlimport boto3
import requests
import urllib.parse
from botocore.config import Config
# ************* 다음 변수들을 설정하세요 *************
S3_BUCKET_NAME = "YOUR_S3_BUCKET_NAME"
ROBOFLOW_API_KEY = "YOUR_ROBOFLOW_API_KEY"
ROBOFLOW_PROJECT_NAME = "YOUR_ROBOFLOW_PROJECT_NAME"
# ***********************************************
def generate_presigned_url(bucket_name: str, object_name: str, region: str = 'us-east-2') -> str:
"""S3 객체의 사전 서명된 URL을 생성합니다."""
s3 = boto3.client('s3', region_name=region, config=Config(signature_version='s3v4'))
url = s3.generate_presigned_url('get_object',
Params={'Bucket': bucket_name, 'Key': object_name},
ExpiresIn=3600)
return url
def get_s3_objects(bucket_name: str) -> list:
"""주어진 S3 버킷의 객체 키 목록을 가져옵니다."""
s3 = boto3.client('s3')
objects = []
response = s3.list_objects_v2(Bucket=bucket_name)
for obj in response['Contents']:
objects.append(obj['Key'])
return objects
def upload_to_roboflow(api_key: str, project_name: str, presigned_url: str, img_name='', split="train"):
"""이미지를 Roboflow에 업로드합니다."""
API_URL = "https://api.roboflow.com"
if img_name == '':
img_name = presigned_url.split("/")[-1]
upload_url = "".join([
API_URL + "/dataset/" + project_name + "/upload",
"?api_key=" + api_key,
"&name=" + img_name,
"&split=" + split,
"&image=" + urllib.parse.quote_plus(presigned_url),
])
response = requests.post(upload_url)
# 응답 코드 확인
if response.status_code == 200:
print(f"{img_name}을(를) {project_name}에 성공적으로 업로드했습니다")
return True
else:
print(f"{img_name} 업로드에 실패했습니다. 오류: {response.content.decode('utf-8')}")
return False
if __name__ == "__main__":
# 사용 가능한 이미지 목록 가져오기
available_images = get_s3_objects(S3_BUCKET_NAME)
# 선택 사항: 여기서 이미지를 필터링
# 예: available_images = [img for img in available_images if "some_condition"]
# 이미지를 Roboflow에 업로드
for image in available_images:
presigned_url = generate_presigned_url(S3_BUCKET_NAME, image)
upload_to_roboflow(ROBOFLOW_API_KEY, ROBOFLOW_PROJECT_NAME, presigned_url)
aws s3 sync s3://mybucket/folder_path .