분석하고싶은코코

Riot API 활용해서 TFT 시즌8 데이터 분석 해보기 - (1) 크롤링 본문

데이터분석

Riot API 활용해서 TFT 시즌8 데이터 분석 해보기 - (1) 크롤링

코코로코코 2023. 3. 3. 13:54
반응형

지난번에 Riot API키 사용 하는법을 포스팅했는데 이를 활용해서 TFT 데이터를 분석해보기로 했다.

 

이번 프로젝트의 목적은 

'TFT안에서 중요 요소(아이템, 증강, 시너지, 유닛)들은 티어마다 선호하는게 다름을 확인하고 승률을 확인해보자!'

가 목표다. 목적은 티어별 다름이 있는지 확인하는 것이지 무조건 달라야한다. 다름을 찾아야한다가 아니다를 생각하면서 진행하였다.

 

프로젝트 과정은

크롤링 - 전처리 - 시각화 및 분석 순으로 진행했다. 생각보다 과정이 길어져서 이번 포스팅에서는 크롤링만 다루기로 했다.

 

 

크롤링

API 호출 함수

def tft_summoner_info_summonerId(summonerId, api_idx): # 소환사 ID로 탐색
    url = f'https://kr.api.riotgames.com/tft/summoner/v1/summoners/{summonerId}'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers = headers)


# 타겟 소환사의 게임 id 가져오기. 기본 50 게임 조회
def tft_match_ids(puuid, api_idx, count=50):
    url = f'https://asia.api.riotgames.com/tft/match/v1/matches/by-puuid/{puuid}/ids?count={count}'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)

def tft_get_Game_info(matchId, api_idx):
    url = f'https://asia.api.riotgames.com/tft/match/v1/matches/{matchId}'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)
    
def tft_tier_summoners(tier, division, api_idx, page=1):
    url = f'https://kr.api.riotgames.com/tft/league/v1/entries/{tier}/{division}?page={page}'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)
    
def tft_challenger_summoners(api_idx):
    url = f'https://kr.api.riotgames.com/tft/league/v1/challenger'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)

def tft_grandmaster_summoners(api_idx):
    url = f'https://kr.api.riotgames.com/tft/league/v1/grandmaster'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)

def tft_master_summoners(api_idx):
    url = f'https://kr.api.riotgames.com/tft/league/v1/master'
    headers["User-Agent"] =  ua.random
    headers["X-Riot-Token"] = api_keys[api_idx]
    return requests.get(url, headers=headers)

개인 프로젝트로 사용하는 API_KEY라서 초당 20개, 2분당 100개에 제한이 걸려서 5개를 발급받아서 사용했기 때문에 매개변수로 api 뽑아 쓸수 있는 인덱스를 받았다.

 

 

 

TFT 티어가 존재하는 유저 정보 가져오기

def getSummoner_df(): 
    
    results = [pd.DataFrame() for _ in range(5)]

    tiers = ['C', 'GM', 'M', 'DIAMOND','PLATINUM','GOLD', 'SILVER', 'BRONZE', 'IRON']
    divisions = ['I', 'II', 'III', 'IV']

    for tier in tqdm(tiers):
        
        summoners = []
        if tier in ['C', 'GM', 'M']:
            if tier == 'C': #챌린저
                for i in range(5):
                    summoners.append(tft_challenger_summoners(i))

            elif tier == 'GM': #그랜드마스터
                for i in range(5):
                    summoners.append(tft_grandmaster_summoners(i))


            elif tier == "M": #마스터
                for i in range(5):
                    summoners.append(tft_master_summoners(i))


            
            for idx, summoner in enumerate(summoners):
                if idx == 0:
                    texts = json.loads(summoner.text)
                    # 대표 DF
                    top_tier_df = pd.DataFrame(texts['entries'])
                    top_tier_df.sort_values('leaguePoints', ascending=False, inplace=True)
                    top_tier_df['tier'] = texts['tier']
                    top_tier_df['leagueId'] = texts['leagueId']
                    top_tier_df['queueType'] = texts['queue']

                    results[idx] = pd.concat([results[idx], top_tier_df])

                else:
                    texts = json.loads(summoner.text)
                    
                    tier_df = pd.DataFrame(texts['entries'])
                    tier_df.sort_values('leaguePoints', ascending=False, inplace=True)
                    
                    tier_df = tier_df[['summonerId', 'summonerName']]
                    
                    results[idx] = pd.concat([results[idx], tier_df])
            
        else:
            div_df = [pd.DataFrame() for _ in range(5)]
            
            for division in divisions:
                for page in range(1,20):
                    tmp = []
                    for i in range(5):
                        tmp.append(tft_tier_summoners(tier, division, i, page))

                    #소환사 정보가 없는 페이지 확인
                    if tmp[0].text == '[]':
                        break
                    else:
                        for i in range(5):
                            div_df[i] = pd.concat([div_df[i], pd.DataFrame(tmp[i].json())])

                    #요청 제한 시간 범위 때문에 딜레이 설정
                    time.sleep(1)
                    
            for i in range(5):
                if i == 0:
                    results[i] = pd.concat([results[i], div_df[i]])
                else:
                    tmp = div_df[i][['summonerId', 'summonerName']]
                    results[i] = pd.concat([results[i], tmp])
                
    return results

상위 3그룹(챌린저, 그랜드마스터, 마스터)는 API를 따로 호출해야해서 따로 진행하고 나머진 티어들은 20페이지까지 받아오고 더이상 유저가 없다면 탐색하지 않도록 했다. 두 API가 받아오는 정보가 조금은 달라서 형태를 맞춰서 데이터프레임으로 반환해주도록 했다.

 

 

소환사 puuid가져오기

def getPuuids(df, tier):
    result = []
    
    flag = df['tier'] == tier
    summonerId_list = ['summonerId', 'summonerId_1', 'summonerId_2', 'summonerId_3', 'summonerId_4']
    
    target = df[flag]
    
    if tier in ['CHALLENGER', 'GRANDMASTER']:
        pass
    else:
        random_sample = np.random.choice(np.arange(len(target)), size = 1000, replace=False)

        target = target.iloc[random_sample]
    
    api_idx = 0
    cnt = 0
    for idx, row in target.iterrows():
        
        if cnt > 90:
            if api_idx==4:
                api_idx=0
            else:
                api_idx+=1
            
            cnt = 0
                
        summonerId = row[summonerId_list[api_idx]]
        
        try:
            response = tft_summoner_info_summonerId(summonerId, api_idx)
            
            result.append(response.json()['puuid'])
        except:
            print(row)
            print(api_idx)
            print(response)
            break
        cnt+=1
        time.sleep(0.05)

    return result

puuid는 소환사를 구분지을 수 있는 글로벌 고유ID이다. 게임 세부정보를 가져오기 위해서는 게임ID가 필요한데 이 게임ID는 puuid로만 요청이 가능해서 소환사들의 puuid를 가져왔다. 그런데 모든 소환사의 게임을 가져오기에는 요청시간이 너무 오래걸리기 때문에 인원 제한이 없는 챌린저, 그랜드마스터를 제외한 티어는 1,000명을 랜덤하게 뽑아서 사용하기로했다.

 

 

게임ID 가져오기

def getGameIds(puuids, tier):
    result = []
    
    data_su = 50
    
    if tier == 'GRANDMASTER':
        data_su = 25
    elif tier != 'CHALLENGER':
        data_su = 15
    
    api_idx = 0
    api_idxs = [0,1,2,3,4,5,6,7]
    cnt = 0
    for puuid in puuids:
            
        flag = True
        while flag:
            try:
                response = tft_match_ids(puuid, api_idx, data_su)
                if response.status_code != 200:
                    raise
                else:
                    result = result + response.json()
                    flag = False
            except:
                api_idx = np.random.choice(api_idxs, size=1, replace=False)[0]
                time.sleep(1)

        cnt+=1
        time.sleep(0.05)
        
        result = list(set(result))

    return result

최근게임을 다르게 가져왔다. 챌린저는 50게임, 그랜드마스터는 25게임, 나머지티어는 15게임으로 각 티어별로 총 15,000개의 게임ID를 가져왔다. 중복된 게임에 대한 정보를 요청하지 않기 위해서 set을 통해서 중복을 제거해서 값을 반환해주도로 했다.

 

 

게임 정보 가져오기

def getGamesDataDf(game_ids):
    result = pd.DataFrame()

    # unix timestamp 
    season8_start_date = '2022-12-07 08:23:00'
    season8_start_date = datetime.datetime.strptime(season8_start_date,'%Y-%m-%d %H:%M:%S').timestamp()

    api_idxs = [0,1,2,3,4,5,6,7]
    api_idx = 0

    for gameId in tqdm(game_ids):
        
        #게임 데이터 가져오기 - 정보 가져올 수 있는 api_key 값 찾기(랜덤 샘플링)
        flag = True
        while flag:
            try:
                    res = tft_get_Game_info(gameId, api_idx)
                    game_data = res.json()['info']
                flag = False
            except:
                api_idx = np.random.choice(api_idxs, size=1, replace=False)[0]
                time.sleep(1)

        # Millisecond -> second
        game_time = (game_data['game_datetime'] / 1000)

        # 시즌 8 게임이 아니라면 패스
        if season8_start_date > game_time:
            continue

        else:
            game_participants = game_data['participants']

            #게임 데이터 추가
            result = pd.concat([result, pd.DataFrame(game_participants)])

#         time.sleep(0.1)

    return result

이제 Riot API의 크롤링의 마지막!! 게임의 세부정보를 가져오는 작업이다. 시즌8에 대한 정보만 필요하기 때문에 시즌8이 시작한 시점을 잡아두고 그 이후 데이터만 받아오도록 했다.

 

최종결과

Untitled

 

 

위 크롤링 결과를 보면 알겠지만 전부 api에서 사용하는 이름으로 정해져있다. 그래서 시각화할때 보기 좋게 바꿔야할 필요가 있다고 생각되어 사전화 하는 방법을 찾아보다 matatft 사이트를 발견했고 화면에 데이터를 뿌려줄때 데이터를 가져오는 링크를 확인해보니 잘 정리되어있었다. 그래서 감사하게도 여기서 내가 원하는 apiName과 한글이름을 가져오기로 했다.

 

matatft 크롤링

kr_json = 'https://data.metatft.com/lookups/Set8_latest_ko_kr.json'
res = requests.get(kr_json).json()

tft_dict = {}
for key in res.keys():
    results = pd.DataFrame()
    for col in res[key]:
        
        tmp = pd.DataFrame({'apiName' : col['apiName'],
                            'korName' : col['name']
                                },index=[0])
        
        results = pd.concat([results,tmp])
        
    tft_dict[key] = results
    
with open('./data/tft_dict.pkl', "wb") as f:
    pickle.dump(tft_dict, f)

증강, 유닛, 아이템, 시너지 각 키워도별로 정리되어 있어서 데이터프레임으로 만들고 딕셔너리 안에 넣어서 사전처럼 활용하기로 했다.

 

 

이제 시각화를 위한 데이터는 준비됐다! 이제 시각화를 해보자!!

반응형

'데이터분석' 카테고리의 다른 글

딥러닝을 통한 게임 봇 탐지를 위한 판단 근거 분석 기법  (0) 2023.03.21
온라인 게임에서의 이상 징후 탐지 기법 조사 및 분류  (0) 2023.03.20
최동원 선수 연봉 예측 프로젝트  (2) 2023.01.12
따릉이 수요량 예측  (0) 2023.01.03
스타벅스 매장은 정말 매장마다 차이가 없을까?? - (1)  (0) 2022.12.12
'데이터분석' Related Articles more