Bump version: 0.17.3 → 0.18.0

improved multi bomb deploy with auto traveling,
Citizen.fight() simplified battle data gathering logic -> Citizen.shoot logic improved
Citizen.all_battles are now defaultdict with default value of empty/invalid battle, for times when trying to do things with battle which is not in all_battle dict

erepublik/__init__.py

@@ -4,7 +4,7 @@
 
 __author__ = """Eriks Karls"""
 __email__ = 'eriks@72.lv'
-__version__ = '0.17.2'
+__version__ = '0.18.0'
 
 from erepublik import classes, utils
 from erepublik.citizen import Citizen

erepublik/citizen.py

@@ -1,10 +1,10 @@
 import re
 import sys
-from threading import Event
-
-from itertools import product
+from collections import defaultdict
 from datetime import datetime, timedelta
+from itertools import product
 from json import loads, dumps
+from threading import Event
 from time import sleep
 from typing import Dict, List, Tuple, Any, Union, Set
 
@@ -12,7 +12,6 @@ from requests import Response, RequestException
 
 from erepublik.classes import (CitizenAPI, Battle, Reporter, Config, Energy, Details, Politics, MyCompanies,
                                TelegramBot, ErepublikException, BattleDivision, MyJSONEncoder)
-
 from erepublik.utils import *
 
 
@@ -38,7 +37,7 @@ class Citizen(CitizenAPI):
     ot_points = 0
 
     tg_contract = None
-    promos = None
+    promos: Dict[str, datetime] = None
 
     eday = 0
 
@@ -320,7 +319,11 @@ class Citizen(CitizenAPI):
             return
         ugly_js = re.search(r'"promotions":\s*(\[{?.*?}?])', html).group(1)
         promos = loads(normalize_html_json(ugly_js))
-        self.promos = {k: v for k, v in (self.promos.items() if self.promos else {}) if v > self.now}
+        if self.promos is None:
+            self.promos = {}
+        else:
+            self.promos.clear()
+        self.promos.update({k: v for k, v in self.promos.items() if v > self.now})
         send_mail = False
         for promo in promos:
             promo_name = promo.get("id")
@@ -549,8 +552,16 @@ class Citizen(CitizenAPI):
 
         resp_json = self._get_military_campaigns_json_list().json()
         if resp_json.get("countries"):
-            self.all_battles = {}
-            self.countries = {}
+            if self.all_battles is None:
+                self.all_battles = defaultdict(Battle)
+            else:
+                self.all_battles.clear()
+
+            if self.countries is None:
+                self.countries = {}
+            else:
+                self.countries.clear()
+
             for c_id, c_data in resp_json.get("countries").items():
                 if int(c_id) not in self.countries:
                     self.countries.update({
@@ -561,7 +572,7 @@ class Citizen(CitizenAPI):
             self.__last_war_update_data = resp_json
             if resp_json.get("battles"):
                 for battle_data in resp_json.get("battles", {}).values():
-                    self.all_battles.update({battle_data.get('id'): Battle(battle_data)})
+                    self.all_battles[battle_data.get('id')] = Battle(battle_data)
 
     def eat(self):
         """
@@ -593,8 +604,7 @@ class Citizen(CitizenAPI):
         r_json = response.json()
         next_recovery = r_json.get("food_remaining_reset").split(":")
         self.energy.set_reference_time(
-            good_timedelta(self.now,
-                                 timedelta(seconds=int(next_recovery[1]) * 60 + int(next_recovery[2])))
+            good_timedelta(self.now, timedelta(seconds=int(next_recovery[1]) * 60 + int(next_recovery[2])))
         )
         self.energy.recovered = r_json.get("health")
         self.energy.recoverable = r_json.get("food_remaining")
@@ -797,7 +807,16 @@ class Citizen(CitizenAPI):
                 self.collect_weekly_reward()
                 break
 
-    def fight(self, battle_id: int, side_id: int, count: int = None):
+    def fight(self, battle_id: int, side_id: int = None, count: int = None):
+        """Fight in a battle.
+
+        Will auto activate booster and travel if allowed to do it and
+        in the beginning will switch to default weapon (air - bare hands, ground - q7, if count > 30, else bare hands.
+        :param battle_id: int BattleId - battle to fight in
+        :param side_id: int or None. Battle side to fight in, If side_id not == invader id or not in invader deployed allies list, then defender's side is chosen
+        :param count: How many hits to do, if not specified self.should_fight() is called.
+        :return: None if no errors while fighting, otherwise error count.
+        """
         battle = self.all_battles[battle_id]
         zone_id = battle.div[11 if battle.is_air else self.division].battle_zone_id
         if not battle.is_air and self.config.boosters:
@@ -810,10 +829,10 @@ class Citizen(CitizenAPI):
 
         total_damage = 0
         total_hits = 0
-
+        side = battle.invader.id == side_id or side_id in battle.invader.deployed
         while ok_to_fight and error_count < 10 and count > 0:
             while all((count > 0, error_count < 10, self.energy.recovered >= 50)):
-                hits, error, damage = self._shoot(battle.is_air, battle_id, side_id, zone_id)
+                hits, error, damage = self._shoot(battle_id, side, zone_id)
                 count -= hits
                 total_hits += hits
                 total_damage += damage
@@ -829,8 +848,10 @@ class Citizen(CitizenAPI):
         if error_count:
             return error_count
 
-    def _shoot(self, air: bool, battle_id: int, side_id: int, zone_id: int):
-        if air:
+    def _shoot(self, battle_id: int, inv_side: bool, zone_id: int):
+        battle = self.all_battles[battle_id]
+        side_id = battle.invader.id if inv_side else battle.defender.id
+        if battle.is_air:
             response = self._post_military_fight_air(battle_id, side_id, zone_id)
         else:
             response = self._post_military_fight_ground(battle_id, side_id, zone_id)
@@ -853,6 +874,9 @@ class Citizen(CitizenAPI):
             else:
                 if j_resp.get("message") == "UNKNOWN_SIDE":
                     self._rw_choose_side(battle_id, side_id)
+                elif j_resp.get("message") == "CHANGE_LOCATION":
+                    countries = [side_id] + battle.invader.deployed if inv_side else battle.defender.deployed
+                    self.travel_to_battle(battle_id, countries)
                 err = True
         elif j_resp.get("message") == "ENEMY_KILLED":
             hits = (self.energy.recovered - j_resp["details"]["wellness"]) // 10
@@ -864,9 +888,52 @@ class Citizen(CitizenAPI):
 
         return hits, err, damage
 
-    def deploy_bomb(self, battle_id: int, bomb_id: int):
-        r = self._post_military_deploy_bomb(battle_id, bomb_id).json()
-        return not r.get('error')
+    def deploy_bomb(self, battle_id: int, bomb_id: int, inv_side: bool = None, count: int = 1):
+        """Deploy bombs in a battle for given side.
+
+        :param battle_id: int battle id
+        :param bomb_id: int bomb id
+        :param inv_side: should deploy on invader side, if None then will deploy in currently available side
+        :param count: int how many bombs to deploy
+        :return:
+        """
+        if not isinstance(count, int) or count < 1:
+            count = 1
+        has_traveled = False
+        battle = self.all_battles[battle_id]
+        if inv_side:
+            good_countries = [battle.invader.id] + battle.invader.deployed
+            if self.details.current_country not in good_countries:
+                has_traveled = self.travel_to_battle(battle_id, good_countries)
+        elif inv_side is not None:
+            good_countries = [battle.defender.id] + battle.defender.deployed
+            if self.details.current_country not in good_countries:
+                has_traveled = self.travel_to_battle(battle_id, good_countries)
+        else:
+            involved = [battle.invader.id, battle.defender.id] + battle.invader.deployed + battle.defender.deployed
+            if self.details.current_country not in involved:
+                count = 0
+        errors = deployed_count = 0
+        while (not deployed_count == count) and errors < 10:
+            r = self._post_military_deploy_bomb(battle_id, bomb_id).json()
+            if not r.get('error'):
+                deployed_count += 1
+            elif r.get('message') == 'LOCKED':
+                sleep(0.5)
+
+        if has_traveled:
+            self.travel_to_residence()
+        return deployed_count
+
+    def change_division(self, battle_id: int, division_to: int):
+        """Deploy bombs in a battle for given side.
+
+        :param battle_id: int battle id
+        :param division_to: int target division to switch to
+        :return:
+        """
+        battle = self.all_battles[battle_id]
+        self._post_main_battlefield_change_division(battle_id, battle.div[division_to].battle_zone_id)
 
     def work_ot(self):
         # I"m not checking for 1h cooldown. Beware of nightshift work, if calling more than once every 60min
@@ -1488,7 +1555,7 @@ class Citizen(CitizenAPI):
 
         if self.max_time_till_full_ff > self.time_till_week_change:
             max_count = (int(self.time_till_week_change.total_seconds()) // 360 * self.energy.interval) // 10
-            log_msg = ("End for Weekly challenge is near " 
+            log_msg = ("End for Weekly challenge is near "
                        f"(Recoverable until WC end {max_count}hp | want to do {count}hits)")
             count = count if max_count > count else max_count
 
@@ -1511,8 +1578,7 @@ class Citizen(CitizenAPI):
     @property
     def next_wc_start(self) -> datetime:
         days = 1 - self.now.weekday() if 1 - self.now.weekday() > 0 else 1 - self.now.weekday() + 7
-        return good_timedelta(self.now.replace(hour=0, minute=0, second=0, microsecond=0),
-                                    timedelta(days=days))
+        return good_timedelta(self.now.replace(hour=0, minute=0, second=0, microsecond=0), timedelta(days=days))
 
     @property
     def time_till_week_change(self) -> timedelta:
@@ -1801,7 +1867,7 @@ class Citizen(CitizenAPI):
         if reg_re.findall(html):
             ret.update(regions={}, can_attack=True)
             for reg in reg_re.findall(html):
-                ret["regions"].update({str(reg[0]): reg[1]})
+                ret["regions"].update({int(reg[0]): reg[1]})
         elif re.search(r'<a href="//www.erepublik.com/en/military/battlefield/(\d+)" '
                        r'class="join" title="Join"><span>Join</span></a>', html):
             battle_id = re.search(r'<a href="//www.erepublik.com/en/military/battlefield/(\d+)" '
@@ -1966,19 +2032,19 @@ class Citizen(CitizenAPI):
         process_error(msg, self.name, sys.exc_info(), self, self.commit_id, False)
 
     def get_battle_top_10(self, battle_id: int) -> Dict[int, List[Tuple[int, int]]]:
-        battle = self.all_battles.get(battle_id)
-        round_id = battle.get('zone_id')
-        division = self.division if round_id % 4 else 11
-
-        resp = self._post_military_battle_console(battle_id, round_id, division).json()
-        resp.pop('rounds', None)
-        ret = dict()
-        for country_id, data in resp.items():
-            ret.update({int(country_id): []})
-            for place in sorted(data.get("fighterData", {}).values(), key=lambda _: -_['raw_value']):
-                ret[int(country_id)].append((place['citizenId'], place['raw_value']))
-
-        return ret
+        return {}
+        # battle = self.all_battles.get(battle_id)
+        # round_id = battle.zone_id
+        # division = self.division if round_id % 4 else 11
+        #
+        # resp = self._post_military_battle_console(battle_id, 'battleStatistics', round_id, division).json()
+        # resp.pop('rounds', None)
+        # ret = dict()
+        # for country_id, data in resp.items():
+        #     ret.update({int(country_id): []})
+        #     for place in sorted(data.get("fighterData", {}).values(), key=lambda _: -_['raw_value']):
+        #         ret[int(country_id)].append((place['citizenId'], place['raw_value']))
+        # return ret
 
     def to_json(self, indent: bool = False) -> str:
         return dumps(self.__dict__, cls=MyJSONEncoder, indent=4 if indent else None, sort_keys=True)
@@ -2052,7 +2118,6 @@ class Citizen(CitizenAPI):
 
     def set_default_weapon(self, battle_id: int) -> int:
         battle = self.all_battles[battle_id]
-        battle_zone = battle.div[11 if battle.is_air else self.division].battle_zone_id
         available_weapons = self._get_military_show_weapons(battle_id).json()
         weapon_quality = -1
         if not battle.is_air:

erepublik/classes.py

@@ -599,6 +599,10 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
         data = dict(_token=self.token, sideCountryId=side_id, battleId=battle_id)
         return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
 
+    def _post_main_battlefield_change_division(self, battle_id: int, division_id: int) -> Response:
+        data = dict(_token=self.token, battleZoneId=division_id, battleId=battle_id)
+        return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
+
     def _post_main_buy_gold_items(self, currency: str, item: str, amount: int) -> Response:
         data = dict(itemId=item, currency=currency, amount=amount, _token=self.token)
         return self.post("{}/main/buyGoldItems".format(self.url), data=data)
@@ -796,11 +800,6 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
             data.update(page=page)
         return self.post("{}/military/battle-console".format(self.url), data=data)
 
-    def _post_military_change_weapon(self, battle_id: int, battle_zone_id: int, customization_level: int) -> Response:
-        data = dict(_token=self.token, battleZoneId=battle_zone_id, battleId=battle_id,
-                    customizationLevel=customization_level)
-        return self.post("{}/military/change-weapon".format(self.url), data=data)
-
     def _post_military_deploy_bomb(self, battle_id: int, bomb_id: int) -> Response:
         data = dict(battleId=battle_id, bombId=bomb_id, _token=self.token)
         return self.post("{}/military/deploy-bomb".format(self.url), data=data)
@@ -1102,22 +1101,42 @@ class Battle:
     def is_air(self) -> bool:
         return not bool(self.zone_id % 4)
 
-    def __init__(self, battle: Dict[str, Any]):
-        self.id = int(battle.get('id', 0))
-        self.war_id = int(battle.get('war_id', 0))
-        self.zone_id = int(battle.get('zone_id', 0))
-        self.is_rw = bool(battle.get('is_rw'))
-        self.is_as = bool(battle.get('is_as'))
-        self.is_dict_lib = bool(battle.get('is_dict')) or bool(battle.get('is_lib'))
-        self.start = datetime.datetime.fromtimestamp(int(battle.get('start', 0)), tz=utils.erep_tz)
+    def __init__(self, battle: Dict[str, Any] = None):
+        """Object representing eRepublik battle.
 
-        self.invader = BattleSide(battle.get('inv', {}).get('id'), battle.get('inv', {}).get('points'),
-                                  [row.get('id') for row in battle.get('inv', {}).get('ally_list')],
-                                  [row.get('id') for row in battle.get('inv', {}).get('ally_list') if row['deployed']])
+        :param battle: Dict object for single battle from '/military/campaignsJson/list' response's 'battles' object
+        """
+        if battle is None:
+            battle = {}
+            self.id = 0
+            self.war_id = 0
+            self.zone_id = 0
+            self.is_rw = False
+            self.is_as = False
+            self.is_dict_lib = False
+            self.start = utils.now().min
+            self.invader = BattleSide(0, 0, [], [])
+            self.defender = BattleSide(0, 0, [], [])
+        else:
+            self.id = int(battle.get('id', 0))
+            self.war_id = int(battle.get('war_id', 0))
+            self.zone_id = int(battle.get('zone_id', 0))
+            self.is_rw = bool(battle.get('is_rw'))
+            self.is_as = bool(battle.get('is_as'))
+            self.is_dict_lib = bool(battle.get('is_dict')) or bool(battle.get('is_lib'))
+            self.start = datetime.datetime.fromtimestamp(int(battle.get('start', 0)), tz=utils.erep_tz)
 
-        self.defender = BattleSide(battle.get('def', {}).get('id'), battle.get('def', {}).get('points'),
-                                   [row.get('id') for row in battle.get('def', {}).get('ally_list')],
-                                   [row.get('id') for row in battle.get('def', {}).get('ally_list') if row['deployed']])
+            self.invader = BattleSide(
+                battle.get('inv', {}).get('id'), battle.get('inv', {}).get('points'),
+                [row.get('id') for row in battle.get('inv', {}).get('ally_list')],
+                [row.get('id') for row in battle.get('inv', {}).get('ally_list') if row['deployed']]
+            )
+
+            self.defender = BattleSide(
+                battle.get('def', {}).get('id'), battle.get('def', {}).get('points'),
+                [row.get('id') for row in battle.get('def', {}).get('ally_list')],
+                [row.get('id') for row in battle.get('def', {}).get('ally_list') if row['deployed']]
+            )
 
         self.div = {}
         for div, data in battle.get('div', {}).items():

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.17.2
+current_version = 0.18.0
 commit = True
 tag = True
 

setup.py

@@ -42,6 +42,6 @@ setup(
     test_suite='tests',
     tests_require=test_requirements,
     url='https://github.com/eeriks/erepublik/',
-    version='0.17.2',
+    version='0.18.0',
     zip_safe=False,
 )