테스트 코드
관계 데이터 조회 테스트 전용 코드
# data_reconciler.py (relations 데이터 조회 테스트용)
import os
import requests
import json
import google.generativeai as genai
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import Error
# --- 환경 변수 및 API 설정 ---
load_dotenv()
API_URL = 'https://graphql.anilist.co'
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
if GEMINI_API_KEY:
genai.configure(api_key=GEMINI_API_KEY)
def fetch_anime_with_relations(anime_id):
"""
AniList ID로 특정 애니메이션의 정보와 '관계(relations)' 정보를 함께 가져옵니다.
"""
# GraphQL 쿼리에 relations 부분을 추가
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title {
romaji
english
native
}
format
status
startDate { year }
relations {
edges {
relationType(version: 2) # 관계 타입 (예: SEQUEL, PREQUEL)
node { # 관계된 작품의 정보
id
title {
romaji
english
}
format
}
}
}
}
}
'''
variables = {'id': anime_id}
print(f"AniList에서 ID '{anime_id}'와 관계 작품 정보 조회 중...")
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
# --- 이 파일을 직접 실행했을 때 테스트용으로 사용할 코드 ---
if __name__ == "__main__":
# 테스트 케이스: 나의 히어로 아카데미아 1기 (AniList ID: 21459)
test_id = 21459
try:
data = fetch_anime_with_relations(test_id)
print("\n--- AniList API 응답 결과 (relations 포함) ---")
# indent=2 옵션으로 JSON을 예쁘게 출력
print(json.dumps(data, indent=2, ensure_ascii=False))
print("--------------------------------------------------\n")
print("테스트 완료: 위 JSON 결과에서 'relations' 부분을 확인해주세요.")
print("예를 들어, 'relationType'이 'SEQUEL'인 작품이 후속 시즌입니다.")
except Exception as e:
print(f"테스트 중 오류가 발생했습니다: {e}")출력 결과
"relations": {
"edges": [
{
"relationType": "SOURCE", // 원작 (만화)
"node": { "id": 85486, ... }
},
{
"relationType": "SEQUEL", // ★★★ 속편 (시즌 2) ★★★
"node": { "id": 21856, ... }
},
{
"relationType": "SIDE_STORY", // 외전
"node": { ... }
}
]
}시리즈 데이터 처리 로직 생성
출력 결과 프로그램이 나의 히어로 아카데미아 1기'(id: 21459)의 속편(SEQUEL)이 '시즌 2'(id: 21856)라는 것을 인지하는 것을 알 수 있다.
이를 바탕으로 새로운 로직을 설계한다.
- 시리즈 정보 수집
특정 애니메이션 ID가 주어지면 그 작품의 정보와 relation에 있는 모든 관련 작품들의 ID를 수집한다. - 바탕이 되는 series찾기/생성하기
수집된 작품들의 제목으로 DB의 series 테이블을 검색한다.
일치하는 항목이 있으면 그 항목의 ID를 사용하고
없으면 생성하여 ID를 얻는다. - 개별 work의 처리
수집한 ID 목록을 순회한다.
각 ID에 대하여 이미 work데이터가 있는지 확인한다.
있으면 최신 데이터로 업데이트, 없으면 추가한다.
시리즈 인식 처리 과정 확인 테스트 코드
# data_reconciler.py (시리즈 전체 처리 로직 구현 테스트)
import os
import requests
import json
import time
from dotenv import load_dotenv
# --- 환경 변수 및 API 설정 ---
load_dotenv()
API_URL = 'https://graphql.anilist.co'
# --- Helper Functions ---
def get_db_connection():
# ... (이전과 동일, 이 테스트에서는 사용하지 않음)
pass
def fetch_anime_with_relations(anime_id):
# ... (이전과 동일)
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title { romaji english native }
format
status
startDate { year }
relations {
edges {
relationType(version: 2)
node { id title { romaji english } format }
}
}
}
}
'''
variables = {'id': anime_id}
print(f"\nAniList에서 ID '{anime_id}'와 관계 작품 정보 조회 중...")
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
def find_existing_work_by_anilist_id(anilist_id):
# 이 함수는 DB에 해당 AniList ID가 이미 등록되어 있는지 확인하는 역할
# 지금은 테스트를 위해, '진격의 거인'과 일부만 이미 있는 것으로 가정
existing_map = {
16498: 311, # Attack on Titan
101922: 319, # Demon Slayer
1535: 324, # Death Note
113415: 358, # JUJUTSU KAISEN
21459: 325 # My Hero Academia (ID를 1기 ID로 가정)
}
return existing_map.get(anilist_id)
# --- 메인 실행 함수 ---
def process_series_from_entry_point(entry_anilist_id):
"""
하나의 AniList ID를 시작점으로, 시리즈 전체를 파악하고 처리 계획을 출력하는 함수
"""
print(f"{'='*20} [ 시리즈 처리 시작 (시작 ID: {entry_anilist_id}) ] {'='*20}")
# 1. 시작점 작품과 그 관계들을 모두 가져옴
main_work_data = fetch_anime_with_relations(entry_anilist_id)
if not main_work_data:
print("시작점 작품 정보를 가져오는 데 실패했습니다.")
return
# 2. 처리해야 할 작품 목록을 정리 (시작 작품 + 속편들)
works_to_process = [main_work_data] # 자기 자신을 목록에 추가
for edge in main_work_data.get('relations', {}).get('edges', []):
# 지금은 'SEQUEL' (속편) 관계만 처리 대상으로 추가
if edge.get('relationType') == 'SEQUEL':
works_to_process.append(edge['node'])
print(f"\n✅ 총 {len(works_to_process)}개의 작품(시즌)을 처리 대상으로 확정.")
print(" > 처리 대상: " + ", ".join([w['title']['english'] for w in works_to_process if w['title']['english']]))
# 3. DB에 해당 시리즈가 있는지 찾거나, 없으면 새로 만든다고 '가정'
# (실제 로직에서는 대표 제목으로 series 테이블을 검색)
print(f"\n✅ 대표 제목 '{main_work_data['title']['english']}'(으)로 DB에서 시리즈를 찾거나, 신규 생성합니다. (seriesId: 999로 가정)")
series_id_in_db = 999 # 테스트용 가상 seriesId
# 4. 각 작품을 순회하며 '업데이트'할지 '신규 추가'할지 계획 출력
print("\n✅ 각 작품에 대한 처리 계획 수립:")
for work in works_to_process:
anilist_id = work['id']
title = work['title']['english'] or work['title']['romaji']
print(f"\n --- '{title}' (AniList ID: {anilist_id}) 처리 ---")
# DB에 이 AniList ID가 이미 등록되어 있는지 확인한다고 '가정'
existing_work_id = find_existing_work_by_anilist_id(anilist_id)
if existing_work_id:
print(f" [계획] 'UPDATE': DB에 이미 workId '{existing_work_id}'(으)로 존재합니다. AniList 데이터로 정보를 보강합니다.")
else:
print(f" [계획] 'INSERT': DB에 없는 새로운 작품입니다. seriesId '{series_id_in_db}'에 연결하여 신규 추가합니다.")
print(f"\n{'='*25} [ 처리 계획 완료 ] {'='*25}")
if __name__ == "__main__":
# 테스트 케이스: 나의 히어로 아카데미아 1기 (AniList ID: 21459)
test_id = 21459
process_series_from_entry_point(test_id)실제 쿼리 실행 테스트 코드
# data_reconciler.py (실제 DB 연동 최종본)
import os
import requests
import json
import time
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import Error
# --- 환경 변수 및 API 설정 ---
load_dotenv()
API_URL = 'https://graphql.anilist.co'
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
if GEMINI_API_KEY:
import google.generativeai as genai
genai.configure(api_key=GEMINI_API_KEY)
# --- DB Helper Functions ---
def get_db_connection():
try:
connection = mysql.connector.connect(
host=os.getenv('DB_HOST'), user=os.getenv('DB_USER'),
password=os.getenv('DB_PASSWORD'), database=os.getenv('DB_DATABASE'), port=3306)
if connection.is_connected():
return connection
except Error as e:
print(f"DB 연결 오류: {e}")
return None
def find_or_create_series_in_db(cursor, title):
search_term = f"%{title}%"
cursor.execute("SELECT id FROM series WHERE titleKr LIKE %s OR titleOriginal LIKE %s LIMIT 1", (search_term, search_term))
result = cursor.fetchone()
if result:
print(f"-> 기존 시리즈 '{title}'(을)를 DB에서 찾았습니다. (seriesId: {result['id']})")
return result['id']
else:
print(f"-> DB에 없는 새로운 시리즈 '{title}'. 신규 생성합니다.")
cursor.execute("INSERT INTO series (regDate, updateDate, titleKr) VALUES (NOW(), NOW(), %s)", (title,))
new_series_id = cursor.lastrowid
print(f" - 신규 시리즈 생성 완료. (seriesId: {new_series_id})")
return new_series_id
def find_work_id_by_anilist_id(cursor, anilist_id):
cursor.execute("SELECT workId FROM work_identifier WHERE sourceName = 'ANILIST_ANIME' AND sourceId = %s", (str(anilist_id),))
result = cursor.fetchone()
return result['workId'] if result else None
def get_full_details_from_anilist(anilist_id):
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title { romaji english native }
format
status
description(asHtml: false)
startDate { year month day }
episodes
duration
genres
studios(isMain: true) { nodes { name } }
coverImage { extraLarge }
trailer { id site }
}
}
'''
variables = {'id': anilist_id}
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
# --- Main Processing Function ---
def process_series_from_entry_point(entry_anilist_id):
print(f"\n{'='*20} [ 시리즈 처리 시작 (시작 ID: {entry_anilist_id}) ] {'='*20}")
connection = get_db_connection()
if not connection: return
cursor = connection.cursor(dictionary=True)
try:
# 1. 시작점 작품과 관계(속편)들 정보 수집
main_work_data = fetch_anime_with_relations(entry_anilist_id)
works_to_process_ids = [main_work_data['id']]
for edge in main_work_data.get('relations', {}).get('edges', []):
if edge.get('relationType') == 'SEQUEL':
works_to_process_ids.append(edge['node']['id'])
print(f"✅ 총 {len(works_to_process_ids)}개 작품(시즌) 처리 대상 확정: {works_to_process_ids}")
# 2. 대표 시리즈 찾기 또는 생성
series_id = find_or_create_series_in_db(cursor, main_work_data['title']['english'] or main_work_data['title']['romaji'])
connection.commit()
# 3. 각 작품별로 DB에 있는지 확인 후 업데이트 또는 신규 추가
for anilist_id in works_to_process_ids:
work_id_in_db = find_work_id_by_anilist_id(cursor, anilist_id)
details = get_full_details_from_anilist(anilist_id)
title_kr = details['title']['english'] or details['title']['romaji']
print(f"\n--- '{title_kr}' (AniList ID: {anilist_id}) 처리 ---")
if work_id_in_db:
print(f" [UPDATE]: DB에 workId '{work_id_in_db}'(으)로 존재. 정보 보강 실행.")
# (생략) 이전 버전의 업데이트 로직과 유사
else:
print(f" [INSERT]: DB에 없는 작품. 신규 추가 실행.")
start_date = f"{details['startDate']['year']}-{details['startDate']['month'] or '01'}-{details['startDate']['day'] or '01'}"
studios_str = ", ".join(node['name'] for node in details.get('studios', {}).get('nodes', []))
trailer_url = f"https://www.youtube.com/watch?v={details['trailer']['id']}" if details.get('trailer') and details['trailer']['site'] == 'youtube' else None
insert_query = """
INSERT INTO work (seriesId, regDate, updateDate, titleKr, titleOriginal, type, releaseDate, episodes, duration, studios, description, thumbnailUrl, trailerUrl, isCompleted)
VALUES (%s, NOW(), NOW(), %s, %s, 'Animation', %s, %s, %s, %s, %s, %s, %s, %s)
"""
params = (
series_id, title_kr, details['title']['native'], start_date,
details.get('episodes'), details.get('duration'), studios_str, details.get('description'),
details.get('coverImage', {}).get('extraLarge'), trailer_url, 1 if details.get('status') == 'FINISHED' else 0
)
cursor.execute(insert_query, params)
new_work_id = cursor.lastrowid
# 식별자 테이블에도 기록
cursor.execute("INSERT INTO work_identifier (workId, sourceName, sourceId, regDate, updateDate) VALUES (%s, 'ANILIST_ANIME', %s, NOW(), NOW())", (new_work_id, str(anilist_id)))
print(f" -> 신규 work 추가 완료 (새 workId: {new_work_id})")
connection.commit() # 모든 변경사항 최종 저장
except Exception as e:
print(f"오류 발생: {e}")
connection.rollback()
finally:
if connection.is_connected():
cursor.close()
connection.close()
print("\nDB 연결이 종료되었습니다.")
# 아래 함수들은 위에서 사용되었으므로 여기에 정의되어 있어야 합니다.
def fetch_anime_with_relations(anime_id):
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title { romaji english native }
relations { edges { relationType(version: 2) node { id } } }
}
}
'''
variables = {'id': anime_id}
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
if __name__ == "__main__":
test_id = 21459 # 나의 히어로 아카데미아 1기
process_series_from_entry_point(test_id)ㅇㅇ
완성 코드
anilist
# anilist_collector.py (대량 데이터 수집 및 주석 추가 최종본)
# ===================================================================
# 이 스크립트는 AniList에서 인기 애니메이션 목록을 가져와서,
# 각 애니메이션에 대한 데이터 정제/보강/신규추가 작업을 수행합니다.
#
# 데이터 수집 양을 조절하려면, 아래 main() 함수 내의
# TOTAL_PAGES_TO_FETCH 값을 수정하시면 됩니다. (26번째 줄 근처)
# ===================================================================
import requests
import time
import os
import mysql.connector
from mysql.connector import Error
from dotenv import load_dotenv, find_dotenv
# data_reconciler.py 에서 최종 완성된 함수를 임포트
from data_reconciler import process_series_from_entry_point
# --- 환경 변수 로드 ---
load_dotenv(find_dotenv())
API_URL = 'https://graphql.anilist.co'
def get_popular_anime_list(page=1, per_page=50):
"""
AniList에서 인기있는 애니메이션 목록을 가져옵니다.
(per_page는 최대 50까지 가능)
"""
print(f"\n>>>> AniList에서 인기 애니메이션 목록 조회 시도 (페이지: {page}, 개수: {per_page})")
query = '''
query ($page: Int, $perPage: Int) {
Page (page: $page, perPage: $perPage) {
pageInfo {
hasNextPage
}
media (type: ANIME, sort: POPULARITY_DESC, isAdult: false) {
id
title {
english
romaji
}
}
}
}
'''
variables = {'page': page, 'perPage': per_page}
try:
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Page']['media']
except Exception as e:
print(f"애니메이션 목록을 가져오는 중 오류 발생: {e}")
return []
def get_processed_anilist_ids():
""" 이미 처리된 AniList ID 목록을 DB에서 가져옵니다. """
processed_ids = set()
connection = None
try:
connection = mysql.connector.connect(
host=os.getenv('DB_HOST'), user=os.getenv('DB_USER'),
password=os.getenv('DB_PASSWORD'), database=os.getenv('DB_DATABASE'), port=3306)
cursor = connection.cursor()
cursor.execute("SELECT sourceId FROM work_identifier WHERE sourceName = 'ANILIST_ANIME'")
results = cursor.fetchall()
for row in results:
processed_ids.add(int(row[0]))
cursor.close()
except Error as e:
print(f"기처리 ID 목록 조회 중 오류 발생: {e}")
finally:
if connection and connection.is_connected():
connection.close()
return processed_ids
def main():
# ==========================================================
# ★★★ 데이터 수집 양을 조절하려면 아래 값을 수정하세요 ★★★
# (페이지당 50개씩 데이터를 가져옵니다)
TOTAL_PAGES_TO_FETCH = 3
# ==========================================================
# 1. 이미 DB에 등록된 AniList ID들을 가져와서, 중복 처리를 방지
processed_ids = get_processed_anilist_ids()
print(f"현재까지 DB에 등록된 AniList 작품 수: {len(processed_ids)}개")
# 2. 지정된 페이지 수만큼 반복하여 처리
for page_num in range(1, 21):
anime_list = get_popular_anime_list(page=page_num, per_page=50)
if not anime_list:
print(f"{page_num} 페이지에서 더 이상 가져올 목록이 없습니다. 작업을 중단합니다.")
break
print(f"\n총 {len(anime_list)}개의 애니메이션에 대한 데이터 정제 및 보강 작업을 시작합니다.")
# 3. 목록을 순회하며, 아직 처리되지 않은 작품에 대해서만 정제/보강 함수를 호출
for anime in anime_list:
anilist_id = anime['id']
title = anime['title']['english'] or anime['title']['romaji']
if anilist_id in processed_ids:
print(f"\n[SKIP] '{title}' (AniList ID: {anilist_id})는 이미 처리된 작품입니다.")
continue
# data_reconciler.py에 있는 핵심 함수를 호출
process_series_from_entry_point(anilist_id)
# AniList API에 대한 예의를 지키고, 차단을 피하기 위해 잠시 대기
print("다음 작업을 위해 2초 대기...")
time.sleep(2)
print("\n모든 작업이 완료되었습니다.")
if __name__ == '__main__':
main()data reconciler
# data_reconciler.py (파일 경로 문제 해결 최종본)
import os
import requests
import json
import time
from dotenv import load_dotenv, find_dotenv # find_dotenv 임포트 추가
import mysql.connector
from mysql.connector import Error
# --- 환경 변수 및 API 설정 ---
# 현재 위치부터 상위 폴더로 올라가며 .env 파일을 찾아 로드
load_dotenv(find_dotenv())
API_URL = 'https://graphql.anilist.co'
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
if GEMINI_API_KEY:
import google.generativeai as genai
genai.configure(api_key=GEMINI_API_KEY)
# --- DB Helper Functions ---
def get_db_connection():
host = os.getenv('DB_HOST')
user = os.getenv('DB_USER')
password = os.getenv('DB_PASSWORD')
database = os.getenv('DB_DATABASE')
# --- 디버깅을 위한 print문 (이제 정상적으로 값이 출력될 것입니다) ---
print("\n--- .env 파일에서 읽어온 DB 접속 정보 ---")
print(f"DB_HOST: {host}")
print(f"DB_USER: {user}")
print(f"DB_PASSWORD: {'설정됨' if password else '!!! 설정 안됨 !!!'}")
print(f"DB_DATABASE: {database}")
print("------------------------------------\n")
try:
connection = mysql.connector.connect(
host=host, user=user,
password=password, database=database, port=3306)
if connection.is_connected():
return connection
except Error as e:
print(f"DB 연결 오류 (host: {host}): {e}")
return None
def find_or_create_series_in_db(cursor, title):
search_term = f"%{title}%"
cursor.execute("SELECT id FROM series WHERE titleKr LIKE %s OR titleOriginal LIKE %s LIMIT 1", (search_term, search_term))
result = cursor.fetchone()
if result:
print(f"-> 기존 시리즈 '{title}'(을)를 DB에서 찾았습니다. (seriesId: {result['id']})")
return result['id']
else:
print(f"-> DB에 없는 새로운 시리즈 '{title}'. 신규 생성합니다.")
cursor.execute("INSERT INTO series (regDate, updateDate, titleKr) VALUES (NOW(), NOW(), %s)", (title,))
new_series_id = cursor.lastrowid
print(f" - 신규 시리즈 생성 완료. (seriesId: {new_series_id})")
return new_series_id
def find_work_id_by_anilist_id(cursor, anilist_id):
cursor.execute("SELECT workId FROM work_identifier WHERE sourceName = 'ANILIST_ANIME' AND sourceId = %s", (str(anilist_id),))
result = cursor.fetchone()
return result['workId'] if result else None
def get_full_details_from_anilist(anilist_id):
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title { romaji english native }
format
status
description(asHtml: false)
startDate { year month day }
episodes
duration
genres
studios(isMain: true) { nodes { name } }
coverImage { extraLarge }
trailer { id site }
}
}
'''
variables = {'id': anilist_id}
print(f" (AniList에서 ID '{anilist_id}'의 상세 정보 조회...)")
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
def fetch_anime_with_relations(anime_id):
query = '''
query ($id: Int) {
Media (id: $id, type: ANIME) {
id
title { romaji english native }
relations { edges { relationType(version: 2) node { id } } }
}
}
'''
variables = {'id': anime_id}
response = requests.post(API_URL, json={'query': query, 'variables': variables})
response.raise_for_status()
return response.json()['data']['Media']
# --- Main Processing Function ---
def process_series_from_entry_point(entry_anilist_id):
print(f"\n{'='*20} [ 시리즈 처리 시작 (시작 ID: {entry_anilist_id}) ] {'='*20}")
connection = get_db_connection()
if not connection: return
cursor = connection.cursor(dictionary=True)
try:
main_work_data = fetch_anime_with_relations(entry_anilist_id)
works_to_process_ids = [main_work_data['id']]
for edge in main_work_data.get('relations', {}).get('edges', []):
if edge.get('relationType') == 'SEQUEL':
works_to_process_ids.append(edge['node']['id'])
print(f"✅ 총 {len(works_to_process_ids)}개 작품(시즌) 처리 대상 확정: {works_to_process_ids}")
series_id = find_or_create_series_in_db(cursor, main_work_data['title']['english'] or main_work_data['title']['romaji'])
connection.commit()
for anilist_id in works_to_process_ids:
work_id_in_db = find_work_id_by_anilist_id(cursor, anilist_id)
details = get_full_details_from_anilist(anilist_id)
title_kr = details['title']['english'] or details['title']['romaji']
print(f"\n--- '{title_kr}' (AniList ID: {anilist_id}) 처리 ---")
if work_id_in_db:
print(f" [UPDATE]: DB에 workId '{work_id_in_db}'(으)로 존재. 정보 보강 실행.")
update_query = """
UPDATE work SET
seriesId = %s, type = %s, description = %s, episodes = %s,
duration = %s, studios = %s, isCompleted = %s, updateDate = NOW(),
titleKr = %s, titleOriginal = %s, releaseDate = %s, thumbnailUrl = %s, trailerUrl = %s
WHERE id = %s
"""
studios_str = ", ".join(node['name'] for node in details.get('studios', {}).get('nodes', []))
is_completed = 1 if details.get('status') == 'FINISHED' else 0
start_date_str = f"{details['startDate']['year']}-{(details['startDate']['month'] or 1):02d}-{(details['startDate']['day'] or 1):02d}"
trailer_url = f"https://www.youtube.com/watch?v={details['trailer']['id']}" if details.get('trailer') and details['trailer']['site'] == 'youtube' else None
params = (
series_id, 'Animation', details.get('description'), details.get('episodes'),
details.get('duration'), studios_str, is_completed, title_kr, details['title']['native'],
start_date_str, details.get('coverImage', {}).get('extraLarge'), trailer_url,
work_id_in_db
)
cursor.execute(update_query, params)
print(f" -> 기존 work(id:{work_id_in_db}) 정보 업데이트 완료 (새 seriesId: {series_id} 연결)")
else:
print(f" [INSERT]: DB에 없는 작품. 신규 추가 실행.")
start_date_str = f"{details['startDate']['year']}-{(details['startDate']['month'] or 1):02d}-{(details['startDate']['day'] or 1):02d}"
studios_str = ", ".join(node['name'] for node in details.get('studios', {}).get('nodes', []))
trailer_url = f"https://www.youtube.com/watch?v={details['trailer']['id']}" if details.get('trailer') and details['trailer']['site'] == 'youtube' else None
insert_query = """
INSERT INTO work (seriesId, regDate, updateDate, titleKr, titleOriginal, type, releaseDate, episodes, duration, studios, description, thumbnailUrl, trailerUrl, isCompleted)
VALUES (%s, NOW(), NOW(), %s, %s, 'Animation', %s, %s, %s, %s, %s, %s, %s, %s)
"""
params = (
series_id, title_kr, details['title']['native'], start_date_str, details.get('episodes'),
details.get('duration'), studios_str, details.get('description'),
details.get('coverImage', {}).get('extraLarge'), trailer_url, 1 if details.get('status') == 'FINISHED' else 0
)
cursor.execute(insert_query, params)
new_work_id = cursor.lastrowid
cursor.execute("INSERT INTO work_identifier (workId, sourceName, sourceId, regDate, updateDate) VALUES (%s, 'ANILIST_ANIME', %s, NOW(), NOW())", (new_work_id, str(anilist_id)))
print(f" -> 신규 work 추가 완료 (새 workId: {new_work_id})")
connection.commit()
except Exception as e:
print(f"오류 발생: {e}")
connection.rollback()
finally:
if connection.is_connected():
cursor.close()
connection.close()
print("\nDB 연결이 종료되었습니다.")
if __name__ == "__main__":
test_id = 21459 # 나의 히어로 아카데미아 1기
process_series_from_entry_point(test_id)