tmdb_to_sql.py

import requests
import random
import re

# TMDB API 키 및 설정
API_KEY = "TMDB API 키"
BASE_URL = "https://api.themoviedb.org/3"
GENRE_MAP = {
    28: "액션", 16: "애니메이션", 10749: "로맨스", 99: "다큐멘터리",
    18: "드라마", 35: "코미디", 27: "공포", 10402: "음악",
    10770: "Reality TV", 878: "SF", 10751: "가족", 80: "범죄",
    10752: "전쟁", 53: "스릴러", 14: "판타지", 36: "역사",
    37: "서부", 12: "어드벤처", 9648: "미스터리"
}
LANGUAGE = "ko-KR"
OUTPUT_FILE = "optimized_tmdb_data.sql"

# 한글 확인 함수
def is_korean(text):
    return bool(re.search("[가-힣]", text))

# API 요청 공통 함수
def fetch_from_api(endpoint, params):
    url = f"{BASE_URL}/{endpoint}"
    response = requests.get(url, params=params)
    if response.status_code != 200:
        print(f"Error fetching {endpoint}: {response.status_code}")
        return {}
    return response.json()

# 특수문자 처리 함수
def escape_single_quotes(text):
    return text.replace("'", "''")

# 데이터 추출 공통 함수 (출연진, 감독, 트레일러, 런타임)
def extract_movie_data(item_id, data_type):
    # 출연진 및 감독 정보 추출
    #1.출연진정보가져오기
    #https://api.themoviedb.org/3/movie/1100099/credits?api_key=키값&language=ko-KR&&page=1
    credits = fetch_from_api(f"{data_type}/{item_id}/credits", {"api_key": API_KEY, "language": LANGUAGE})
    cast = credits.get("cast", [])
    crew = credits.get("crew", [])
    top_actors = ", ".join([escape_single_quotes(actor["name"]) for actor in cast[:5]])  # 따옴표 처리
    director = next((escape_single_quotes(member["name"]) for member in crew if member["job"] == "Director"), "감독 정보 없음")
    
    # 트레일러 URL 추출
    #3.예고편 가져오기
    #https://api.themoviedb.org/3/movie/1100099/videos?api_key=키값&language=ko-KR
    videos = fetch_from_api(f"{data_type}/{item_id}/videos", {"api_key": API_KEY, "language": LANGUAGE})
    youtube_trailer = next(
        (f"{video['key']}" for video in videos.get("results", [])
         if video["site"] == "YouTube" and video["type"] == "Trailer"),
        "MKrNC5GINjI"
    )
    
    # 런타임 추출
    #1.영화 상세보기 https://api.themoviedb.org/3/movie/1100099?api_key=키값&language=ko-KR
    details = fetch_from_api(f"{data_type}/{item_id}", {"api_key": API_KEY, "language": LANGUAGE})
    running_time = details.get("runtime", 0)
    
    return top_actors, director, youtube_trailer, running_time


# SQL 생성 함수
def generate_sql(data, category, data_type):
    sql_list = []
    for item in data:
        if not is_korean(item.get("title" if data_type == "movie" else "name", "")):
            continue
        
        # 공통 데이터 처리
        item_id = item.get("id")
        title = escape_single_quotes(item.get("title" if data_type == "movie" else "name", ""))
        genres = ", ".join([GENRE_MAP.get(genre_id, "") for genre_id in item.get("genre_ids", [])])
        date_key = "release_date" if data_type == "movie" else "first_air_date"
        release_date = item.get(date_key, "")
        year, month = release_date.split("-")[:2] if release_date else ("", "")
        content = escape_single_quotes(item.get("overview", "").replace("'", "''"))
        poster = f"https://image.tmdb.org/t/p/original{item.get('poster_path', '')}"
        price = random.choice([10000, 12000, 15000, 20000])
        
        # 추가 데이터 추출
        top_actors, director, youtube_trailer, running_time = extract_movie_data(item_id, data_type)
        
        # SQL 생성
        sql = f"""
        INSERT INTO tbl_movie (id, category, genre, actor, trailer, price, director, title, content, filename1, opening_year, opening_month, running_time, comment_count, read_count)
        VALUES ('admin', '{category}', '{genres}', '{top_actors}', '{youtube_trailer}', {price}, '{director}', '{title}', '{content}', '{poster}', '{year}', '{month}', {running_time}, 0, 0);
        """
        sql_list.append(sql)
    return sql_list



# 데이터 처리 및 SQL 파일 생성
def process_data():
    categories = {
        "tv": ["on_the_air"],
        "movie": ["top_rated", "now_playing", "upcoming",  "popular"]        
    }
    
    with open(OUTPUT_FILE, "w", encoding="utf-8") as file:
        file.write("-- TMDB Movie and TV Data SQL Dump\n\n")
        
        for data_type, category_list in categories.items():
            for category in category_list:
                print(f"Processing {data_type} data: {category}...")                
                for page in range(5, 0, -1):  # 최신 데이터를 우선
                    params = {"api_key": API_KEY, "language": LANGUAGE, "page": page}
                    data = fetch_from_api(f"{data_type}/{category}", params).get("results", [])
                    sql_list = generate_sql(data, category, data_type)
                    file.writelines(sql_list)
                    print(f"Page {page} processed for {category}.")
    
    print(f"SQL 파일 생성 완료: {OUTPUT_FILE}")

# 실행
if __name__ == "__main__":
    process_data()

 

 

주요 추가 설명

1. category_data_count 딕셔너리: 각 카테고리별로 처리된 데이터 개수를 기록.

 

1. print 출력 개선: 데이터를 처리하면서 현재 상태를 실시간으로 확인 가능하며, 마지막에는 총괄 요약을 출력.

 

실행 후 결과

처리 완료 후 출력되는 메시지 예시:

 

 

=== 데이터 처리 결과 ===
popular: 87개 데이터 처리됨
now_playing: 62개 데이터 처리됨
upcoming: 55개 데이터 처리됨
top_rated: 90개 데이터 처리됨
on_the_air: 40개 데이터 처리됨
Data processing complete. SQL file generated.