log weekly august boto3 s3

boto3로 S3 관리하기

2024년 8월 22일 목요일

Today I Learned

날짜

2024년 8월 22일 목요일

내용

불친절한 이웃 네이버

그래도 내가 쓴 코드라 그런지, 로직은 쉽게 고쳤다. 곤란한 경우가 있었는데, 보고서 생성 요청을 보내도 반환값의 downloadUrl 은 빈값이 들어온다. status 가 REGIST 라고 들어오는데 “니가 보낸 만들어달란 요청 받았음” 이라는 뜻이란다. 그냥 바로 만든거 달라고… 뭐 별수있나 만들고 요청을 보내야지. 그래서

만들고 ⇒ 응답값에 들어있는 reportId 로 다운로드 링크 조회하고 ⇒ 그 링크에서 다운로드

로 변경했다.

근데 웬걸? 조회 요청의 응답에 status 가 RUNNING 으로 온다. 아직 만드는 중이라는 뜻인데 생각보다 생성이 느리다. 어쩔수 없이 40일치 생성 ⇒ 40일치 다운로드 로 로직을 변경했다.

네이버 검색광고 API에 대해 불만인 점은 Docs가 너무 불친절하다는 거다. status 응답값이 각각 어떤 의미인지를 안말해준다. 본인들 github issue에 있는 4년전 질문에 대한 답변에서 찾을 수 있었다. 응답값이 어떤 의미인지 Docs에 적으라고..

재요청 시간 단축하기

많게는 400일치, 적게는 90일치의 데이터를 매일 가져와야 한다. 이 중 의미없이 반복되는 데이터의 양을 줄이려고 한다. 알아보니 30일이 지난 데이터는 더이상 갱신되지 않는다고 하니, 그 이후 데이터는 처리해서 저장하려고 한다.

데이터베이스에 저장할까 생각했는데 나중에 너무 커질 것 같기도 하고.. {고객의 ID}/{날짜}/{보고서 형식}.csv 디렉토리를 만들어 S3에 저장해두고 써야겄다. 로직을 잘게 분리해놨더니 확실히 구현은 훨씬 쉬워짐! 단일 책임의 원칙은 위대하다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

def download_data_from_s3(naver_customer_id: str):
    """
    기존에 저장되어있는 S3 데이터를 가져오는 함수
    """
    try:
        file_names = []
        s3_client = boto3.client("s3")
        s3_bucket = "imreport.io"

        response = s3_client.list_objects_v2(
            Bucket=s3_bucket,
            Prefix=f"data_reports/{naver_customer_id}",
        )

        # 객체가 있는지 확인
        if 'Contents' in response:
            for obj in response['Contents']:
                key = obj['Key']
                size = obj['Size']
                
                # 크기가 0인 객체는 무시
                if size == 0:
                    continue

                local_file_path = os.path.join("./", os.path.basename(key))

                # 객체 다운로드
                s3_client.download_file(s3_bucket, key, local_file_path)
                
                file_name = key.split("/")[-1]
                file_names.append(file_name)
        return file_names

    except Exception as e:
        logging.exception(f"S3에서 다운로드 중 에러 발생 {e}")

def get_data_from_downloaded_s3_file(file_names: list):
    try:
        all_data = []
        for file_name in file_names:
            with open(file_name, "r") as f:
                rows = f.readlines()
                for row in rows:
                    # \n 제거
                    row = row.strip()
                    # ,로 구분된 데이터를 리스트로 변환
                    row = list(row.split('|'))
                    # 날짜를 제외하고 숫자로 변환 가능한 데이터는 숫자로 변환
                    for index, value in enumerate(row):
                        if index == 0:
                            continue
                        if value.isdigit():
                            row[index] = int(value)
                    all_data.append(row)
        return all_data
    except Exception as e:
        logging.exception(f"S3에서 다운로드 받은 파일에서 데이터 꺼내는 중 에러 발생 {e}")
        return []

회고

데이터가 적으면 테스트가 안되고, 많으면 오래걸리고 거참 힘들다.

CATALOG
    weekly