Clash of Clans APIからプレイヤーデータをダウンロードする

pip install requests pandas tqdm ratelimit

import time
import math
import json
import requests
import logging
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
from ratelimit import limits, sleep_and_retry
from tqdm import tqdm

file_path = 'others/clans.json'

# Function to open the file and extract tags, specifying the encoding
def extract_tags_from_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:  # Specifying the encoding here
        data = json.load(file)
        return [item.get("tag") for item in data.get("items", [])]

# Extract tags from the specified file
try:
    extracted_tags = extract_tags_from_file(file_path)
    print(extracted_tags)

except UnicodeDecodeError as e:
    print(f"Error reading the file: {e}")

# the extracted clan tags have '#' in beigning we have to replace it with URL encode '%23'
def update_tags(extracted_tags):
    # Replace '#' with '%23' for each tag in the list
    updated_tags = [tag.replace('#', '%23') for tag in extracted_tags]
    return updated_tags

# Get the updated list of tags
updated_extracted_tags = update_tags(extracted_tags)

# Print or return the updated list
print(updated_extracted_tags)

# Replace your API_KEY
api_key = 'your_api_key_here'  

# Base URL for the Clash of Clans API clans endpoint
base_url = 'https://api.clashofclans.com/v1/clans/'

# Header to include in the request
headers = {
   'Authorization': f'Bearer {api_key}',
   'Accept': 'application/json'
}

# Function to get clan member list for each clan tag
def get_clan_members(clan_tags):
    clan_members = {}  # Dictionary to store clan members list by clan tag

    for tag in clan_tags:
        # Constructing the full URL for the clan members endpoint
        full_url = f'{base_url}{tag}/members'
        response = requests.get(full_url, headers=headers)

        if response.status_code == 200:
            # Successful response
            data = response.json()
            # Assuming the API returns a list of clan members directly
            clan_members[tag] = data.get('items', [])

        else:
            # Handle errors or unsuccessful responses
            print(f'Failed to fetch clan members for {tag}: HTTP {response.status_code}')

    return clan_members

# Get clan members for each tag
clan_members_lists = get_clan_members(updated_extracted_tags)

# Example: print the result for the first clan
first_tag = updated_extracted_tags[0]
print(f'Clan members for {first_tag}:', clan_members_lists[first_tag])

# Assuming clan_members_lists is your dictionary from the modified get_clan_members function
def convert_to_dataframe(clan_members_lists):
# Create a list of tuples (clan_tag, player_tag) for all clans
    data = [(clan_tag, player_tag) for clan_tag, player_tags in clan_members_lists.items() for player_tag in player_tags]

# Convert the list of tuples into a DataFrame
    df = pd.DataFrame(data, columns=['Clan Tag', 'Player Tag'])

    return df

# Convert the dictionary to a DataFrame
df_clan_members = convert_to_dataframe(clan_members_lists)
print(df_clan_members)

def convert_to_dataframe(clan_members_lists):
    # Initialize an empty list to store the data
    data = []

    # Loop through each clan tag and its corresponding list of members
    for clan_tag, members in clan_members_lists.items():
        for member in members:
            # For each member, extract the clan tag and the player tag, ensuring the player tag is a string
            player_tag = member['tag']  # Assuming 'tag' key exists and its value is the player's tag
            data.append((clan_tag, player_tag))

    # Convert the list of tuples into a DataFrame
    df = pd.DataFrame(data, columns=['Clan Tag', 'Player Tag'])

    # Optional: Convert clan and player tags to ensure they are URL-friendly
    # This step is optional and depends on whether you need to use these tags in URLs
    df['Clan Tag'] = df['Clan Tag'].apply(lambda x: x.replace('%23', '#'))
    # df['Player Tag'] = df['Player Tag'].apply(lambda x: x.replace('#', '%23'))

    return df

# Example usage with your clan_members_lists dictionary
# Make sure to replace 'clan_members_lists' with your actual dictionary variable
df_clan_members = convert_to_dataframe(clan_members_lists)
print(df_clan_members)

df_clan_members.to_csv('datasets/clan_and_player_tags.csv', index=False)

# Set up logging to file
logging.basicConfig(filename='others/error_log.log', level=logging.ERROR, 
                    format='%(asctime)s - %(levelname)s - %(message)s')

# Load the CSV file into a DataFrame
df = pd.read_csv('datasets/clan_and_player_tags.csv')  # Update this path to your actual CSV file location

# API details
api_key = 'your_api_key_here'
headers = {'Authorization': 'Bearer ' + api_key}

# Rate limit: 10 requests per second (adjust as needed)
@sleep_and_retry
@limits(calls=10, period=1)
def call_api(url):
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

def fetch_player_details(tag):
    url = f'https://api.clashofclans.com/v1/players/{tag.replace("#", "%23")}'
    try:
        data = call_api(url)
        return {
            'name': data.get('name', ''),
            'role': data.get('role', ''),
            'league': data.get('league', {}).get('name', ''),
            'builderBaseLeague': data.get('builderBaseLeague', {}).get('name', ''),
            'townHallLevel': data.get('townHallLevel', 0),
            'builderHallLevel': data.get('builderHallLevel', 0),
            'expLevel': data.get('expLevel', 0),
            'trophies': data.get('trophies', 0),
            'bestTrophies': data.get('bestTrophies', 0),
            'builderBaseTrophies': data.get('builderBaseTrophies', 0),
            'bestBuilderBaseTrophies': data.get('bestBuilderBaseTrophies', 0),
            'attackWins': data.get('attackWins', 0),
            'defenseWins': data.get('defenseWins', 0),
            'warStars': data.get('warStars', 0),
            'clanCapitalContributions': data.get('clanCapitalContributions', 0),
            'donations': data.get('donations', 0),
            'donationsReceived': data.get('donationsReceived', 0),
        }
    except requests.exceptions.RequestException as e:
        logging.error(f'Error fetching data for {tag}: {e}')

def process_batch(batch):
    player_details = []
    with ThreadPoolExecutor(max_workers=20) as executor:
        future_to_tag = {executor.submit(fetch_player_details, tag): tag for tag in batch}
        for future in as_completed(future_to_tag):
            tag = future_to_tag[future]
            try:
                details = future.result()
                player_details.append(details)
            except Exception as exc:
                logging.error(f'{tag} generated an exception: {exc}')

    return player_details

def save_checkpoint(data, filename):
    with open(filename, 'w') as f:
        json.dump(data, f)

def load_checkpoint(filename):
    try:
        with open(filename, 'r') as f:
            return json.load(f)
    except FileNotFoundError:
        return []

def main():
    batch_size = 1000
    checkpoint_file = 'others/player_details_checkpoint.json'

    # Load checkpoint if it exists
    player_details = load_checkpoint(checkpoint_file)
    start_index = len(player_details)

    num_batches = math.ceil((len(df) - start_index) / batch_size)

    for i in tqdm(range(start_index, len(df), batch_size), total=num_batches, desc="Processing batches"):
        batch = df['Player Tag'].iloc[i:i+batch_size].tolist()
        batch_details = process_batch(batch)
        player_details.extend(batch_details)

        # Save checkpoint after each batch
        save_checkpoint(player_details, checkpoint_file)

     # Filter out None values from player_details
    player_details = [detail for detail in player_details if detail is not None]

    # Create a new DataFrame with the player details
    details_df = pd.DataFrame(player_details)

    # Merge the original DataFrame with the new details DataFrame
    final_df = pd.concat([df, details_df], axis=1)

    # Save the final DataFrame to a new CSV file
    final_df.to_csv('datasets/player_details.csv', index=False)
    print("Data fetching and processing complete. Results saved to 'Clan_and_Player_Details.csv'")

if __name__ == "__main__":
    start_time = time.time()
    main()
    end_time = time.time()
    print(f"Total execution time: {end_time - start_time:.2f} seconds")