<commit_before>from contextlib import closing, suppress
from itertools import takewhile
import datetime
import os
import argparse
from json import loads, dumps

from cassiopeia import baseriotapi
from cassiopeia.dto.matchlistapi import get_match_list
from cassiopeia.dto.matchapi import get_match
from cassiopeia.type.api.exception import APIError
from persist import TierStore, JSONConfigEncoder, datetime_to_dict
from data_types import TierSet, TierSeed, Tier, Queue, Maps, slice_time, epoch
from summoners_api import update_participants, summoner_names_to_id, leagues_by_summoner_ids

current_state_extension = '.checkpoint'
delta_3_hours = datetime.timedelta(hours=3)
delta_30_days = datetime.timedelta(days=30)

def make_store_callback(store):
    def store_callback(match_json, tier):
        store.store(match_json, tier)
    return store_callback

def download_matches(store_callback, seed_players_by_tier, minimum_tier = Tier.bronze,
                     start=None,end=None, duration=delta_3_hours,
                     include_timeline=True, matches_per_time_slice=2000,
                     map_type = Maps.SUMMONERS_RIFT, queue=Queue.RANKED_SOLO_5x5, end_of_time_slice_callback=None,
                     prints_on=False, minimum_match_id=0, starting_matches_in_first_time_slice=0):

    if end is None:
        end = datetime.datetime.now()
    else:
        end = min(datetime.datetime.now, end)

    if start is None:
        start = end - delta_30_days
    else:
        start = min(start, end)

    def checkpoint(time_slice, players_to_analyze, total_matches, time_slice_downloaded_matches, max_match_id):
        if prints_on:
                print("{} - Reached the checkpoint."
                      .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), total_matches))
        if end_of_time_slice_callback:
            end_of_time_slice_callback(datetime.datetime.utcfromtimestamp(time_slice.end/1000), players_to_analyze,
                                       total_matches, time_slice_downloaded_matches, max_match_id)

    players_to_analyze = TierSeed(tiers=seed_players_by_tier)

    total_matches = 0
    # We store the maximum match id ever downloaded. Since the set pop is in hash order, and hash(int)=int, we are
    # guaranteed to not have popped any match id greater of the maximum_downloaded_id. When restarting by loading a state
    # we can prevent to store matches with a smaller id in order to avoid match duplicates. This leaves out some
    # matches which could be valid, but since usually it's not important which specific match we download, it's a safe
    # constraint to put in order to allow correct resumes.
    maximum_downloaded_id = 0
    for time_slice in slice_time(start, end, duration):
        if prints_on:
            print("{} - Starting time slice {}. Downloaded {} matches so far.".format(datetime.datetime.now().strftime("%m-%d %H:%M:%S") ,time_slice, total_matches))
        #It's impossible that matches overlap between time slices. Reset the history of downloaded matches
        downloaded_matches_by_tier = TierSet()
        matches_to_download_by_tier = TierSet()
        analyzed_players = TierSeed()

        matches_in_time_slice = starting_matches_in_first_time_slice
        starting_matches_in_first_time_slice = 0
        try:
            # Iterate until matches_in_time_slice is big enough, and stop anyways after matches_per_time_slice iterations
            # this ensures the script will always terminate even in strange situations
            # (like when all our seeds have no matches in the time slice)
            for _ in takewhile(lambda x: matches_in_time_slice <= matches_per_time_slice, range(matches_per_time_slice)):
                for tier in Tier.equals_and_above(minimum_tier):
                    if prints_on:
                        print("{} - Starting player download for tier {}. Players in queue: {}"
                              .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), tier.name, len(players_to_analyze)))

                    for player_id in players_to_analyze.consume(tier, 10):
                        try:
                            match_list = get_match_list(player_id, begin_time=time_slice.begin, end_time=time_slice.end, ranked_queues=queue.name)
                            for match in match_list.matches:
                                match_id = match.matchId
                                if not match_id in downloaded_matches_by_tier[tier] and match_id > minimum_match_id:
                                    matches_to_download_by_tier[tier].add(match_id)

                            analyzed_players[tier].add(player_id)
                        except APIError as e:
                            if 400 <= e.error_code < 600:
                                # Might be a connection problem
                                if prints_on:
                                    print("{} - Encountered error {}"
                                            .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), e))
                                from time import sleep
                                sleep(5)
                                continue
                            else:
                                raise e

                    if prints_on:
                        print("{} - Starting matches download for tier {}. Downloads in queue: {}. Downloaded: {}"
                              .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), tier.name,
                                      len(matches_to_download_by_tier), matches_in_time_slice))

                    for match_id in matches_to_download_by_tier.consume(tier, 10, 0.2):
                        try:
                            match = get_match(match_id, include_timeline)
                            if match.mapId == map_type.value:
                                match_min_tier = update_participants(players_to_analyze, match.participantIdentities, minimum_tier, queue)
                                if match_min_tier.is_better_or_equal(minimum_tier):
                                    maximum_downloaded_id = max(maximum_downloaded_id, match_id)
                                    store_callback(match.to_json(sort_keys=False,indent=None), match_min_tier.name)
                                    matches_in_time_slice += 1
                                downloaded_matches_by_tier[tier].add(match_id)
                        except APIError as e:
                            if 400 <= e.error_code < 600:
                                # Might be a connection problem
                                if prints_on:
                                    print("{} - Encountered error {}"
                                            .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), e))
                                from time import sleep
                                sleep(5)
                                continue
                            else:
                                raise e
                    players_to_analyze -= analyzed_players
            total_matches += matches_in_time_slice
        finally:
            #Always call the checkpoint, so that we can resume the download in case of exceptions.
            checkpoint(time_slice, players_to_analyze, total_matches, matches_in_time_slice, maximum_downloaded_id)

def download_from_config(config, config_file, save_state=True):

    prints_on = config.get('prints_on', False)

    #Set up the api
    cassioepia = config['cassiopeia']
    baseriotapi.set_api_key(cassioepia['api_key'])
    baseriotapi.set_region(cassioepia['region'])

    limits = cassioepia.get('rate_limits', None)
    if limits:
        if isinstance(limits[0], list):<commit_msg>The callback now takes a MatchDescription, not a json anymore<commit_after>from contextlib import closing, suppress
from itertools import takewhile
import datetime
import os
import argparse
from json import loads, dumps

from cassiopeia import baseriotapi
from cassiopeia.dto.matchlistapi import get_match_list
from cassiopeia.dto.matchapi import get_match
from cassiopeia.type.api.exception import APIError
from persist import TierStore, JSONConfigEncoder, datetime_to_dict
from data_types import TierSet, TierSeed, Tier, Queue, Maps, slice_time, epoch
from summoners_api import update_participants, summoner_names_to_id, leagues_by_summoner_ids

current_state_extension = '.checkpoint'
delta_3_hours = datetime.timedelta(hours=3)
delta_30_days = datetime.timedelta(days=30)

def make_store_callback(store):
    def store_callback(match, tier):
        store.store(match.to_json(sort_keys=False,indent=None), tier)
    return store_callback

def download_matches(store_callback, seed_players_by_tier, minimum_tier = Tier.bronze,
                     start=None,end=None, duration=delta_3_hours,
                     include_timeline=True, matches_per_time_slice=2000,
                     map_type = Maps.SUMMONERS_RIFT, queue=Queue.RANKED_SOLO_5x5, end_of_time_slice_callback=None,
                     prints_on=False, minimum_match_id=0, starting_matches_in_first_time_slice=0):

    if end is None:
        end = datetime.datetime.now()
    else:
        end = min(datetime.datetime.now, end)

    if start is None:
        start = end - delta_30_days
    else:
        start = min(start, end)

    def checkpoint(time_slice, players_to_analyze, total_matches, time_slice_downloaded_matches, max_match_id):
        if prints_on:
                print("{} - Reached the checkpoint."
                      .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), total_matches))
        if end_of_time_slice_callback:
            end_of_time_slice_callback(datetime.datetime.utcfromtimestamp(time_slice.end/1000), players_to_analyze,
                                       total_matches, time_slice_downloaded_matches, max_match_id)

    players_to_analyze = TierSeed(tiers=seed_players_by_tier)

    total_matches = 0
    # We store the maximum match id ever downloaded. Since the set pop is in hash order, and hash(int)=int, we are
    # guaranteed to not have popped any match id greater of the maximum_downloaded_id. When restarting by loading a state
    # we can prevent to store matches with a smaller id in order to avoid match duplicates. This leaves out some
    # matches which could be valid, but since usually it's not important which specific match we download, it's a safe
    # constraint to put in order to allow correct resumes.
    maximum_downloaded_id = 0
    for time_slice in slice_time(start, end, duration):
        if prints_on:
            print("{} - Starting time slice {}. Downloaded {} matches so far.".format(datetime.datetime.now().strftime("%m-%d %H:%M:%S") ,time_slice, total_matches))
        #It's impossible that matches overlap between time slices. Reset the history of downloaded matches
        downloaded_matches_by_tier = TierSet()
        matches_to_download_by_tier = TierSet()
        analyzed_players = TierSeed()

        matches_in_time_slice = starting_matches_in_first_time_slice
        starting_matches_in_first_time_slice = 0
        try:
            # Iterate until matches_in_time_slice is big enough, and stop anyways after matches_per_time_slice iterations
            # this ensures the script will always terminate even in strange situations
            # (like when all our seeds have no matches in the time slice)
            for _ in takewhile(lambda x: matches_in_time_slice <= matches_per_time_slice, range(matches_per_time_slice)):
                for tier in Tier.equals_and_above(minimum_tier):
                    if prints_on:
                        print("{} - Starting player download for tier {}. Players in queue: {}"
                              .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), tier.name, len(players_to_analyze)))

                    for player_id in players_to_analyze.consume(tier, 10):
                        try:
                            match_list = get_match_list(player_id, begin_time=time_slice.begin, end_time=time_slice.end, ranked_queues=queue.name)
                            for match in match_list.matches:
                                match_id = match.matchId
                                if not match_id in downloaded_matches_by_tier[tier] and match_id > minimum_match_id:
                                    matches_to_download_by_tier[tier].add(match_id)

                            analyzed_players[tier].add(player_id)
                        except APIError as e:
                            if 400 <= e.error_code < 600:
                                # Might be a connection problem
                                if prints_on:
                                    print("{} - Encountered error {}"
                                            .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), e))
                                from time import sleep
                                sleep(5)
                                continue
                            else:
                                raise e

                    if prints_on:
                        print("{} - Starting matches download for tier {}. Downloads in queue: {}. Downloaded: {}"
                              .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), tier.name,
                                      len(matches_to_download_by_tier), matches_in_time_slice))

                    for match_id in matches_to_download_by_tier.consume(tier, 10, 0.2):
                        try:
                            match = get_match(match_id, include_timeline)
                            if match.mapId == map_type.value:
                                match_min_tier = update_participants(players_to_analyze, match.participantIdentities, minimum_tier, queue)
                                if match_min_tier.is_better_or_equal(minimum_tier):
                                    maximum_downloaded_id = max(maximum_downloaded_id, match_id)
                                    store_callback(match, match_min_tier.name)
                                    matches_in_time_slice += 1
                                downloaded_matches_by_tier[tier].add(match_id)
                        except APIError as e:
                            if 400 <= e.error_code < 600:
                                # Might be a connection problem
                                if prints_on:
                                    print("{} - Encountered error {}"
                                            .format(datetime.datetime.now().strftime("%m-%d %H:%M:%S"), e))
                                from time import sleep
                                sleep(5)
                                continue
                            else:
                                raise e
                    players_to_analyze -= analyzed_players
            total_matches += matches_in_time_slice
        finally:
            #Always call the checkpoint, so that we can resume the download in case of exceptions.
            checkpoint(time_slice, players_to_analyze, total_matches, matches_in_time_slice, maximum_downloaded_id)

def download_from_config(config, config_file, save_state=True):

    prints_on = config.get('prints_on', False)

    #Set up the api
    cassioepia = config['cassiopeia']
    baseriotapi.set_api_key(cassioepia['api_key'])
    baseriotapi.set_region(cassioepia['region'])

    limits = cassioepia.get('rate_limits', None)
    if limits:
        if isinstance(limits[0], list):
{'fraction_non_alphanumeric': 0.07162975291255921, 'fraction_numerical': 0.005889130713096914, 'mean_word_length': 1.9521919879062737, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 94, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5611980', 'n_tokens_mistral': 4364, 'n_tokens_neox': 4053, 'n_words': 1026}