Hsreplay.netCrawler/heros/heroes.py

import requests
import logging
import csv
import os
from datetime import datetime


def get_heroes_info(params, headers, logging):
    def get_battlegrounds_list_heroes():
        url = "https://hsreplay.net/analytics/query/battlegrounds_list_heroes/"
        response = requests.get(url, params=params, headers=headers)
        if response.status_code == 200:
            json_response = response.json()
            data = json_response["series"]["data"]
            return data
        else:
            logging.error(f"请求失败，状态码: {response.status_code}")

    def get_cards():
        url = "https://api.hearthstonejson.com/v1/latest/zhCN/cards.json"
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            json_response = response.json()
            data = json_response
            return data
        else:
            logging.error(f"请求失败，状态码: {response.status_code}")

    def get_composition_affinity_by_hero():
        url = "https://hsreplay.net/analytics/query/battlegrounds_composition_affinity_by_hero/"
        response = requests.get(url, params=params, headers=headers)
        if response.status_code == 200:
            json_response = response.json()
            data = json_response["series"]["data"]
            return data
        else:
            logging.error(f"请求失败，状态码: {response.status_code}")

    def get_compositions():
        url = "https://hsreplay.net/api/v1/battlegrounds/compositions/"
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            json_response = response.json()
            data = json_response
            return data
        else:
            logging.error(f"请求失败，状态码: {response.status_code}")

    compositions = get_compositions()
    heroes_stats = get_battlegrounds_list_heroes()
    cards = get_cards()
    composition_affinity_by_hero = get_composition_affinity_by_hero()

    compositions_mapping = {}
    for entry in compositions:
        compositions_mapping[entry["id"]] = entry["name"]

    mapping = {}
    for entry in cards:
        mapping[entry["dbfId"]] = {}
        mapping[entry["dbfId"]]["name"] = entry["name"]
        if "heroPowerDbfId" in entry:
            mapping[entry["dbfId"]]["heroPowerDbfId"] = entry["heroPowerDbfId"]
        if "text" in entry:
            mapping[entry["dbfId"]]["text"] = entry["text"]
    modified_heroes = []
    for entry in heroes_stats[::-1]:
        hero_dbf_id = entry["hero_dbf_id"]
        name = mapping.get(hero_dbf_id)["name"]
        heroPowerDbfId = mapping.get(hero_dbf_id)["heroPowerDbfId"]
        composition_of_this_hero = composition_affinity_by_hero[str(
            hero_dbf_id)]
        modified_compositions = []
        temp_points = 0
        for entry_composition in composition_of_this_hero:
            temp_points += entry_composition["data_points"]
        if temp_points != 0:
            for entry_composition in composition_of_this_hero:
                new_entry_composition = {
                    "Name":
                    compositions_mapping[entry_composition["composition_id"]],
                    "Popularity":
                    "{:.2f}%".format(
                        (entry_composition["data_points"] / temp_points) *
                        100),
                    "Comp_Impact":
                    entry_composition["comp_impact"]
                }
                modified_compositions.append(new_entry_composition)
        new_entry = {
            "ID": hero_dbf_id,
            "Name": name,
            "Text": mapping.get(heroPowerDbfId)["text"],
            "Pick_Rate": entry["pick_rate"],
            "Avg_Final_Placement": entry["avg_final_placement"],
            "Popularity": entry["popularity"],
            "Final_Placement_Distribution":
            entry["final_placement_distribution"],
            "Compositions": modified_compositions
        }
        modified_heroes.append(new_entry)
    
    output_directory = "output/heroes"
    current_time = datetime.now().strftime("%Y%m%d_%H%M%S")
    csv_filename = f"{current_time}.csv"
    csv_output_file = os.path.join(output_directory, csv_filename)
    os.makedirs(output_directory, exist_ok=True)

    fieldnames = [
        "ID", "Name", "Text", "Pick_Rate", "Avg_Final_Placement", "Popularity",
        "Final_Placement_Distribution", "Compositions"
    ]

    with open(csv_output_file, mode="w", newline="",
              encoding="utf-8") as outfile:
        csv_writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        csv_writer.writeheader()
        csv_writer.writerows(modified_heroes)