Fflogs网站api

Posted by TheD Blog on June 3, 2024

fflogs第二版api采用了 GraphQL API,功能较为强大,其具体的api文档可以在这里找到:

url: https://www.archon.gg/ffxiv/articles/help/api-documentation  
title: "API Documentation"  
description: "Information on the API documentation of FF Logs."  
host: www.archon.gg  
favicon: https://assets.rpglogs.com/img/archon/logo.svg  
image: https://assets.rpglogs.com/img/ff/featured-article-fallback.jpg  

认证

第二版api采用 OAuth认证,需要注册客户端后获取该次会话的token:

def get_access_token():  
# Your Client ID and Client Secret  
client_id = secret['client_id']  
client_secret = secret['client_secret']  
  
# OAuth Token URL  
token_url = 'https://www.fflogs.com/oauth/token'  
  
# Prepare the authentication request  
auth_response = requests.post(  
    token_url,  
    auth=(client_id, client_secret),  
    data={'grant_type': 'client_credentials'},  
    timeout=10  
)  
  
# Extract the access token from the response  
access_token = auth_response.json().get('access_token')  
return access_token  

数据获取

数据分为公开数据和私有数据,一个report如果被设置为公开,则可以通过公开数据的api直接查到,如果未公开则需要通过私有api。api的查询方式是一样的,能获取的数据可以通过这里查看:

url: https://www.fflogs.com/v2-api-docs/warcraft/  
title: "Graphql schema documentation"  
host: www.fflogs.com  

根据用户名获取绝本相关数据的一个demo,从中可以看到api有速率限制,达到限制后会返回429状态码:

def fetch_graphql_data(  
    character_name, group_name, client, start_time=start_time  
):  
  # 定义GraphQL端点和查询  
  url = 'https://www.fflogs.com/api/v2/client'  
  headers = {  
      'Authorization': f'Bearer {client}',  
      'Content-Type': 'application/json'  
  }  
  query = f"""  
    query {{  
        characterData {{  
            character(  
                name: "{character_name}"  
                serverSlug: "{group_name}"  
                serverRegion: "cn"  
            ) {{  
                UCoB: encounterRankings(encounterID: 1060)  
                UwU: encounterRankings(encounterID: 1061)  
                TEA: encounterRankings(encounterID: 1062)  
                DSR: encounterRankings(encounterID: 1065)  
                TOP: encounterRankings(encounterID: 1068)  
            }}  
        }}  
    }}  
    """  
  response = requests.post(  
      url, json={'query': query}, headers=headers, timeout=10  
  )  
  for i in range(10):  
    if response.status_code == 200:  
      break  
    if response.status_code == 401:  
      # 如果访问令牌过期,重新获取访问令牌并重试  
      logging.info("Access token expired, refreshing...")  
      client = get_access_token()  
      headers['Authorization'] = f'Bearer {client}'  
      response = requests.post(  
          url, json={'query': query}, headers=headers, timeout=10  
      )  
    elif response.status_code == 429:  
      # 如果达到速率限制,等待一段时间后重试  
      logging.info("Rate limit reached, waiting for 1 hour...")  
      time_struct = time.localtime(start_time + 3660)  
  
      print("sleep until", time.strftime("%Y-%m-%d %H:%M:%S", time_struct))  
      time.sleep((start_time + 3660 - time.time()) % 3660)  
      start_time = time.time()  
      response = requests.post(  
          url, json={'query': query}, headers=headers, timeout=10  
      )  
  if i == 9:  
    raise requests.exceptions.HTTPError(  
        f"Failed to fetch data for {character_name} in {group_name}, status code: {response.status_code}"  
    )  
  
  return response.json(), client, start_time