Herunterladen von Spielerdaten von der 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")