영화진흥위원회 API (pytest 실습)

Data Engineering/실습

영화진흥위원회 API (pytest 실습)

qqprty 2024. 8. 1. 12:04

1. 새로운 환경 세팅

pyenv global 3.11.9
pdm init
ls -l .venv #제대로 생겼는지 확인
source .venv/bin/activate

2. 영화진흥위원회

https://www.kobis.or.kr/kobisopenapi/homepg/main/main.do

영화진흥위원회 오픈API

OPEN API 서비스 영화진흥위원회 영화관입장권통합전산망에서 제공하는 오픈API 서비스로 더욱 풍요롭고 편안한 영화 서비스를 즐겨보세요.

www.kobis.or.kr

1) 키발급

회원가입을 하고, 키 발급/관리에 가서 키 발급받기 신청하면 쉽게 발급 받을 수 있다.

2) 제공서비스

제공서비스는 일별, 주간/주말 박스오피스가 있는데 그 중 일별 박스오피스 API 서비스를 활용할 것이다.

특정 일자 상영작들의 박스오피스 정보를 영화구분(다양성영화,상업영화), 한국/외국 구분, 상영지역 등의 조건을 통해 조회가능하며 REST/SOAP 방식 중 선택적으로 호출가능하다. REST 방식의 응답형식은 XML과 JSON을 지원한다.

자세한건 홈페이지 참고! 우리는 REST 방식, 응답형식은 JSON을 선택했다.

(1) REST 방식

기본 요청 URL

http://www.kobis.or.kr/kobisopenapi/webservice/rest/boxoffice/searchDailyBoxOfficeList.xml (또는 .json)

요청 parameter : 요청 인터페이스 정보를 참조하여 GET 방식으로 호출

(2) 요청 인터페이스

key	문자열(필수)	발급받은키 값을 입력합니다.
targetDt	문자열(필수)	조회하고자 하는 날짜를 yyyymmdd 형식으로 입력합니다.
itemPerPage	문자열	결과 ROW 의 개수를 지정합니다.(default : “10”, 최대 : “10“)
multiMovieYn	문자열	다양성 영화/상업영화를 구분지어 조회할 수 있습니다. “Y” : 다양성 영화 “N” : 상업영화 (default : 전체)
repNationCd	문자열	한국/외국 영화별로 조회할 수 있습니다. “K: : 한국영화 “F” : 외국영화 (default : 전체)
wideAreaCd	문자열	상영지역별로 조회할 수 있으며, 지역코드는 공통코드 조회 서비스에서 “0105000000” 로서 조회된 지역코드입니다. (default : 전체)

(3) 응답구조

boxofficeType	문자열	박스오피스 종류를 출력합니다.
showRange	문자열	박스오피스 조회 일자를 출력합니다.
rnum	문자열	순번을 출력합니다.
rank	문자열	해당일자의 박스오피스 순위를 출력합니다.
rankInten	문자열	전일대비 순위의 증감분을 출력합니다.
rankOldAndNew	문자열	랭킹에 신규진입여부를 출력합니다. “OLD” : 기존 , “NEW” : 신규
movieCd	문자열	영화의 대표코드를 출력합니다.
movieNm	문자열	영화명(국문)을 출력합니다.
openDt	문자열	영화의 개봉일을 출력합니다.
salesAmt	문자열	해당일의 매출액을 출력합니다.
salesShare	문자열	해당일자 상영작의 매출총액 대비 해당 영화의 매출비율을 출력합니다.
salesInten	문자열	전일 대비 매출액 증감분을 출력합니다.
salesChange	문자열	전일 대비 매출액 증감 비율을 출력합니다.
salesAcc	문자열	누적매출액을 출력합니다.
audiCnt	문자열	해당일의 관객수를 출력합니다.
audiInten	문자열	전일 대비 관객수 증감분을 출력합니다.
audiChange	문자열	전일 대비 관객수 증감 비율을 출력합니다.
audiAcc	문자열	누적관객수를 출력합니다.
scrnCnt	문자열	해당일자에 상영한 스크린수를 출력합니다.
showCnt	문자열	해당일자에 상영된 횟수를 출력합니다.

(4) 응답예시

http://kobis.or.kr/kobisopenapi/webservice/rest/boxoffice/searchDailyBoxOfficeList.json?key=<키값>&targetDt=<YYYYMMDD>
http://kobis.or.kr/kobisopenapi/webservice/rest/boxoffice/searchDailyBoxOfficeList.json?key=82ca741a2844c5c180a208137bb92bd7&targetDt=20120101

3. 코드

1) URL 생성

import requests #HTTP 요청을 보내는 모듈
import os #환경변수로 등록된 MOVIE_API_KEY 호출을 위한 os 모듈
import pandas as pd

def gen_url(dt="20120101"):
    base_url = "http://www.kobis.or.kr/kobisopenapi/webservice/rest/boxoffice/searchDailyBoxOfficeList.json"
    key = get_key()
    url = f"{base_url}?key={key}&targetDt={dt}"

    return url

응답예시 형식으로 url을 생성하기 위해 다음과 같이 gen_url을 정의한다.

def get_key():
    key = os.getenv('MOVIE_API_KEY')
    return key

MOVIE_API_KEY는 .zshrc 파일에 다음과 같이 추가하여 환경변수로 등록해뒀다.

export MOVIE_API_KEY="<키값>"

테스트 코드는 다음과 같다

def test_비밀키숨기기():
    key = get_key()  # get_key 함수 호출
    assert key  # API 키가 존재하는지 확인
    
def test_유알엘테스트():
    url = gen_url()  # gen_url 함수 호출
    assert "http" in url  # URL에 'http'가 포함되어 있는지 확인
    assert "kobis" in url  # URL에 'kobis'가 포함되어 있는지 확인

2) Request

def req(dt="20120101"):
    url = gen_url(dt)  #gen_url() 함수를 호출하여 URL을 생성
    r = requests.get(url)  #HTTP GET 요청을 보냄
    code = r.status_code  #응답 코드 확인
    data = r.json()  #JSON 형식으로 응답 데이터를 파싱
    print(data)  #데이터 출력
    return code, data  #응답 코드와 데이터를 반환

def test_req():
    code, data = req()  # req 함수 호출 (기본 날짜)
    assert code == 200  # 응답 코드가 200인지 확인

    code, data = req('20230101')  # 특정 날짜로 req 함수 호출
    assert code == 200  # 응답 코드가 200인지 확인

1. test_req 실행

2. req 함수 호출 -> 주어진 날짜에 대한 API request -> 상태코드, 데이터 반환

3. assert문으로 반환된 상태코드가 200인지 확인

특정 날짜의 박스 오피스 데이터를 API에 요청하여 응답을 반환

HTTP 요청: 클라이언트가 서버에 정보를 요청하는 메소드로, GET 요청은 서버에서 데이터를 가져오는 역할
JSON 파싱: 서버 응답을 JSON 형식으로 변환하는 과정

+) assert : 테스트 조건을 확인하는 데 사용됨

조건이 참이면 아무 일도 일어나지 않는다.
조건이 거짓이면 AssertionError가 발생하며, 테스트는 실패로 기록된다.

request 모듈 공식문

https://docs.python-requests.org/en/latest/user/quickstart/

3) req2list

def req2list(load_dt='20120101') -> list: #리스트를 반환하는 req2list 함수 정의
    _, data = req() #req() 함수를 호출하여 데이터를 가져옴
    l = data['boxOfficeResult']['dailyBoxOfficeList'] # 받아온 데이터에서 'dailyBoxOfficeList' 부분을 추출
    return l

() -> list는 이 함수가 인자를 받지 않으며, 리스트를 반환한다는 것을 나타낸다
req 함수는 API 요청을 보내고 응답을 받는 함수로써 상태 코드와 응답 데이터, 두 개의 값을 반환한다
_는 상태 코드를 무시하겠다는 의미로 사용된다. 여기서는 상태 코드를 사용하지 않기 때문에 _로 받아온다
data는 JSON 형식의 응답 데이터
data['boxOfficeResult']는 박스오피스 결과를 포함하는 딕셔너리 (key-value의 쌍, {})
data['boxOfficeResult']['dailyBoxOfficeList']는 일별 박스오피스 리스트

API 응답 데이터 예시

{
  "boxOfficeResult": {
    "dailyBoxOfficeList": [
      {
        "rnum": "1",
        "rank": "1",
        "rankInten": "0",
        "rankOldAndNew": "OLD",
        "movieCd": "20112207",
        "movieNm": "Mission: Impossible - Ghost Protocol",
        "openDt": "20111215",
        "salesAmt": "2776067000",
        "salesShare": "33.4",
        "salesInten": "-415699000",
        "salesChange": "-13",
        "salesAcc": "40541108500",
        "audiCnt": "353274",
        "audiInten": "-47982",
        "audiChange": "-11.9",
        "audiAcc": "5328435",
        "scrnCnt": "697",
        "showCnt": "3223"
      },
      // 더 많은 데이터...
    ]
  }
}

여기서 boxOfficeResult 객체는 박스오피스 결과를 담고 있으며, 그 안에 dailyBoxOfficeList라는 배열이 있다.

이 배열에는 일일 박스오피스 리스트가 들어 있다.

따라서, data['boxOfficeResult']['dailyBoxOfficeList']는 data라는 JSON 객체에서 boxOfficeResult 객체를 가져오고, 그 안의 dailyBoxOfficeList 배열을 가져오는 방식이다. 즉, data는 JSON 전체 객체를, data['boxOfficeResult']는 박스오피스 결과 객체를, data['boxOfficeResult']['dailyBoxOfficeList']는 그 안의 일일 박스오피스 리스트를 의미한다.

req2list를 통해서 JSON 데이터를 처리하여 리스트 형태로 반환한다. 리스트의 각 요소는 일별 박스오피스의 데이터를 포함하는 딕셔너리 형태로 구성된다.

def test_req2list():
    l = req2list()  # req2list 함수 호출
    assert len(l) > 0  # 리스트가 비어 있지 않은지 확인
    
    v = l[0]  # 리스트의 첫 번째 요소 가져오기
    assert 'rnum' in v.keys()  # 첫 번째 요소에 'rnum' 키가 있는지 확인
    assert v['rnum'] == '1'  # 첫 번째 요소의 'rnum' 값이 '1'인지 확인

v.keys()는 딕셔너리 v의 모든 키를 반환
v['rnum']은 v 딕셔너리에서 'rnum' 키의 값을 반환

1. test_req2list 실행

2. req2list 함수 호출 -> 데이터를 리스트 형태로 반환

3. list 길이 확인 (리스트가 비어있는지 확인)

4. 리스트 첫번째 요소 가져와서 v 변수에 저장

5. assert문으로 v 딕셔너리의 'rnum' 키 확인

3. assert문으로 'rnum' 키의 값이 1인지 확인

4) list2df

def list2df(load_dt='20120101'): 
    l = req2list(load_dt) ## req2list 함수를 호출하여 데이터를 리스트 형태로 받아오기
    df = pd.DataFrame(l) ## 받아온 리스트를 데이터프레임으로 변환
    return df

def test_list2df():
    df = list2df()
    print(df)
    assert isinstance(df, pd.DataFrame) # 반환된 객체가 데이터프레임인지 확인
    assert 'rnum' in df.columns
    assert 'openDt' in df.columns
    assert 'movieNm' in df.columns
    assert 'audiAcc' in df.columns

isinstance 함수는 Python 내장 함수 중 하나로, 특정 객체가 특정 클래스나 클래스의 하위 클래스의 인스턴스인지 확인하는 데 사용. isinstance 함수는 두 개의 인수를 받으며, 다음과 같은 형태로 사용

isinstance(object, classinfo)

1. test_list2df 실행

2. list2df 함수 호출 -> 리스트로 반환된 데이터를 데이터프레임으로 변환

3. 데이터프레임 출력

4. isinstance 함수를 사용하여 df가 pandas.DataFrame 인스턴스인지 확인

5. assert문으로 칼럼 존재 확인

5) save2df

def save2df(load_dt='20120101'):
    df=list2df(load_dt) # list2df 함수를 호출하여 데이터를 DataFrame으로 변환
    
    # df에 load_dt 칼럼 추가 (조회 일자 YYYYMMDD 형식)
    df['load_dt'] = load_dt
    print(df.head(5))
    
    # 아래 파일 저장시 load_dt 기준으로 파티셔닝
    df.to_parquet('~/tmp/test_parquet', partition_cols=['load_dt'])
    return df

def test_save2df():
    df = save2df()
    assert isinstance(df, pd.DataFrame)
    assert 'load_dt' in df.columns

1. test_savet2df 실행

2. list2df 함수 호출 -> 리스트로 반환된 데이터를 데이터프레임으로 변환

3. 반환된 DataFrame에 'load_dt' 칼럼 추가 (값은 함수 호출 시 전달된 날짜인 load_dt)

4. df.to_parquet 메서드를 사용하여 DataFrame을 파케이 파일로 저장

5. load_dt 컬럼을 기준으로 파티셔닝

6) apply_type2df

def apply_type2df(load_dt='20120101', path="~/tmp/test_parquet"):
    df=pd.read_parquet(f'{path}/load_dt={load_dt}') # 주어진 경로에서 주어진 날짜의 파케이 파일을 불러오기
    
    # 숫자형으로 변환할 컬럼 리스트를 정의
    num_cols = ['rnum', 'rank', 'rankInten', 'salesAmt', 'audiCnt', 'audiAcc', 'scrnCnt', 'showCnt', 'salesShare', 'salesInten', 'salesChange', 'audiInten', 'audiChange']
    
    # 모든 숫자형 컬럼들을 순회하며 숫자형으로 변환합니다.
    for col_name in num_cols:
        df[col_name] = pd.to_numeric(df[col_name])
    
    return df

def test_apply_type2df():
    df = apply_type2df()
    assert isinstance(df, pd.DataFrame)

    num_cols = ['rnum', 'rank', 'rankInten', 'salesAmt', 'audiCnt', 'audiAcc', 'scrnCnt', 'showCnt', 'salesShare', 'salesInten', 'salesChange', 'audiInten', 'audiChange']

    for c in num_cols:
        assert df[c].dtype in ['int64', 'float64']

1. test_apply_type2df 실행

2. apply_type2df 함수 호출

3. 지정된 경로에서 'load_dt' 날짜의 파케이 파일 읽어오기

4. 특정 칼럼 리스트 숫자형 변환

5. isinstance 함수를 사용하여 df가 pandas.DataFrame 인스턴스인지 확인

6. assert문으로 숫자형 타입 확인

4. 전체 코드

# ~/code/mov/src/api/call.py

import requests
import os
import pandas as pd

def echo(yaho):
    return yaho

def apply_type2df(load_dt='20120101', path="~/tmp/test_parquet"):
    df=pd.read_parquet(f'{path}/load_dt={load_dt}')
    df['rnum']=pd.to_numeric(df['rnum'])
    df['rank']=pd.to_numeric(df['rank'])

    num_cols = ['rnum', 'rank', 'rankInten', 'salesAmt', 'audiCnt', 'audiAcc', 'scrnCnt', 'showCnt', 'salesShare', 'salesInten', 'salesChange', 'audiInten', 'audiChange']
    for col_name in num_cols:
        df[col_name] = pd.to_numeric(df[col_name])
    return df


def save2df(load_dt='20120101'):
    df=list2df(load_dt)
    # df에 load_dt 칼럼 추가 (조회 일자 YYYYMMDD 형식)
    # 아래 파일 저장시 load_dt 기준으로 파티셔닝
    df['load_dt'] = load_dt
    print(df.head(5))
    df.to_parquet('~/tmp/test_parquet', partition_cols=['load_dt'])
    return df

def list2df(load_dt='20120101'):
    l = req2list(load_dt)
    df = pd.DataFrame(l)
    return df

def req2list(load_dt='20120101') -> list:
    _, data = req(load_dt)
    l = data['boxOfficeResult']['dailyBoxOfficeList']
    return l

def get_key():
    key = os.getenv('MOVIE_API_KEY')
    return key

def req(load_dt="20120101"):
    #url = gen_url('20240720')
    url = gen_url(load_dt)
    r = requests.get(url)
    code = r.status_code
    data = r.json()
    print(data)
    return code, data


def gen_url(dt="20120101"):
    base_url = "http://www.kobis.or.kr/kobisopenapi/webservice/rest/boxoffice/searchDailyBoxOfficeList.json"
    key = get_key()
    url = f"{base_url}?key={key}&targetDt={dt}"

    return url

# ~/code/mov/tests/test_call.py

from mov.api.call import gen_url, req, get_key, req2list, list2df, save2df, echo, apply_type2df
import pandas as pd

def test_apply_type2df():
    df = apply_type2df()
    assert isinstance(df, pd.DataFrame)
    assert str(df['rnum'].dtype) in ['int64']
    assert str(df['rank'].dtype) in ['int64']
    assert str(df['audiInten'].dtype) in ['int64']

    num_cols = ['rnum', 'rank', 'rankInten', 'salesAmt', 'audiCnt', 'audiAcc', 'scrnCnt', 'showCnt', 'salesShare', 'salesInten', 'salesChange', 'audiInten', 'audiChange']

    for c in num_cols:
        assert df[c].dtype in ['int64', 'float64']

def test_echo():
    r = echo("hello")
    assert r == "hello"

def test_save2df():
    df = save2df()
    assert isinstance(df, pd.DataFrame)
    assert 'load_dt' in df.columns

def test_list2df():
    df = list2df()
    print(df)
    assert isinstance(df, pd.DataFrame)
    assert 'rnum' in df.columns
    assert 'openDt' in df.columns
    assert 'movieNm' in df.columns
    assert 'audiAcc' in df.columns

def test_req2list():
    l = req2list()
    assert len(l) > 0
    v = l[0]
    assert 'rnum' in v.keys()
    assert v['rnum'] == '1'

def test_비밀키숨기기():
    key = get_key()
    assert key


def test_유알엘테스트():
    url = gen_url()
    assert "http" in url
    assert "kobis" in url

def test_req():
    code, data = req()
    assert code == 200

    code, data = req('20230101')
    assert code == 200

'Data Engineering > 실습' 카테고리의 다른 글

Spark 프로그램 Airflow에서 돌리기 (0)	2024.08.14
fastapi + movie api (2)	2024.08.13
argparse 를 이용한 히스토리 cli 고도화 (0)	2024.07.29
DB 파티셔닝 (Partitioning) (0)	2024.07.29
Parquet 파일 형식 (0)	2024.07.26

현재글영화진흥위원회 API (pytest 실습)

oddsummer

oddsummer 입니다.

Today :
Yesterday :

oddsummer