이번 학기 딥러닝 과목 기말 프로젝트로 AWS를 활용한 개인 맞춤형 영화 추천 시스템을 개발하게 되었다. 전체적인 프로젝트 진행은 데이터 수집 및 전처리 - 모델 개발 - 모델 서빙 - 웹 인터페이스 통합 순으로 진행할 예정이다. 가장 먼저 데이터 수집 및 전처리 단계부터 시작하려고 한다. 영화 추천 AI 모델 개발을 위해서는 양질의 영화 데이터가 필수적이기에, 이번 글에서는 TMDB API를 활용하여 대량의 영화 데이터를 자동으로 수집하는 시스템을 어떻게 설계하고 구현했는지에 대해 자세히 공유하고자 한다.
본 시스템은 크게 두 가지 주요 기능으로 나뉜다. 첫 번째는 인기 영화 목록을 수집하는 것이고, 두 번째는 각 영화의 상세 정보를 가져오는 것이다. 이를 위해 requests 라이브러리를 사용하여 TMDB API에 호출을 보내고, 수집된 데이터를 pandas를 사용해서 구조화된 형태로 저장하는 프로세스를 거친다.
TMDBCollector 클래스 구현
TMDBCollector 클래스는 데이터 수집을 위한 준비 작업을 초기화하는 __init__ 메소드를 포함한다. 이 메소드에서는 TMDB API를 사용하기 위한 API 키와 기본 URL을 설정하고, API 호출을 담당할 requests.Session 객체를 생성한다. requests.Session은 여러 번의 HTTP 요청을 보내는 경우 각 요청마다 동일한 설정을 재사용할 수 있도록 도와준다.
class TMDBCollector:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.themoviedb.org/3"
self.session = requests.Session()
_make_request 메소드는 TMDB API와의 통신을 담당한다. 이 메소드는 URL과 파라미터를 입력받아 HTTP 요청을 보내고, 응답을 받아 JSON 형태로 반환하는 역할을 한다. 또한 네트워크 요청 중에 오류가 발생할 수 있기 때문에 최대 3번까지 재시도하는 로직을 추가하였다. 만약 요청이 세 번 모두 실패하면 오류 메시지를 출력하고 None을 반환한다. 이 메소드는 requests.Session을 활용하여 API 호출 속도 제한을 고려하여 각 요청 사이에 0.25초의 대기시간을 삽입한다.
def _make_request(self, url, params):
"""API 요청 with 재시도 로직"""
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.get(url, params=params)
response.raise_for_status()
time.sleep(0.25)
return response.json()
except Exception as e:
if attempt == max_retries - 1:
print(f"Error after {max_retries} attempts: {e}")
return None
time.sleep(1)
discover_movies 메소드는 특정 연도의 인기 영화를 페이지네이션을 통해 수집하는 기능을 수행한다. TMDB API의 /discover/movie 엔드포인트를 호출하여 인기 영화를 가져오며, 기본적으로 popularity.desc로 영화 목록을 정렬하고, primary_release_year 파라미터를 사용하여 특정 연도의 영화만 필터링한다. 한 번에 한 페이지씩 데이터를 받아오며, 페이지 번호를 파라미터로 넘겨서 여러 페이지에 걸쳐 데이터를 수집할 수 있도록 하였다. 기본적으로 한 페이지의 데이터를 반환하지만, 페이지 수가 많을 경우에는 페이지네이션을 통해 여러 페이지에 걸쳐 데이터를 계속해서 수집하게 된다.
def discover_movies(self, year, page=1):
"""특정 연도의 인기 영화 목록을 수집"""
url = f"{self.base_url}/discover/movie"
params = {
'api_key': self.api_key,
'language': 'ko-KR',
'sort_by': 'popularity.desc',
'page': page,
'primary_release_year': year,
'include_adult': 'false'
}
return self._make_request(url, params)
get_movie_details 메소드는 특정 영화의 상세 정보를 가져오는 기능을 담당한다. 영화의 고유 ID를 입력받아 TMDB API의 /movie/{movie_id} 엔드포인트를 호출하여 해당 영화의 상세 데이터를 요청한다. 이때 append_to_response 파라미터를 사용하여 추가적인 정보를 요청할 수 있다. 예를 들어 출연진 정보, 감독 정보, 영화 키워드, 유사 영화 목록 등을 함께 포함시킬 수 있다.
def get_movie_details(self, movie_id):
"""영화 상세 정보를 수집"""
url = f"{self.base_url}/movie/{movie_id}"
params = {
'api_key': self.api_key,
'language': 'ko-KR',
'append_to_response': 'credits,keywords,similar'
}
return self._make_request(url, params)
전체 로직 코드는 아래와 같다.
class TMDBCollector:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.themoviedb.org/3"
self.session = requests.Session()
def _make_request(self, url, params):
"""API 요청 with 재시도 로직"""
max_retries = 3
for attempt in range(max_retries):
try:
response = self.session.get(url, params=params)
response.raise_for_status()
time.sleep(0.25)
return response.json()
except Exception as e:
if attempt == max_retries - 1:
print(f"Error after {max_retries} attempts: {e}")
return None
time.sleep(1)
def discover_movies(self, year, page=1):
"""특정 연도의 영화 발견"""
url = f"{self.base_url}/discover/movie"
params = {
'api_key': self.api_key,
'language': 'ko-KR',
'sort_by': 'popularity.desc',
'page': page,
'primary_release_year': year,
'include_adult': 'false'
}
return self._make_request(url, params)
def get_movie_details(self, movie_id):
"""영화 상세 정보 수집"""
url = f"{self.base_url}/movie/{movie_id}"
params = {
'api_key': self.api_key,
'language': 'ko-KR',
'append_to_response': 'credits,keywords,similar'
}
return self._make_request(url, params)데이터 수집 프로세스
본격적으로 데이터를 수집하는 로직에서는 tqdm 라이브러리를 사용하여 진행 상황을 실시간으로 모니터링하고 수집 상태를 확인할 수 있도록 했다. collect_all_movies 함수는 주어진 연도 범위에 대해 인기 영화 목록을 수집한다. 이때 TMDBCollector 클래스의 discover_movies 메소드를 사용하여 해당 연도의 인기 영화를 페이지네이션 방식으로 가져온다. 각 영화의 ID와 기본 정보는 all_movie_ids와 movie_basic_info에 저장된다.
def collect_all_movies(collector, start_year=1990, end_year=2024):
"""전체 영화 데이터 수집"""
all_movie_ids = set()
movie_basic_info = {}
# 진행 상황 파일 확인
progress_file = f'{SAVE_DIR}/collection_progress.json'
if os.path.exists(progress_file):
with open(progress_file, 'r') as f:
progress = json.load(f)
start_year = progress['current_year']
all_movie_ids = set(progress['collected_ids'])
print(f"이어서 수집 시작: {start_year}년부터")
# 연도별 수집
for year in range(start_year, end_year + 1):
print(f"\n{year}년 영화 수집 중...")
page = 1
while True:
response = collector.discover_movies(year, page)
if not response or not response.get('results'):
break
for movie in response['results']:
all_movie_ids.add(movie['id'])
movie_basic_info[movie['id']] = {
'title': movie['title'],
'release_date': movie['release_date'],
'popularity': movie['popularity']
}
with open(progress_file, 'w') as f:
json.dump({
'current_year': year,
'collected_ids': list(all_movie_ids)
}, f)
if page >= response['total_pages']:
break
page += 1
print(f"{year}년 완료: {len(all_movie_ids)}개의 영화 ID 수집")
if year % 5 == 0:
save_path = f'{SAVE_DIR}/movie_ids_{year}.json'
with open(save_path, 'w') as f:
json.dump(list(all_movie_ids), f)
return all_movie_ids, movie_basic_info
인기 영화 목록은 여러 페이지로 나누어 제공되는데, 마찬가지로 페이지네이션을 통해 각 페이지를 순차적으로 호출하여 모든 영화 데이터를 확보할 수 있도록 하였다. page 변수를 사용하여 페이지 번호를 갱신하고 마지막 페이지까지 데이터를 가져온 후 수집을 완료한다.
수집한 데이터는 5년 단위로 영화 ID 목록을 JSON 파일로 저장하여 중간에 시스템이 종료되더라도 데이터 손실 없이 이어서 수집할 수 있도록 구성하였다. 실제로 데이터를 수집하는 도중에 네트워크 문제로 한번, 노트북 방전 문제로 한번 중단되었는데, 덕분에 정신적 타격 없이 이어서 데이터를 수집할 수 있었다.
수집된 영화 정보는 all_movie_ids와 movie_basic_info에 저장된 후 상세 정보 수집 단계로 넘어가게 된다.
영화 상세 정보 수집
collect_movie_details 함수는 주어진 영화 ID 목록을 기반으로 각 영화의 상세 정보를 수집한다. 이때 TMDBCollector 클래스의 get_movie_details 메소드를 사용하여 각 영화의 상세 정보를 요청한다. 수집되는 상세 정보는 영화의 기본 정보, 장르, 출연진, 감독, 키워드, 평점, 인기 등으로 구성되며, 수집된 정보를 pandas DataFrame 형태로 변환하여 관리한다.
def collect_movie_details(collector, movie_ids):
"""영화 상세 정보 수집"""
detailed_movies = []
progress_file = f'{SAVE_DIR}/details_progress.json'
if os.path.exists(progress_file):
with open(progress_file, 'r') as f:
progress = json.load(f)
collected_ids = set(progress['collected_ids'])
movie_ids = [id for id in movie_ids if id not in collected_ids]
detailed_movies = progress['collected_movies']
print(f"이어서 수집: {len(movie_ids)}개 남음")
각 영화에 대해 collector.get_movie_details(movie_id) 메소드를 호출하여 데이터를 가져온다. 해당 데이터는 각 영화에 대한 다양한 세부 속성들로 구성되어 있다.
- id: 영화의 고유 ID
- title: 영화 제목
- original_title: 원제
- overview: 줄거리
- genres: 영화의 장르
- keywords: 관련된 키워드
- cast: 주요 출연진 (최대 10명)
- crew: 감독과 작가 등 주요 제작진
- release_date: 개봉일
- runtime: 영화의 길이
- vote_average: 평균 평점
- vote_count: 평점 수
- popularity: 인기 지표
- poster_path: 포스터 경로
- similar_movies: 유사 영화 리스트
for movie_id in tqdm(movie_ids):
details = collector.get_movie_details(movie_id)
if details:
movie_data = {
'id': details['id'],
'title': details['title'],
'original_title': details['original_title'],
'overview': details['overview'],
'genres': [g['name'] for g in details['genres']],
'keywords': [k['name'] for k in details.get('keywords', {}).get('keywords', [])],
'cast': [{'id': c['id'], 'name': c['name']} for c in details['credits']['cast'][:10]],
'crew': [{'id': c['id'], 'name': c['name'], 'job': c['job']}
for c in details['credits']['crew'] if c['job'] in ['Director', 'Writer']],
'release_date': details['release_date'],
'runtime': details['runtime'],
'vote_average': details['vote_average'],
'vote_count': details['vote_count'],
'popularity': details['popularity'],
'poster_path': details['poster_path'],
'similar_movies': [m['id'] for m in details.get('similar', {}).get('results', [])]
}
detailed_movies.append(movie_data)
100개마다 수집된 영화 정보를 CSV 파일로 저장하고, 진행 상태를 JSON 파일에 기록하여 데이터 수집이 중단되더라도 중간 저장된 데이터를 활용하여 다시 이어서 작업할 수 있도록 했다.
if len(detailed_movies) % 100 == 0:
df = pd.DataFrame(detailed_movies)
df.to_csv(f'{SAVE_DIR}/movies_detailed_temp.csv', index=False)
with open(progress_file, 'w') as f:
json.dump({
'collected_ids': [m['id'] for m in detailed_movies],
'collected_movies': detailed_movies
}, f)
초기 30만개 영화 데이터 수집하는데 약 2시간 정도 소요되었고, 모든 영화의 상세 정보를 수집하는 데 약 6시간 이상 걸릴 것으로 예상된다. (현재진행형) 엄청난 시간을 견디게 해준 것은 역시 tqdm 라이브러리 덕분이다. 제대로 잘 되고 있는지 시각적으로 확인할 수 있다는 것은 생각보다 엄청난 정신적 안정을 가져다준다.
데이터 저장 및 분석
마지막 단계로, detailed_movies 리스트에 담긴 각 영화의 상세 정보는 pandas의 DataFrame으로 변환되어 CSV 파일 형식으로 저장된다. 이 과정에서 index=False 옵션을 설정하여 DataFrame의 인덱스는 저장하지 않도록 했다.
df = pd.DataFrame(detailed_movies)
df.to_csv(f'{SAVE_DIR}/movies_detailed_final.csv', index=False)
print("\n수집 완료 통계:")
print(f"총 영화 수: {len(df)}")
print("\n장르별 영화 수:")
genre_counts = pd.DataFrame([genre for genres in df['genres'] for genre in genres]).value_counts()
print(genre_counts)
print("\n데이터 검증:")
print(df.info())
print("\n결측치 확인:")
print(df.isnull().sum())
수집 완료된 영화 데이터는 구글 드라이브에 저장되는 것을 확인할 수 있다. 1990년부터 2024년까지 연도별 인기 영화를 모두 수집했기 때문에 상당히 오랜 시간이 걸렸지만.. 그만큼 영화 추천 모델을 개발하는 데 도움이 되지 않을까 싶다.
다음 포스팅에서는 이렇게 수집한 전체 영화 데이터를 모델 학습 데이터로 사용할 수 있도록 전처리하는 작업에 대해 다뤄볼 예정이다.
To be continued...
'프로젝트' 카테고리의 다른 글
| 영화 추천 시스템 (4) - BERT와 GAT를 활용한 추천 모델 개발하기 (1) | 2024.12.23 |
|---|---|
| 영화 추천 시스템 - (3) Docker Compose를 이용하여 React + Django 개발환경 구축하기 (2) | 2024.12.16 |
| 영화 추천 시스템 - (2) React + Django CI/CD 파이프라인 및 AWS 아키텍처 설계하기 (2) | 2024.12.16 |
