您是《部落衝突》愛好者，同時也熱衷於數據分析或應用程式開發嗎？在這篇文章中，我將引導您完成使用 Python 從《部落衝突》API 下載玩家資料的過程。這對於部落管理工具、玩家統計分析，甚至建立您自己的《部落衝突》相關應用程式非常有用。
在我們深入之前，請確保您已經：
首先，您需要註冊一個《部落衝突》開發者帳號：
我們將在此專案中使用多個 Python 函式庫。使用 pip 安裝它們：
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")
運行腳本：
腳本將批量處理玩家標籤，獲取每個玩家的詳細信息，並將結果保存在新的 CSV 檔案中。
此腳本提供了一種從 Clash of Clans API 下載玩家資料的可靠方法。它包括錯誤處理、速率限制以遵守 API 約束，並使用多線程來提高效能。
請記得始終遵守 API 使用條款和條件。快樂編碼，一起衝突吧！
以上是從《部落衝突》API 下載玩家數據的詳細內容。更多資訊請關注PHP中文網其他相關文章！
