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