# def task_fight(self): # if self.restricted_ip: # self.tasks.pop("fight") # return # elif self.now > utils.localize_dt(datetime(2021, 2, 8)): # self.write_warning("Fight is now disabled") # self.tasks.pop("fight") # return # # self.write_log(self.health_info) # # count, log_msg, force_fight = self.should_fight(False) # # if count or force_fight: # fought = False # if self.tasks.get_default("cheap_medals", False): # cheap_medals = self.get_cheap_tp_divisions() # air_div = ground_div = None # if self.config.air and cheap_medals["air"]: # air_div = cheap_medals["air"][0] # if self.config.ground and cheap_medals["ground"]: # ground_div = cheap_medals["ground"][0] # if air_div and (not air_div[0] or not self.config.ground): # div_data = air_div # else: # div_data = ground_div # if div_data: # medal_damage, division = div_data # self.write_log(f"Chose {division} with {medal_damage}dmg medal") # if self.change_division(division.battle, division): # side: BattleSide # if division.battle.defender.country == self.details.citizenship: # side = division.battle.defender # else: # side = division.battle.invader # # air = int(division.is_air) # tgt_dmg = DMG_MAP[division.div][side.is_defender] # hit = self.get_air_hit_dmg_value() if air else self.get_ground_hit_dmg_value() # # tgt_dmg = tgt_dmg if tgt_dmg > medal_damage else medal_damage # # hits = min((self.energy.food_fights - 5, max((count, int(tgt_dmg / hit) + 1)))) # self.set_default_weapon(division.battle, division) # # if self.details.current_country not in side.deployed + [side.country]: # self.travel_to_battle(division.battle, side.deployed + [side.country]) # fought = bool(self.fight(division.battle, division, side, hits + 2)) # battle_count = len(cheap_medals["air"]) + len(cheap_medals["ground"]) # self.report_action( # "CHEAP MEDAL CHOOSER", # f"Had {battle_count} options and chose {div_data}", # kwargs=dict(cheap_medals=cheap_medals), # ) # if not fought: # self.find_battle_and_fight() # # self.travel_to_residence() # self.update_weekly_challenge() # # self.tasks["fight"] = self._get_next_fight_time() # def _get_required_fight_energy(self) -> int: # def _check_energy(__ec: List[Tuple[int, str]], __ne: int, __msg: str) -> int: # __ec.append((__ne, __msg)) # return min(__ec, key=lambda _x: _x[0])[0] # # e_checks: List[Tuple[int, str]] = [] # # Default - fight when full energy # energy = self.energy.limit * 2 - self.energy.interval * 2 # e_checks.append((energy, f"Full energy: {self.energy.limit} * 2 - {self.energy.interval} * 2")) # # if self.is_levelup_reachable: # energy = 2 * (self.energy.limit - self.energy.interval) # msg = f"Levelup reachable: 2 * ({self.energy.limit} - {self.energy.interval})" # e_checks.append((energy, msg)) # elif self.is_levelup_close: # energy = (self.details.xp_till_level_up + 5) * 10 - self.energy.limit # msg = f"Near level up: ({self.details.xp_till_level_up} + 5) * 10 - {self.energy.limit}" # e_checks.append((energy, msg)) # else: # # Obligatory need 75pp # if self.details.pp < 75: # energy = _check_energy(e_checks, 75 - self.details.pp, f"Obligatory need 75pp: 75 - {self.details.pp}") # # # Continuous fighting # if self.config.continuous_fighting and self.has_battle_contribution: # energy = _check_energy(e_checks, self.energy.interval, f"Continuous_fighting: {self.energy.interval}") # # # All-in for AIR battles # if self.config.all_in: # needed = self.energy.limit * 2 - self.energy.interval * 3 # energy = _check_energy(e_checks, needed, f"All-in: {needed}") # # # Get to next Energy +1 # if self.config.next_energy and self.details.next_pp: # self.details.next_pp.sort() # energy = _check_energy( # e_checks, # (self.details.next_pp[0] - self.details.pp) * 10, # f"Get to next Energy +1: ({self.details.next_pp}[0] - {self.details.pp}) * 10", # ) # for e, i in e_checks: # self.logger.debug(f"{e}hp {i}") # return energy # def _get_next_fight_time(self) -> datetime: # needed_energy = self._get_required_fight_energy() - self.energy.energy # if needed_energy > 0: # next_minutes = max([6, needed_energy // self.energy.interval * 6]) # else: # next_minutes = 0 # next_wc_start_fight_time = _norm(self.next_wc_start + timedelta(minutes=6)) # next_fight_time = _norm(self.energy.reference_time + timedelta(minutes=next_minutes)) # # return min((next_wc_start_fight_time, next_fight_time)) # def task_epic_hunt(self): # if self.restricted_ip: # self.tasks.pop("epic_hunt") # return # elif self.now > utils.localize_dt(datetime(2021, 2, 8)): # self.write_warning("Fight is now disabled") # self.tasks.pop("fight") # return # # self.update_war_info() # epic_divs: Generator[BattleDivision, Any, None] = ( # div for battle in self.all_battles.values() for div in battle.div.values() if div.epic # ) # # for div in epic_divs: # if div.div_end: # continue # air = bool(div.is_air and self.config.air) # ground = bool(self.config.ground and bool(div.div == self.division or self.maverick)) # if air or ground: # target_div = div # battle = target_div.battle # invaders = [battle.invader.country] + battle.invader.deployed # side = battle.invader if self.details.citizenship in invaders else battle.defender # self.change_division(target_div.battle, target_div) # # if self.details.current_country not in side.deployed + [side.country]: # self.travel_to_battle(battle, side.deployed + [side.country]) # self.fight(target_div.battle, target_div, side, self.energy.food_fights) # self.travel_to_residence() # break # def _get_hits_for_dmg(self, is_tp: bool, damage: int) -> int: # booster = self.get_active_ground_damage_booster() # booster_kwargs = {} # if booster: # booster_kwargs.update({f"booster_{booster}": True}) # hit_dmg = self.get_ground_hit_dmg_value(tp=is_tp, **booster_kwargs) # return int(damage / hit_dmg + 1) # def get_medal_damage(self, division: BattleDivision, side: BattleSide): # medal = self.get_battle_round_data(division) # medal = medal[side.is_defender] # if medal: # return medal.get("1").get("raw_value") # return 0 # def _pay_for_hit(self, division: BattleDivision, is_defender: bool): # hit_dmg = self.get_division_max_hit(division) # max_hit_fighter_id = None # hit_price = self._default_pay_table[division.div] # if not division.div == self.division and hit_dmg != 10_000: # data = self.get_battle_round_data(division)[is_defender] # for fighter in data.values(): # hover_card_data = self._get_main_citizen_hovercard(fighter.get("citizenId")).json() # hc_md = hover_card_data["fighterInfo"]["military"] # if not hc_md["division"] == division.div: # continue # base_hit = hc_md["damagePerHit"] # manual_base_hit = utils.calculate_hit( # hc_md["strength"], # hc_md["rank"], # hover_card_data["citizenship"]["id"] == self.details.citizenship, # False, # False, # ) # if int(hit_dmg) in [int(base_hit), int(manual_base_hit)] or abs(base_hit - hit_dmg) <= 2: # max_hit_fighter_id = fighter.get("citizenId") # break # else: # for fighter in data.values(): # hover_card_data = self._get_main_citizen_hovercard(fighter.get("citizenId")).json() # if hover_card_data["fighterInfo"]["military"]["division"] == division.div: # max_hit_fighter_id = fighter.get("citizenId") # break # else: # self.report_error("Unable to find player to whom I should pay for the hit!") # if max_hit_fighter_id: # if max_hit_fighter_id in self._preset_pay_table: # hit_price = self._preset_pay_table[max_hit_fighter_id] # else: # fighter_profile = self.get_citizen_profile(max_hit_fighter_id) # about_me_price = re.findall(r"(\d{3,})", fighter_profile["aboutMe"]) # if about_me_price: # hit_price = int(min(about_me_price, key=int)) # if max_hit_fighter_id in self._preset_pay_table: # if self._hits_to_pay[max_hit_fighter_id] + hit_price < 10 * hit_price and not self.stop_threads.is_set(): # self._hits_to_pay[max_hit_fighter_id] += hit_price # hit_price = 0 # self.report_action( # "PAYMENT_AGGREGATION", # f"Current debt for player #{max_hit_fighter_id} is {self._hits_to_pay[max_hit_fighter_id]}cc", # _hits_to_pay=self._hits_to_pay, # ) # else: # hit_price = self._hits_to_pay[max_hit_fighter_id] + hit_price # self.report_action( # "PAYMENT_AGGREGATION", # f"Debt for player #{max_hit_fighter_id} is cleared ({hit_price}cc)", # _hits_to_pay=self._hits_to_pay, # ) # self._hits_to_pay[max_hit_fighter_id] = 0 # if hit_price and max_hit_fighter_id: # # self.write_log(f"self.donate_money({max_hit_fighter_id}, {hit_price}, 1)") # self.donate_money(max_hit_fighter_id, hit_price, 1) # return True # return False # def _sub_task_fight_in_empty_battle(self, queue: List[_EmptyMedal], _processed_divisions: Set[int]): # while not self.stop_threads.is_set(): # try: # if self.stop_threads.is_set(): # return # try: # queue.sort(key=attrgetter("time", "round", "division_id")) # next_medal = queue.pop(0) # except IndexError: # next_medal = None # # if next_medal is not None: # self.stop_threads.wait(utils.get_sleep_seconds(next_medal.time)) # if self.stop_threads.is_set(): # break # elif self.now > utils.localize_dt(datetime(2021, 2, 8)): # self.write_warning("Fight is now disabled") # return # try: # self.update_war_info() # self._update_lock.clear() # try: # battle: Optional[Battle] = self.all_battles[next_medal.battle_id] # except KeyError: # self.report_error( # f"Battle '{next_medal.battle_id}' not found in all_battles " # f"(Medal data: {next_medal})", # sentry=False, # ) # self.report_action( # "EMPTY_BH_ERROR", # f"Error while checking {next_medal.battle_id}\n" # f"https://www.erepublik.com/en/military/battlefield/{next_medal.battle_id}", # ) # raise ErepublikException("No battle data!") # # if battle is None: # raise ErepublikException("No battle or division data!") # elif battle.zone_id != next_medal.round: # raise ErepublikException("Wrong battle round!") # # try: # division: Optional[BattleDivision] = battle.div[next_medal.division_id] # except KeyError: # self.report_error( # f"Division '{next_medal.division_id}' not found in " # f"battle.div (Medal data: {next_medal})", # sentry=False, # ) # self.report_action("EMPTY_BH_ERROR", f"Error while checking {repr(battle)}\n{battle.link}") # raise ErepublikException("No division data!") # # if division is None: # raise ErepublikException("No battle or division data!") # elif division.is_air: # raise ErepublikException("Air battle!") # elif division.div_end: # raise ErepublikException("Division finished") # # if not self.maverick and not division.div == self.division: # raise ErepublikException("Don't have a MP can't fight in non native divisions!") # side = battle.defender if next_medal.defender_side else battle.invader # # try: # if self.details.current_country not in side.deployed + [side.country] or battle.is_rw: # self.change_division(battle, division, side) # medal_damage = self.get_medal_damage(division, side) # except AttributeError: # battle = self.all_battles.get(next_medal.battle_id) # side = battle.defender if next_medal.defender_side else battle.invader # if self.details.current_country not in side.deployed + [side.country] or battle.is_rw: # self.change_division(battle, division, side) # medal_damage = self.get_medal_damage(division, side) # # if medal_damage <= DMG_MAP[division.div][next_medal.defender_side] / 2: # damage_amount = DMG_MAP[division.div][next_medal.defender_side] # if division.div != self.division: # Maverick # hit = self.get_division_max_hit(division) # if hit > 200_000: # my_profile = self._get_main_citizen_profile_json(self.details.citizen_id).json() # rang = my_profile["military"]["militaryData"]["ground"]["rankNumber"] # active_booster = self.get_active_ground_damage_booster() # hit_dmg = utils.get_final_hit_dmg( # hit, rang, True, self.details.is_elite, booster=active_booster # ) # ground_hits = int(damage_amount / hit_dmg + Decimal("0.49")) # else: # ground_hits = self.energy.limit * 2 # # else: # ground_hits = self._get_hits_for_dmg(True, damage_amount) # fight = False # hits_ok = self.my_companies.ff_lockdown < ground_hits < self.energy.food_fights # if not hits_ok and division.div != self.division: # hits_ok = self.my_companies.ff_lockdown < ground_hits < self.energy.food_fights # if hits_ok: # self._update_lock.set() # self.change_division(battle, division, side) # fight = bool(self.fight(battle, division, side, ground_hits)) # if fight and not division.div == self.division: # self._pay_for_hit(division, side.is_defender) # elif next_medal.second_attempt: # bombs = self.inventory.final.get("bomb", {}) # if 215 in bombs: # bomb_id = 215 # bomb_data = bombs.get(215) # for bomb_id, bomb_data in bombs.items(): # if (bomb_id == 21 and division.div != self.division) or bomb_id == 216: # continue # if bomb_data.get("fire_power"): # bombs_required = int(damage_amount / bomb_data.get("fire_power")) # if bomb_data.get("amount") >= bombs_required: # self._update_lock.set() # fight = True # if self.details.current_country not in side.deployed + [side.country]: # self.travel_to_battle(battle, side.deployed + [side.country]) # self.change_division(battle, division) # bombs_deployed = self.deploy_bomb( # battle, division, bomb_id, not side.is_defender, bombs_required # ) # self.report_action( # "CLEARED_MEDAL", # f"Deployed {bombs_deployed} bombs in " # f"{division.div} in battle {battle.id}", # extra_info=dict( # battle=battle, # division=division, # next_medal=next_medal, # bombs_deployed=bombs_deployed, # bomb_data=bomb_data, # ), # ) # # break # if fight: # self.report_action( # "EMPTY_BH", f"Cleared empty d{division.div} BH in {battle}", battle=battle.as_dict # ) # elif not next_medal.second_attempt: # next_medal.second_attempt = True # next_medal.time = _norm(self.now + timedelta(minutes=10)) # queue.append(next_medal) # else: # if division.id in _processed_divisions: # _processed_divisions.remove(division.id) # # except ErepublikException: # continue # except: # noqa # self.report_error(f"Thread {threading.current_thread().name} ran into an error") # finally: # self._update_lock.set() # else: # self.travel_to_residence() # self.stop_threads.wait(60) # battle = division = side = damage_amount = bombs = ground_hits = None # noqa # bombs_required = bombs = fight = bombs_deployed = next_medal = None # noqa # except: # noqa # self.report_error(f"Thread {threading.current_thread().name} ran into an error") # # # def background_task_clear_bhs(self, minute: Union[int, bool]): # _processed_divisions: Set[int] = set() # if self.now > utils.localize_dt(datetime(2021, 2, 8)): # self.write_warning("Fight is now disabled") # return # if minute: # try: # minute = 15 if isinstance(minute, bool) else int(minute) # except ValueError: # return # else: # return # next_check: datetime = self.now # next_check: datetime = _norm(next_check.replace(minute=next_check.minute // 5 * 5, second=0)) # queue: List[_EmptyMedal] = [] # # empty_bh_thread = threading.Thread( # target=self._sub_task_fight_in_empty_battle, # args=(queue, _processed_divisions), # name=self.thread_name("empty_bh_queue"), # ) # empty_bh_thread.start() # self._bg_task_queue.append(empty_bh_thread) # # while not self.stop_threads.is_set(): # if self.now > utils.localize_dt(datetime(2021, 2, 8)): # self.write_warning("Fight is now disabled") # return # try: # self.update_war_info() # # Check TP battles # tp_battles = self.sorted_battles(True, True) # # # Removes old division ids from processed division ids set {1,2,3}.inter_update({3,4,5}) == {3} # _processed_divisions.intersection_update({d for b in tp_battles for d in b.div.keys()}) # for battle in tp_battles: # if self.details.citizenship.id in [battle.defender.id, battle.invader.id]: # if battle.has_air: # continue # else: # fight_time = _norm(battle.start + timedelta(minutes=minute)) # defender_side = battle.defender.id == self.details.citizenship.id # battle_stats = self.get_battle_division_stats(list(battle.div.values())[0]) # try: # round_stats = battle_stats.get("stats").get("current").get(f"{battle.zone_id}") # for div in battle.div.values(): # td = ( # round_stats.get(str(div.div)) # .get(str(self.details.citizenship.id)) # .get(str(div.id)) # .get("top_damage") # ) # if td and not td[0].get("sector", ""): # td = td[0] # div = battle.div[td["battle_zone_id"]] # empty_medal = _EmptyMedal( # fight_time, battle.id, div.id, defender_side, battle.zone_id # ) # if ( # td["damage"] <= DMG_MAP[div.div][defender_side] / 2 # and empty_medal not in queue # ): # queue.append(empty_medal) # queue.sort( # key=attrgetter( # "time", "defender_side", "round", "battle_id", "division_id" # ) # ) # except Exception: # pass # for div in battle.div.values(): # if div.id in _processed_divisions: # continue # else: # _processed_divisions.add(div.id) # empty_medal = _EmptyMedal(fight_time, battle.id, div.id, defender_side, battle.zone_id) # side = battle.defender if defender_side else battle.invader # if empty_medal in queue: # continue # else: # try: # medal_damage = self.get_medal_damage(div, side) # except AttributeError: # battle = self.all_battles.get(battle.id) # div: BattleDivision = battle.div[div.id] # side = battle.defender if defender_side else battle.invader # medal_damage = self.get_medal_damage(div, side) # if medal_damage <= DMG_MAP[div.div][defender_side] / 2 and empty_medal not in queue: # queue.append(empty_medal) # queue.sort( # key=attrgetter("time", "defender_side", "round", "battle_id", "division_id") # ) # else: # break # # next_check = _norm(next_check + timedelta(minutes=5)) # self.stop_threads.wait(utils.get_sleep_seconds(next_check)) # except Exception as e: # self.report_error(f"Task error: empty_tp_bh_hunter {e.args}") # empty_bh_thread.join() # return # def get_cheap_tp_divisions(self) -> Dict[str, List[Tuple[int, BattleDivision]]]: # ret = super().get_cheap_tp_divisions() # real_return = {"ground": [], "air": []} # for dmg, division in ret["ground"]: # if division.div == self.division: # real_return["ground"].append((dmg, division)) # real_return["air"] = ret["air"] # return real_return # def fight(self, battle: classes.Battle, division, side=None, count=None, use_ebs=False) -> Optional[int]: # if self.inventory.final.get("other", {}).get(3, {}).get("amount", 0) > 10: # self.activate_battle_effect(battle.id, "snowFight") # if not self.inventory.active.get("prestige_points"): # ppb = self.inventory.boosters.get("prestige_points", {}) # for _q in sorted(ppb): # _qd = ppb.get(_q) # for _dur in sorted(_qd): # pp_boost: types.InvFinalItem = _qd[_dur] # if pp_boost.get("expiration") and pp_boost.get("amount") > 10: # self.activate_pp_booster(pp_boost) # # return super().fight(battle, division, side, count, use_ebs)