🎮RIOT API 분석 프로젝트(8) - GrandMaster 티어의 DB구축

✅ 배경 : API를 활용해 DB 구축 및 서비스 운영 경험

✅ 목적 : RIOT LOL API를 활용해 '그랜드마스터' 구간의 실제 티어를 확인 가능한 웹 서비스 만들기

✅ 기간 : 2024.08.25 ~ 2024.09.22 (예정)

✅ 인원 : 1명 (2명으로 시작했으나 프로젝트에서 하차하여 혼자 다 하게 되었습니다)

✅ 역할 : python을 활용해 api 호출 및 유저들의 정보가 담긴 DB 생성,

                       SQL 쿼리문을 활용해 데이터 추출 및 가공,

                       Tableau를 활용해 DB 지표 대시보드 생성,

                       html, flask를 활용하여 웹페이지 구현

---

LEAGUE - VR - GRANDMASTER - queue를 통해 summonerID 추출

import requests

def get_grandmaster_summoner_ids(api_key):
    request_header = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0",
        "Accept-Language": "ko,en;q=0.9,en-US;q=0.8",
        "Accept-Charset": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://developer.riotgames.com",
        "X-Riot-Token": api_key  # 환경 변수에 저장된 API 키 사용
    }

    base_url = "https://kr.api.riotgames.com/lol/league/v4/grandmasterleagues/by-queue/RANKED_SOLO_5x5"
    
    # 그랜드마스터 플레이어 데이터를 Riot API에서 가져옴
    response = requests.get(base_url, headers=request_header)
    
    if response.status_code == 200:
        data = response.json()
        summoner_ids = [entry['summonerId'] for entry in data['entries']]
        return summoner_ids
    else:
        return {"error": f"데이터를 가져오는 데 실패했습니다. 상태 코드: {response.status_code}"}

# API 키 설정
api_key = "my_api"

# 함수 호출
grandmaster_summoners = get_grandmaster_summoner_ids(api_key)

# 결과 출력
print(len(grandmaster_summoners))

 

그랜드마스터 티어는 다른 티어들과 달리 tier, division 같은 정보를 필요로 하지 않습니다.

그 대신 Riot API에서는 그랜드마스터와 마스터 티어 데이터를 각각 전용 엔드포인트를 통해 제공하고 있습니다.

이번 코드에서는 그랜드마스터 데이터를 가져오기 위해 아래 엔드포인트를 사용했습니다.

그랜드마스터나 마스터 티어에서는 일반적인 티어/디비전 시스템과 다르게 페이징도 필요하지 않습니다.

 

 

 

SUMMONER - V4 에서 PUUID 추출

import aiohttp
import asyncio
from asyncio_throttle import Throttler  # Throttler를 사용하여 API 요청 수를 제어

# 비동기적으로 소환사 PUUID를 가져오는 함수
async def fetch_summoner_puuid(session, summoner_id, header, throttler):
    url = f"https://kr.api.riotgames.com/lol/summoner/v4/summoners/{summoner_id}"
    header = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0",
        "Accept-Language": "ko,en;q=0.9,en-US;q=0.8",
        "Accept-Charset": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://developer.riotgames.com",
        "X-Riot-Token": "my_api"
    }
    
    async with throttler:  # Throttler로 요청 속도를 제어
        async with session.get(url, headers=header) as response:
            if response.status == 429:  # Rate limit 처리
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"Rate limit exceeded. Sleeping for {retry_after} seconds for {summoner_id}.")
                await asyncio.sleep(retry_after)
                return await fetch_summoner_puuid(session, summoner_id, header, throttler)  # 재귀 호출로 재시도
            elif response.status == 200:
                summoner_data = await response.json()
                return summoner_data.get('puuid')
            else:
                print(f"Error: {response.status} for summoner ID {summoner_id}")
                return None

# 비동기적으로 여러 소환사의 PUUID를 가져오는 함수
async def fetch_all_summoner_puuids(api_key, grandmaster_summoners):
    header = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0",
        "Accept-Language": "ko,en;q=0.9,en-US;q=0.8",
        "Accept-Charset": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://developer.riotgames.com",
        "X-Riot-Token": "my_api"
    }

    summoner_puuid_list = []
    throttler = Throttler(rate_limit=20, period=1)  # 초당 20개의 요청만 허용, riot api가 초당 20개가 한계임

    async with aiohttp.ClientSession() as session:
        tasks = []
        for summoner_id in grandmaster_summoners:
            task = fetch_summoner_puuid(session, summoner_id, header, throttler)  # 비동기 작업 생성
            tasks.append(task)
        
        # 모든 비동기 작업을 병렬로 실행
        responses = await asyncio.gather(*tasks)
        
        for puuid in responses:
            if puuid:
                summoner_puuid_list.append(puuid)

    return summoner_puuid_list

# 비동기 실행을 위한 메인 함수
async def main():
    api_key = "MY_API"  # 환경 변수에서 API 키 가져오기
    
    # 여러 소환사의 PUUID를 비동기로 가져오기
    summoner_puuid_list = await fetch_all_summoner_puuids(api_key, grandmaster_summoners)
    
    # 결과 반환
    return summoner_puuid_list

# 비동기 함수 실행
summoner_puuid_list = await main()

# 결과 출력
print(f"Summoner PUUID List: {summoner_puuid_list}")

 

12분 정도의 시간이 걸려서 해당 데이터가 추출 되었습니다.

이제 SQL DB에 저장하겠습니다. 사실 MATCHID 추출할때 저장하면 되지만 데이터 날아갈까봐 미리 저장합니다!

 

 

 

MYSQL DB에 PUUID 저장 및 불러오기 

import aiomysql
import asyncio


# MySQL에 데이터를 저장하는 함수
async def save_puuids_to_mysql(summoner_puuid_list, db_config):
    # MySQL에 연결
    conn = await aiomysql.connect(
        host=db_config["host"],
        port=db_config["port"],
        user=db_config["user"],
        password=db_config["password"],
        db=db_config["db"],
        charset='utf8mb4'
    )
    
    async with conn.cursor() as cursor:
        for puuid in summoner_puuid_list:
            # 데이터 삽입
            query = """
            INSERT INTO summoner_puuid (puuid)
            VALUES (%s)
            ON DUPLICATE KEY UPDATE puuid=VALUES(puuid)
            """
            await cursor.execute(query, (puuid,))
        
        # 변경 사항 커밋
        await conn.commit()

    # MySQL 연결 종료
    conn.close()

# 비동기 실행을 위한 메인 함수
async def main():
    # MySQL 데이터베이스 정보 설정
    db_config = {
        "host": "localhost",       # MySQL 서버 호스트
        "port": 3306,              # MySQL 포트
        "user": "root",            # MySQL 사용자 이름
        "password": "비밀번호 입력",    # MySQL 비밀번호
        "db": "riot_game_db"       # 사용할 데이터베이스 이름
    }

    # PUUID 데이터를 MySQL에 저장
    await save_puuids_to_mysql(summoner_puuid_list, db_config)

# 비동기 함수 실행
await main()

 

저장은 잘 되었고 다시 제대로 불러올 수 있는지 확인해봅시다.

 

 

import mysql.connector

# MySQL 서버에 연결
connection = mysql.connector.connect(
    host="localhost",         # MySQL 서버 호스트 이름 (보통 로컬일 경우 localhost)
    user="root",      # MySQL 사용자 이름
    password="0224",  # MySQL 사용자 비밀번호
    database="riot_game_db"    # 사용할 데이터베이스 이름
)

# 연결이 성공했는지 확인
if connection.is_connected():
    print("MySQL 연결 성공")
else:
    print("MySQL 연결 실패")

MySQL 서버에 연결하여 DB를 제대로 가지고 왔습니다.

 

 

import pandas as pd

# 쿼리 실행 후 DataFrame으로 변환
query = "SELECT * FROM summoner_puuid"  # 사용할 쿼리
df = pd.read_sql(query, connection)

# DataFrame 출력
print(df)

summoner_puuid_list = df['puuid'].tolist()

 

제가 쓸 테이블만을 SQL에서 가져와서 해당 테이블을 DF화 하고, 필요한 칼럼만을 리스트 형식으로 만들었습니다.

 

 

 

MATCH-V4 에서 MATCHID 가져오기

import aiohttp
import asyncio
from asyncio_throttle import Throttler  # asyncio_throttle 가져오기

# 비동기적으로 소환사 matchID를 가져오는 함수
async def fetch_summoner_matchID(session, puuid, header, throttler):
    url = f"https://asia.api.riotgames.com/lol/match/v5/matches/by-puuid/{puuid}/ids?type=ranked&start=0&count=1"
    
    async with throttler:  # 요청 속도를 제한하는 Throttler 사용
        async with session.get(url, headers=header) as response:
            if response.status == 429:  # Rate limit 처리
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"Rate limit exceeded for {puuid}. Sleeping for {retry_after} seconds.")
                await asyncio.sleep(retry_after)
                return await fetch_summoner_matchID(session, puuid, header, throttler)  # 재시도
            
            if response.status == 200:
                summoner_match_data = await response.json()
                return {'puuid': puuid, 'match_ids': summoner_match_data}
            else:
                print(f"Error: {response.status} for PUUID {puuid}")
                return {'puuid': puuid, 'match_ids': None}

# 비동기적으로 여러 소환사의 matchID를 가져오는 함수
async def fetch_all_summoner_matchIDs(api_key, summoner_puuid_list):
    header = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0",
        "Accept-Language": "ko,en;q=0.9,en-US;q=0.8",
        "Accept-Charset": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://developer.riotgames.com",
        "X-Riot-Token": api_key
    }

    summoner_matchID_list = []
    throttler = Throttler(rate_limit=20, period=1)  # 초당 최대 20개의 요청만 허용

    async with aiohttp.ClientSession() as session:
        tasks = []
        for puuid in summoner_puuid_list:
            task = fetch_summoner_matchID(session, puuid, header, throttler)  # 비동기 작업 생성
            tasks.append(task)
        
        # 모든 비동기 작업을 병렬로 실행
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            if response and response.get('match_ids') is not None:
                summoner_matchID_list.append(response)

    return summoner_matchID_list

# 비동기 실행을 위한 메인 함수
async def main():
    api_key = "my_api"  # 환경 변수에서 API 키 가져오기
    
    # 여러 소환사의 matchID를 비동기로 가져오기
    summoner_matchID_list = await fetch_all_summoner_matchIDs(api_key, summoner_puuid_list)
    
    # 결과 반환
    return summoner_matchID_list

# 비동기 함수 실행
summoner_matchID_list = await main()

# 결과 출력
print(f"Summoner MatchID List: {summoner_matchID_list}")

 

최근 한판만의 정보를 가져옵시다.

MATCHID를 활용해 게임정보를 불러온 다음에 PUUID에 리스트에 있는 정보만을 남겨서 뽑겠습니다.

우선 유저 DB를 구축해야 하니까요. 

 

 

 

 

MYSQL DB에 puuid, matchID 저장

import mysql.connector

# MySQL 데이터베이스 연결
connection = mysql.connector.connect(
    host="localhost",        # MySQL 서버 주소
    user="root",    # MySQL 사용자명
    password="pw", # MySQL 비밀번호
    database="riot_game_db"  # 사용할 데이터베이스 이름
)

# 커서 생성
cursor = connection.cursor()

# puuid와 match_id 데이터를 테이블에 삽입하는 함수
def insert_match_id(puuid, match_id):
    sql = """
    INSERT INTO summoner_list (puuid, match_id)
    VALUES (%s, %s)
    """
    cursor.execute(sql, (puuid, match_id))


# puuid와 각각의 match_id를 테이블에 삽입
for entry in summoner_matchID_list:
    puuid = entry['puuid']
    match_ids = entry['match_ids']
    
    for match_id in match_ids:
        insert_match_id(puuid, match_id)

# 변경사항 커밋
connection.commit()

# 커서와 연결 종료
cursor.close()
connection.close()

제대로 저장됐음을 확인할 수 있습니다.

 

 

 

해당 matchID를 가지고 게임정보 가져오기(puuid,summonerName)

import aiohttp
import asyncio
import mysql.connector
from asyncio_throttle import Throttler

# 비동기적으로 match 데이터를 가져오는 함수
async def fetch_match_data(session, match_id, api_key, throttler):
    url = f"https://asia.api.riotgames.com/lol/match/v5/matches/{match_id}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0",
        "Accept-Language": "ko,en;q=0.9,en-US;q=0.8",
        "Accept-Charset": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://developer.riotgames.com",
        "X-Riot-Token": api_key
    }
    
    async with throttler:  # 요청 속도 제한
        async with session.get(url, headers=headers) as response:
            if response.status == 429:  # Rate limit 처리
                retry_after = int(response.headers.get('Retry-After', 60))  # Retry-After 헤더 확인
                print(f"Rate limit exceeded for match {match_id}. Sleeping for {retry_after} seconds.")
                await asyncio.sleep(retry_after)  # 비동기 대기
                return await fetch_match_data(session, match_id, api_key, throttler)  # 다시 시도
            elif response.status == 200:
                match_data = await response.json()  # JSON 데이터 반환
                queue_id = match_data.get('info', {}).get('queueId', None)
                if queue_id == 420:  # Solo Rank 모드 (Queue ID 420)만 가져옴
                    return match_data
                else:
                    print(f"Skipping match {match_id} because queueId is not 420")
                    return None
            else:
                print(f"Error: {response.status} for match ID {match_id}")
                return None

async def fetch_all_matches(match_ids, api_key):
    throttler = Throttler(rate_limit=20, period=1)  # 초당 20개의 요청으로 제한
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_match_data(session, match_id, api_key, throttler) for match_id in match_ids]
        responses = await asyncio.gather(*tasks)
        return [response for response in responses if response is not None]

# MYSQL 연결 설정
def create_db_connection():
    connection = mysql.connector.connect(
        host="localhost",         # MySQL 서버 호스트 이름
        user="root",              # MySQL 사용자 이름
        password="0224",          # MySQL 사용자 비밀번호
        database="riot_game_db"  # 사용할 데이터베이스 이름
    )
    return connection

# MYSQL에 match_id, puuid, summoner_name 데이터 삽입
def insert_match_data(connection, match_data):
    cursor = connection.cursor()
    inser_query = """
    INSERT INTO matches (match_id, puuid, summoner_name)
    VALUES (%s, %s, %s)
    """
    
    for match in match_data:
        match_id = match.get('metadata', {}).get('matchId', 'Unknown Match ID')
        for participant in match.get('info', {}).get('participants', []):
            data = (
                match_id,
                participant.get('puuid'),
                participant.get('summonerName')
            )
            try:
                cursor.execute(inser_query, data)
            except mysql.connector.Error as err:
                print(f"Error: {err}")
    
    connection.commit()
    cursor.close()

# 비동기 메인 함수
async def main():
    api_key = "my_api"  # Riot Games API 키
    
    # 여러 매치 데이터를 비동기로 가져오기
    match_data_list = await fetch_all_matches(match_ids, api_key)
    
    # MySQL 연결 및 데이터 삽입
    connection = create_db_connection()
    insert_match_data(connection, match_data_list)
    connection.close()

# asyncio 실행
await main()

이 코드를 실행하여 SQL에서 개수를 세어봅시다.

 

 

4910개로 이렇게 많이 나옵니다...

그 이유는 그마 티어의 puuid 추출 > 최근 한판 matchID 추출 > matchID 1개에는 10명의 정보가 들어있기에 참여한 사람의 모든 정보가 제공됨 > 그러므로 그마 티어 puuid 추출한 것과 일치되는 것만을 남겨야 합니다

 

 

GrandMaster 티어에 해당되는 유저만 추출하기

CREATE TABLE GM_users AS
SELECT m.match_id, m.puuid, m.summoner_name
FROM matches m
JOIN summoner_list s ON m.puuid = s.puuid;

 

그래도 여전히 2611개입니다. 일단 그랜드 마스터 티어가 아닌경우는 다 제외하였습니다.  

중복되는 항이 있는지 확인해봅시다. 

 

 

#중복항 있는지 확인
SELECT puuid, COUNT(*)
FROM GM_users
GROUP BY puuid
HAVING COUNT(*) > 1;

SELECT COUNT(*)
FROM (
    SELECT puuid
    FROM GM_users
    GROUP BY puuid
    HAVING COUNT(*) > 1
) AS duplicate_puuid;

 

동일한 puuid가 여러판에 참여하면서 matchID만 다르고 중복된 puuid가 있는 것 같습니다.

 

SELECT *
FROM GM_users
WHERE puuid IN (
    SELECT puuid
    FROM GM_users
    GROUP BY puuid
    HAVING COUNT(DISTINCT summoner_name) > 1
);

puuid가 중복되고 summonerName이 다른 모든 행이 있는지 찾아봤는데 없네요.

닉변한사람이 없나 봅니다. 그러면 일단 새로운 테이블을 만들어서 중복값을 없애줍시다. 

 

 

CREATE TABLE GM_users2 AS
SELECT puuid, summoner_name
FROM GM_users;

puuid 리스트 값은 700개인데 제가 갖고 있는 것인 648개 밖에 안되네요.

어떤 값들이 누락 되었는지 확인해봅시다. duplicated_GM_users라는 테이블을 또 만들었습니다.

duplicated_GM_users 테이블에 있는 puuid와 summoner_list 테이블에 있는 puuid를 비교하여,

duplicated_GM_users 테이블에 없는 puuid 값을 확인하려면 LEFT JOIN 또는 NOT IN을 사용할 수 있습니다.

 

SELECT puuid
FROM summoner_list
WHERE puuid NOT IN (
    SELECT puuid
    FROM duplicated_GM_users
);

SELECT COUNT(*) AS missing_puuid_count
FROM summoner_list
WHERE puuid NOT IN (
    SELECT puuid
    FROM duplicated_GM_users
);

역시 52개가 누락되었네요.

 

원인은 다음과 같이 생각해 볼 수 있습니다.

일부 puuid가 게임 정보 테이블에 존재하지 않아서 누락된 것일 수 있습니다.

matchID에서 그랜드마스터 유저의 게임 정보가 일부 누락된 경우입니다.

만약 matchID로 데이터를 추출하는 과정에서 필터링이 잘못되었거나 일부 데이터가 누락되었다면,

700개의 puuid를 완전히 가져오지 못했을 수 있습니다.

 

누락된 52명에 대해서는 검색했을 시 Summoner not found로 표현하겠습니다.

이렇게 하여 puuid 와 summoner_name으로 조합된 그랜드마스터 유저 648명의 데이터 베이스를 구축하였습니다.

 

이제 유저가 자신의 이름을 검색했을 시, 제가 만든 DB에서 puuid를 호출하여 게임 정보를 가져올수 있습니다!