Hsreplay.netCrawler/minion/minion.py

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


def get_minion_info(params, headers, logging):
    def get_battlegrounds_minion_list():
        url = "https://hsreplay.net/analytics/query/battlegrounds_minion_list/"
        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_aggregates_data(aggregates):
        win_rates = []
        for entry in aggregates:
            total_wins = entry["total_wins"]
            total_losses = entry["total_losses"]
            at_least_one = entry["at_least_one"]
            if total_wins != None and total_losses != None:
                rate = "{:.2f}%".format(total_wins /
                                        (total_wins + total_losses) * 100)
            else:
                rate = "None"
            if at_least_one != None:
                popularity = "{:.2f}%".format(at_least_one)
            else:
                popularity = "None"
            win_rate = {
                "combat_round": entry["combat_round"],
                "win_rate": rate,
                "popularity": popularity
            }
            win_rates.append(win_rate)
        return win_rates

    battlegrounds_minion_list = get_battlegrounds_minion_list()
    cards = get_cards()

    mapping = {}
    for entry in cards:
        mapping[entry["dbfId"]] = {}
        mapping[entry["dbfId"]]["name"] = entry["name"]
        if "text" in entry:
            mapping[entry["dbfId"]]["text"] = entry["text"]
    modified_minion = []
    for entry in battlegrounds_minion_list:
        minion_dbf_id = entry["minion_dbf_id"]
        name = mapping.get(minion_dbf_id)["name"]
        if "text" not in mapping.get(minion_dbf_id):
            text = ""
        else:
            text = mapping.get(minion_dbf_id)["text"]
        normal_aggregates = entry["normal_aggregates"]
        premium_aggregates = entry["premium_aggregates"]
        new_entry = {
            "ID": minion_dbf_id,
            "Name": name,
            "Text": text,
            "Minion_Tier": entry["minion_tier"],
            "Normal": get_aggregates_data(normal_aggregates),
            "Premium": get_aggregates_data(premium_aggregates),
        }
        modified_minion.append(new_entry)

    output_directory = "output/minion"
    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", "Minion_Tier", "Normal", "Premium"]

    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_minion)