Source code for d2api.src.entities

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import os
import shutil

import requests

# TODO : Implement cleaner way to ensure data is up to date and multi-language compliant
# This appears to be an especially weird problem since the data has to parsed from a local Dota 2 installation

_here = os.path.abspath(os.path.dirname(__file__))

def _ensure_data_folder():
    """Helper method to ensure data folder existence."""
    directory = os.path.abspath(os.path.join(_here, '..', 'ref'))
    if not os.path.exists(directory):
        os.makedirs(directory)

def _load_local_json(file_name):
    """Helper method to read local data."""
    try:
        p = os.path.abspath(os.path.join(_here, '..', 'ref', file_name))
        with open(p, 'r') as f:
            return json.load(f)
    except IOError:
        return {}

def _load_remote_json(file_name):
    """Helper method to retrieve remote data."""
    url = "https://raw.githubusercontent.com/whoophee/d2api/master/d2api/ref/{}".format(file_name)
    res = requests.get(url)
    if res.status_code == 200:
        return res.json()
    else: # pragma: no cover
        return {}

def _write_local_json(data, file_name):
    """Helper method to write local data."""
    p = os.path.abspath(os.path.join(_here, '..', 'ref', file_name))
    with open(p, 'w') as outfile:
        if file_name == 'meta.json':
            json.dump(data, outfile, sort_keys=True, indent=4)
        else:
            json.dump(data, outfile)

all_heroes = _load_local_json('heroes.json')
all_items = _load_local_json('items.json')
all_abilities = _load_local_json('abilities.json')


# Most ID based response values have more data associated with them. 
# This wrapper helps fetch them without having to use auxillary/helper functions.
class Entity(dict):
    """Generic entity class"""
    def __str__(self):
        return self.__repr__()

[docs]class Hero(Entity): """Wrapper to map hero information to hero_id :var hero_id: Unique identifier of hero :var hero_name: Name of the hero :vartype hero_id: int :vartype hero_name: str """ def __repr__(self): return "Hero(hero_id = {})".format(self['hero_id']) def __bool__(self): return self['hero_id'] != None def __init__(self, hero_id): if hero_id != None: hero_id = str(hero_id) self['hero_id'] = hero_id cur_hero = all_heroes.get(hero_id, {}) self['hero_name'] = cur_hero.get('hero_name', 'unknown_hero')
[docs]class Item(Entity): """Wrapper to map item information to item_id :var item_id: Unique identifier of item :var item_cost: Cost of the item :var item_aliases: List of names by which the item is known :var item_name: Name of the item :vartype item_id: int :vartype item_cost: int :vartype item_aliases: list(str) :vartype item_name: str """ def __repr__(self): return "Item(item_id = {})".format(self['item_id']) def __bool__(self): return self['item_id'] != None def __init__(self, item_id): if item_id != None: item_id = str(item_id) self['item_id'] = item_id cur_item = all_items.get(item_id, {}) self['item_cost'] = cur_item.get('item_cost', 0) self['item_aliases'] = cur_item.get('item_aliases', []) self['item_name'] = cur_item.get('item_name', 'unknown_item')
[docs]class Ability(Entity): """Wrapper to map ability data to ability_id :var ability_id: Unique identifier of ability :var ability_name: Name of the ability :vartype ability_id: int :vartype ability_name: str """ def __repr__(self): return "Ability(ability_id = {})".format(self['ability_id']) def __bool__(self): return self['ability_id'] != None def __init__(self, ability_id): if ability_id != None: ability_id = str(ability_id) self['ability_id'] = ability_id cur_ability = all_abilities.get(ability_id, {}) self['ability_name'] = cur_ability.get('ability_name', 'unknown_ability')
# Removes the hassle of having to manually convert between Steam 32-bit/64-bit IDs
[docs]class SteamAccount(Entity): """Wrapper to implicitly store steam32 and steam64 account IDs :var id32: 32-bit Steam ID :var id64: 64-bit Steam ID :vartype id32: int :vartype id64: int """ def __repr__(self): return "SteamAccount(account_id = {})".format(self['id32']) def __bool__(self): return self['id32'] != None def __init__(self, account_id): if account_id is None: self['id32'] = self['id64'] = None else: account_id = int(account_id) steam64 = 76561197960265728 if account_id < steam64: self['id32'] = account_id self['id64'] = account_id + steam64 else: self['id32'] = account_id - steam64 self['id64'] = account_id
def _update(purge): """Helper function to synchronize local with remote data.""" global all_heroes global all_items global all_abilities try: path = os.path.abspath(os.path.join(_here, '..', 'ref')) if purge and os.path.exists(path): shutil.rmtree(path) # Find version of local data local_meta = _load_local_json('meta.json') local_version = local_meta.get('version') # find version of remote data remote_meta = _load_remote_json('meta.json') remote_version = remote_meta.get('version') # update local files if they're outdated if local_version != remote_version: _ensure_data_folder() _write_local_json(remote_meta, "meta.json") for content_name in remote_meta.get('content_files', []): remote_content = _load_remote_json(content_name) _write_local_json(remote_content, content_name) all_heroes = _load_local_json('heroes.json') all_items = _load_local_json('items.json') all_abilities = _load_local_json('abilities.json') return remote_meta except: # pragma: no cover return {}