From ba840765bec27c38816672428aeaf89e93b0cc41 Mon Sep 17 00:00:00 2001 From: Eriks K Date: Wed, 8 Jul 2020 17:30:22 +0300 Subject: [PATCH] Moving away from passing object ids around and towards passing objects and even using object functions --- erepublik/__init__.py | 2 +- erepublik/access_points.py | 3 + erepublik/citizen.py | 457 +++++++++++++++++++------------------ erepublik/classes.py | 261 +++++++++++++-------- erepublik/utils.py | 41 +--- 5 files changed, 411 insertions(+), 353 deletions(-) diff --git a/erepublik/__init__.py b/erepublik/__init__.py index 1c7cc09..63ffd59 100644 --- a/erepublik/__init__.py +++ b/erepublik/__init__.py @@ -5,7 +5,7 @@ __author__ = """Eriks Karls""" __email__ = 'eriks@72.lv' __version__ = '0.20.2.2' -__commit_id__ = "cd5c0b1" +__commit_id__ = "3927b9b" from erepublik import classes, utils from erepublik.citizen import Citizen diff --git a/erepublik/access_points.py b/erepublik/access_points.py index 2152628..5769a2b 100644 --- a/erepublik/access_points.py +++ b/erepublik/access_points.py @@ -450,6 +450,9 @@ class ErepublikPoliticsAPI(CitizenBaseAPI): data = dict(_token=self.token, presentation=presentation) return self.post("{}/candidate-for-congress".format(self.url), data=data) + def _get_presidential_elections(self, country_id: int, timestamp: int) -> Response: + return self.get(f"{self.url}/main/presidential-elections/{country_id}/{timestamp}") + class ErepublikPresidentAPI(CitizenBaseAPI): def _post_wars_attack_region(self, war_id: int, region_id: int, region_name: str) -> Response: diff --git a/erepublik/citizen.py b/erepublik/citizen.py index b1559f3..05cc2ea 100644 --- a/erepublik/citizen.py +++ b/erepublik/citizen.py @@ -12,8 +12,10 @@ from requests import HTTPError, RequestException, Response from erepublik import utils from erepublik.access_points import CitizenAPI -from erepublik.classes import Battle, Config, Details, Energy, \ - ErepublikException, MyCompanies, MyJSONEncoder, OfferItem, Politics, Reporter, TelegramBot, BattleDivision +from erepublik.classes import COUNTRIES, Battle, BattleDivision, Config, Country, Details, Energy, \ + ErepublikException, MyCompanies, MyJSONEncoder, OfferItem, Politics, Reporter, TelegramBot, Holding, BattleSide, \ + Company +from erepublik.utils import erep_tz class BaseCitizen(CitizenAPI): @@ -53,8 +55,8 @@ class BaseCitizen(CitizenAPI): self.energy = Energy() self.details = Details() self.politics = Politics() - self.my_companies = MyCompanies() - self.reporter = Reporter() + self.my_companies = MyCompanies(self) + self.reporter = Reporter(self) self.stop_threads = Event() self.telegram = TelegramBot(stop_event=self.stop_threads) @@ -417,12 +419,12 @@ class BaseCitizen(CitizenAPI): return industry_name return "" - def get_countries_with_regions(self) -> Set[int]: + def get_countries_with_regions(self) -> Set[Country]: r_json = self._post_main_travel_data().json() return_set = {*[]} for country_data in r_json['countries'].values(): if country_data['currentRegions']: - return_set.add(country_data['id']) + return_set.add(COUNTRIES[country_data['id']]) return return_set def __str__(self) -> str: @@ -589,9 +591,9 @@ class BaseCitizen(CitizenAPI): self.telegram.report_medal(f"Level *{level}*") self.reporter.report_action("LEVEL_UP", value=level) - def _travel(self, country_id: int, region_id: int = 0) -> Response: + def _travel(self, country: Country, region_id: int = 0) -> Response: data = { - "toCountryId": country_id, + "toCountryId": country.id, "inRegionId": region_id, } return self._post_main_travel("moveAction", **data) @@ -728,16 +730,16 @@ class CitizenAnniversary(BaseCitizen): class CitizenTravel(BaseCitizen): - def _update_citizen_location(self, country_id: int, region_id: int): + def _update_citizen_location(self, country: Country, region_id: int): self.details.current_region = region_id - self.details.current_country = country_id + self.details.current_country = country - def get_country_travel_region(self, country_id: int) -> int: - regions = self.get_travel_regions(country_id=country_id) + def get_country_travel_region(self, country: Country) -> int: + regions = self.get_travel_regions(country=country) regs = [] if regions: for region in regions.values(): - if region['countryId'] == country_id: # Is not occupied by other country + if region['countryId'] == country.id: # Is not occupied by other country regs.append((region['id'], region['distanceInKm'])) if regs: return min(regs, key=lambda _: int(_[1]))[0] @@ -768,41 +770,44 @@ class CitizenTravel(BaseCitizen): return True return False - def travel_to_country(self, country_id: int) -> bool: - data = self._post_main_travel_data(countryId=country_id, check="getCountryRegions").json() + def travel_to_country(self, country: Country) -> bool: + data = self._post_main_travel_data(countryId=country.id, check="getCountryRegions").json() regs = [] if data.get('regions'): for region in data.get('regions').values(): - if region['countryId'] == country_id: # Is not occupied by other country + if region['countryId'] == country: # Is not occupied by other country regs.append((region['id'], region['distanceInKm'])) if regs: region_id = min(regs, key=lambda _: int(_[1]))[0] - r_json = self._travel(country_id, region_id).json() + r_json = self._travel(country, region_id).json() if r_json.get('message', '') == 'success': - self._update_citizen_location(country_id, region_id) - self._report_action("TRAVEL", f"Traveled to {utils.COUNTRIES[country_id]}", response=r_json) + self._update_citizen_location(country, region_id) + self._report_action("TRAVEL", f"Traveled to {country.name}", response=r_json) return True return False - def travel_to_holding(self, holding_id: int) -> bool: - data = self._post_main_travel_data(holdingId=holding_id).json() + def travel_to_holding(self, holding: Holding) -> bool: + data = self._post_main_travel_data(holdingId=holding.id).json() if data.get('alreadyInRegion'): return True else: r_json = self._travel(data.get('preselectCountryId'), data.get('preselectRegionId')).json() if r_json.get('message', '') == 'success': self._update_citizen_location(data.get('preselectCountryId'), data.get('preselectRegionId')) - self._report_action("TRAVEL", f"Traveled to holding {holding_id}", response=r_json) + self._report_action("TRAVEL", f"Traveled to holding {holding}", response=r_json) return True return False - def get_travel_regions(self, holding_id: int = 0, battle_id: int = 0, country_id: int = 0 + def get_travel_regions(self, holding: Holding = None, battle: Battle = None, country: Country = None ) -> Union[List[Any], Dict[str, Dict[str, Any]]]: - d = self._post_main_travel_data(holdingId=holding_id, battleId=battle_id, countryId=country_id).json() - return d.get('regions', []) + return self._post_main_travel_data( + holdingId=holding.id if holding else 0, + battleId=battle.id if battle else 0, + countryId=country.id if country else 0 + ).json().get('regions', []) - def get_travel_countries(self) -> Set[int]: + def get_travel_countries(self) -> Set[Country]: warnings.simplefilter('always') warnings.warn('CitizenTravel.get_travel_countries() are being deprecated, ' 'please use BaseCitizen.get_countries_with_regions()', DeprecationWarning) @@ -825,10 +830,10 @@ class CitizenCompanies(BaseCitizen): return ret - def work_as_manager_in_holding(self, holding_id: int) -> Optional[Dict[str, Any]]: - return self._work_as_manager(holding_id) + def work_as_manager_in_holding(self, holding: Holding) -> Optional[Dict[str, Any]]: + return self._work_as_manager(holding) - def _work_as_manager(self, wam_holding_id: int = 0) -> Optional[Dict[str, Any]]: + def _work_as_manager(self, wam_holding: Holding = None) -> Optional[Dict[str, Any]]: if self.restricted_ip: return None self.update_companies() @@ -836,33 +841,20 @@ class CitizenCompanies(BaseCitizen): data = {"action_type": "production"} extra = {} wam_list = [] - if wam_holding_id: - raw_count = self.my_companies.get_holding_wam_count(wam_holding_id, raw_factory=True) - fab_count = self.my_companies.get_holding_wam_count(wam_holding_id, raw_factory=False) - if raw_count + fab_count <= self.energy.food_fights: - raw_factories = None - elif not raw_count and fab_count <= self.energy.food_fights: - raw_factories = False - else: - raw_factories = True + if wam_holding: + raw_factories = wam_holding.get_wam_companies(raw_factory=True) + fin_factories = wam_holding.get_wam_companies(raw_factory=False) free_inventory = self.inventory["total"] - self.inventory["used"] - wam_list = self.my_companies.get_holding_wam_companies( - wam_holding_id, raw_factory=raw_factories)[:self.energy.food_fights] - has_space = False - while not has_space and wam_list: - extra_needed = self.my_companies.get_needed_inventory_usage(companies=wam_list) - has_space = extra_needed < free_inventory - if not has_space: - wam_list.pop(-1) + wam_list = [raw_factories + fin_factories][:self.energy.food_fights] + while wam_list and free_inventory < self.my_companies.get_needed_inventory_usage(companies=wam_list): + wam_list.pop(-1) if wam_list: data.update(extra) - if wam_list: - wam_holding = self.my_companies.holdings.get(wam_holding_id) - if not self.details.current_region == wam_holding.region: - self.write_log("Unable to work as manager because of location - please travel!") - return + if not self.details.current_region == wam_holding.region: + self.write_log("Unable to work as manager because of location - please travel!") + return employ_factories = self.my_companies.get_employable_factories() if sum(employ_factories.values()) > self.my_companies.work_units: @@ -883,16 +875,12 @@ class CitizenCompanies(BaseCitizen): self.my_companies.prepare_holdings(utils.json.loads(have_holdings.group(1))) self.my_companies.prepare_companies(utils.json.loads(have_companies.group(1))) - def assign_factory_to_holding(self, factory_id: int, holding_id: int) -> Response: + def assign_company_to_holding(self, company: Company, holding: Holding) -> Response: """ Assigns factory to new holding """ - company = self.my_companies.companies[factory_id] - self.write_log(f"{company} moved to {holding_id}") - return self._post_economy_assign_to_holding(factory_id, holding_id) - - def upgrade_factory(self, factory_id: int, level: int) -> Response: - return self._post_economy_upgrade_company(factory_id, level, self.details.pin) + self.write_log(f"{company} moved to {holding}") + return self._post_economy_assign_to_holding(company.id, holding.id) def create_factory(self, industry_id: int, building_type: int = 1) -> Response: """ @@ -912,11 +900,6 @@ class CitizenCompanies(BaseCitizen): self.write_log(f'{company_name} created!') return self._post_economy_create_company(industry_id, building_type) - def dissolve_factory(self, factory_id: int) -> Response: - company = self.my_companies.companies[factory_id] - self.write_log(f"{company} dissolved!") - return self._post_economy_sell_company(factory_id, self.details.pin, sell=False) - class CitizenEconomy(CitizenTravel): def update_money(self, page: int = 0, currency: int = 62) -> Dict[str, Any]: @@ -1061,7 +1044,7 @@ class CitizenEconomy(CitizenTravel): self._report_action("BOUGHT_PRODUCTS", "", kwargs=json_ret) return json_ret - def get_market_offers(self, product_name: str, quality: int = None, country_id: int = None) -> Dict[str, OfferItem]: + def get_market_offers(self, product_name: str, quality: int = None, country: Country = None) -> Dict[str, OfferItem]: raw_short_names = dict(frm="foodRaw", wrm="weaponRaw", hrm="houseRaw", arm="airplaneRaw") q1_industries = ["aircraft"] + list(raw_short_names.values()) if product_name not in self.available_industries and product_name not in raw_short_names: @@ -1081,10 +1064,10 @@ class CitizenEconomy(CitizenTravel): for q in range(max_quality): offers[f"q{q + 1}"] = OfferItem() - if country_id: - countries = [country_id] + if country: + countries: Set[Country] = {country} else: - countries = self.get_countries_with_regions() + countries: Set[Country] = self.get_countries_with_regions() start_dt = self.now iterable = [countries, [quality] if quality else range(1, max_quality + 1)] @@ -1097,7 +1080,7 @@ class CitizenEconomy(CitizenTravel): obj.price == float(offer["priceWithTaxes"]) and obj.amount < int(offer["amount"]) )): offers[f"q{q}"] = obj = OfferItem( - float(offer["priceWithTaxes"]), int(offer["country_id"]), int(offer["amount"]), + float(offer["priceWithTaxes"]), COUNTRIES[int(offer["country_id"])], int(offer["amount"]), int(offer["id"]), int(offer["citizen_id"]) ) self.write_log(f"Scraped market in {self.now - start_dt}!") @@ -1106,7 +1089,7 @@ class CitizenEconomy(CitizenTravel): def buy_food(self, energy_amount: int = 0): hp_needed = energy_amount if energy_amount else 48 * self.energy.interval * 10 - self.food["total"] - local_offers = self.get_market_offers("food", country_id=self.details.current_country) + local_offers = self.get_market_offers("food", country=self.details.current_country) cheapest_q, cheapest = sorted(local_offers.items(), key=lambda v: v[1].price / utils.FOOD_ENERGY[v[0]])[0] @@ -1207,52 +1190,52 @@ class CitizenEconomy(CitizenTravel): self.sleep(5) return self.donate_items(citizen_id, int(available), industry_id, quality) - def contribute_cc_to_country(self, amount=0., country_id: int = 71) -> bool: + def contribute_cc_to_country(self, amount, country: Country) -> bool: self.update_money() amount = int(amount) if self.details.cc < amount or amount < 20: return False - data = dict(country=country_id, action='currency', value=amount) + data = dict(country=country.id, action='currency', value=amount) r = self._post_main_country_donate(**data) if r.json().get('status') or not r.json().get('error'): - self._report_action("CONTRIBUTE_CC", f'Contributed {amount}cc to {utils.COUNTRIES[country_id]}\'s treasury') + self._report_action("CONTRIBUTE_CC", f'Contributed {amount}cc to {country}\'s treasury') return True else: - self._report_action("CONTRIBUTE_CC", f"Unable to contribute {amount}cc to {utils.COUNTRIES[country_id]}'s" + self._report_action("CONTRIBUTE_CC", f"Unable to contribute {amount}cc to {country}'s" f" treasury", kwargs=r.json()) return False - def contribute_food_to_country(self, amount: int = 0, quality: int = 1, country_id: int = 71) -> bool: + def contribute_food_to_country(self, amount, quality, country: Country) -> bool: self.update_inventory() amount = amount // 1 if self.food["q" + str(quality)] < amount or amount < 10: return False - data = dict(country=country_id, action='food', value=amount, quality=quality) + data = dict(country=country.id, action='food', value=amount, quality=quality) r = self._post_main_country_donate(**data) if r.json().get('status') or not r.json().get('error'): self._report_action("CONTRIBUTE_FOOD", f"Contributed {amount}q{quality} food to " - f"{utils.COUNTRIES[country_id]}'s treasury") + f"{country}'s treasury") return True else: self._report_action("CONTRIBUTE_FOOD", f"Unable to contribute {amount}q{quality} food to " - f"{utils.COUNTRIES[country_id]}'s treasury", kwargs=r.json()) + f"{country}'s treasury", kwargs=r.json()) return False - def contribute_gold_to_country(self, amount: int, country_id: int = 71) -> bool: + def contribute_gold_to_country(self, amount: int, country: Country) -> bool: self.update_money() if self.details.cc < amount: return False - data = dict(country=country_id, action='gold', value=amount) + data = dict(country=country.id, action='gold', value=amount) self.reporter.report_action("CONTRIBUTE_GOLD", data, str(amount)) r = self._post_main_country_donate(**data) if r.json().get('status') or not r.json().get('error'): - self._report_action("CONTRIBUTE_GOLD", f"Contributed {amount}g to {utils.COUNTRIES[country_id]}'s treasury") + self._report_action("CONTRIBUTE_GOLD", f"Contributed {amount}g to {country}'s treasury") return True else: - self._report_action("CONTRIBUTE_GOLD", f"Unable to contribute {amount}g to {utils.COUNTRIES[country_id]}'s" + self._report_action("CONTRIBUTE_GOLD", f"Unable to contribute {amount}g to {country}'s" f" treasury", kwargs=r.json()) return False @@ -1388,24 +1371,23 @@ class CitizenMilitary(CitizenTravel): def get_available_weapons(self, battle_id: int): return self._get_military_show_weapons(battle_id).json() - def set_default_weapon(self, battle_id: int, battle_zone: int) -> int: - battle = self.all_battles.get(battle_id) - available_weapons = self._get_military_show_weapons(battle_id).json() + def set_default_weapon(self, battle: Battle, division: BattleDivision) -> int: + available_weapons = self._get_military_show_weapons(battle.id).json() while not isinstance(available_weapons, list): - available_weapons = self._get_military_show_weapons(battle_id).json() + available_weapons = self._get_military_show_weapons(battle.id).json() weapon_quality = -1 weapon_damage = 0 - if not battle.div[battle_zone].is_air: + if not division.is_air: for weapon in available_weapons: try: if weapon['weaponQuantity'] > 30 and weapon['weaponInfluence'] > weapon_damage: weapon_quality = int(weapon['weaponId']) except ValueError: pass - return self.change_weapon(battle_id, weapon_quality, battle_zone) + return self.change_weapon(battle, weapon_quality, division) - def change_weapon(self, battle_id: int, quality: int, battle_zone: int) -> int: - r = self._post_military_change_weapon(battle_id, battle_zone, quality) + def change_weapon(self, battle: Battle, quality: int, battle_zone: BattleDivision) -> int: + r = self._post_military_change_weapon(battle.id, battle_zone.id, quality) influence = r.json().get('weaponInfluence') self._report_action("MILITARY_WEAPON", f"Switched to q{quality} weapon," f" new influence {influence}", kwargs=r.json()) @@ -1456,79 +1438,110 @@ class CitizenMilitary(CitizenTravel): # # self.active_fs = active_fs - def sorted_battles(self, sort_by_time: bool = True) -> List[int]: - cs_battles_priority_air: List[int] = [] - cs_battles_priority_ground: List[int] = [] - cs_battles_air: List[int] = [] - cs_battles_ground: List[int] = [] - deployed_battles_air: List[int] = [] - deployed_battles_ground: List[int] = [] - ally_battles_air: List[int] = [] - ally_battles_ground: List[int] = [] - other_battles_air: List[int] = [] - other_battles_ground: List[int] = [] + def sorted_battles(self, sort_by_time: bool = True, only_tp=False) -> List[Battle]: + cs_battles_priority_air: List[Battle] = [] + cs_battles_priority_ground: List[Battle] = [] + cs_battles_air: List[Battle] = [] + cs_battles_ground: List[Battle] = [] + deployed_battles_air: List[Battle] = [] + deployed_battles_ground: List[Battle] = [] + ally_battles_air: List[Battle] = [] + ally_battles_ground: List[Battle] = [] + other_battles_air: List[Battle] = [] + other_battles_ground: List[Battle] = [] - ret_battles: List[int] = [] + ret_battles: List[Battle] = [] if sort_by_time: battle_list = sorted(self.all_battles.values(), key=lambda b: b.start) battle_list.reverse() else: battle_list = sorted(self.all_battles.values(), key=lambda b: b.id) - contribution_json: Response = self._get_military_campaigns_json_citizen() - contributions: List[Dict[str, int]] = contribution_json.json().get('contributions') or [] + contribution_json = self._get_military_campaigns_json_citizen().json() + contributions: List[Dict[str, int]] = contribution_json.get('contributions') or [] contributions.sort(key=lambda b: -b.get('damage')) - ret_battles += [int(b.get('battle_id', 0)) for b in contributions if b.get('battle_id')] + + for contribution_battle in contributions: + if contribution_battle.get('battle_id') and contribution_battle.get('battle_id') in self.all_battles: + ret_battles.append(self.all_battles[contribution_battle.get('battle_id')]) for battle in battle_list: - battle_sides = [battle.invader.id, battle.defender.id] + battle_sides = [battle.invader.country, battle.defender.country] if battle.id in ret_battles: continue # CS Battles elif self.details.citizenship in battle_sides: if battle.has_air: if battle.defender.id == self.details.citizenship: - cs_battles_priority_air.append(battle.id) + cs_battles_priority_air.append(battle) else: - cs_battles_air.append(battle.id) + cs_battles_air.append(battle) else: if battle.defender.id == self.details.citizenship: - cs_battles_priority_ground.append(battle.id) + cs_battles_priority_ground.append(battle) else: - cs_battles_ground.append(battle.id) + cs_battles_ground.append(battle) # Current location battles: elif self.details.current_country in battle_sides: if battle.has_air: - deployed_battles_air.append(battle.id) + deployed_battles_air.append(battle) else: - deployed_battles_ground.append(battle.id) + deployed_battles_ground.append(battle) # Deployed battles and allied battles: elif self.details.current_country in battle.invader.allies + battle.defender.allies + battle_sides: if self.details.current_country in battle.invader.deployed + battle.defender.deployed: if battle.has_air: - deployed_battles_air.append(battle.id) + deployed_battles_air.append(battle) else: - deployed_battles_ground.append(battle.id) + deployed_battles_ground.append(battle) # Allied battles: else: if battle.has_air: - ally_battles_air.append(battle.id) + ally_battles_air.append(battle) else: - ally_battles_ground.append(battle.id) + ally_battles_ground.append(battle) else: if battle.has_air: - other_battles_air.append(battle.id) + other_battles_air.append(battle) else: - other_battles_ground.append(battle.id) + other_battles_ground.append(battle) cs_battles = cs_battles_priority_air + cs_battles_priority_ground + cs_battles_air + cs_battles_ground + if only_tp: + return cs_battles deployed_battles = deployed_battles_air + deployed_battles_ground other_battles = ally_battles_air + ally_battles_ground + other_battles_air + other_battles_ground ret_battles = ret_battles + cs_battles + deployed_battles + other_battles return ret_battles + def get_cheap_tp_divisions(self) -> Optional[BattleDivision]: + air_divs: List[Tuple[BattleDivision, int]] = [] + ground_divs: List[Tuple[BattleDivision, int]] = [] + for battle in self.sorted_battles(True, True): + for division in battle.div.values(): + if not division.terrain: + if division.is_air: + medal = self.get_battle_round_data(division)[self.details.citizenship == division.battle.defender.id] + if not medal and division.battle.start: + return division + else: + air_divs.append((division, medal.get('1').get('raw_value'))) + else: + medal = self.get_battle_round_data(division)[self.details.citizenship == division.battle.defender.id] + if not medal and division.battle.start: + return division + else: + ground_divs.append((division, medal.get('1').get('raw_value'))) + + if self.config.air: + return min(air_divs, key=lambda x: x[1])[0] + elif self.config.ground: + return min(ground_divs, key=lambda x: x[1])[0] + else: + return + @property def has_battle_contribution(self): return bool(self.__last_war_update_data.get("citizen_contribution", [])) @@ -1536,8 +1549,7 @@ class CitizenMilitary(CitizenTravel): def find_battle_and_fight(self): if self.should_fight()[0]: self.write_log("Checking for battles to fight in...") - for battle_id in self.sorted_battles(self.config.sort_battles_time): - battle = self.all_battles.get(battle_id) + for battle in self.sorted_battles(self.config.sort_battles_time): if not isinstance(battle, Battle): continue battle_zone: Optional[BattleDivision] = None @@ -1555,15 +1567,15 @@ class CitizenMilitary(CitizenTravel): continue if not battle_zone: continue - allies = battle.invader.deployed + battle.defender.deployed + [battle.invader.id, battle.defender.id] + allies = battle.invader.deployed + battle.defender.deployed + [battle.invader.country, battle.defender.country] travel_needed = self.details.current_country not in allies if battle.is_rw: - side_id = battle.defender.id if self.config.rw_def_side else battle.invader.id + side = battle.defender if self.config.rw_def_side else battle.invader else: - side = self.details.current_country in battle.defender.allies + [battle.defender.id, ] - side_id = battle.defender.id if side else battle.invader.id + defender_side = self.details.current_country in battle.defender.allies + [battle.defender.country, ] + side = battle.defender if defender_side else battle.invader self.write_log(battle) @@ -1578,48 +1590,42 @@ class CitizenMilitary(CitizenTravel): if travel_needed: if battle.is_rw: - country_ids_to_travel = [battle.defender.id] + countries_to_travel = [battle.defender.country] elif self.details.current_country in battle.invader.allies: - country_ids_to_travel = battle.invader.deployed + [battle.invader.id] - side_id = battle.invader.id + countries_to_travel = battle.invader.deployed + [battle.invader.country] + side = battle.invader else: - country_ids_to_travel = battle.defender.deployed + [battle.defender.id] - side_id = battle.defender.id + countries_to_travel = battle.defender.deployed + [battle.defender.country] + side = battle.defender - if not self.travel_to_battle(battle_id, country_ids_to_travel): + if not self.travel_to_battle(battle, countries_to_travel): break - self.change_division(battle_id, battle_zone.id) - self.set_default_weapon(battle_id, battle_zone.id) - self.fight(battle_id, battle_zone.id, side_id) + self.change_division(battle, battle_zone) + self.set_default_weapon(battle, battle_zone) + self.fight(battle, battle_zone, side) self.travel_to_residence() break - def fight(self, battle_id: int, battle_zone: int, side_id: int = None, count: int = None) -> int: + def fight(self, battle: Battle, division: BattleDivision, side: BattleSide = None, count: int = None) -> int: """Fight in a battle. Will auto activate booster and travel if allowed to do it. - :param battle_id: int BattleId - battle to fight in - :type battle_id: int - :param side_id: int or None. Battle side to fight in, If side_id not == invader id or not in invader deployed + :param battle: Battle battle to fight in + :type battle: Battle + :param side: BattleSide or None. Battle side to fight in, If side not == invader id or not in invader deployed allies list, then defender's side is chosen - :type side_id: int + :type side: BattleSide :param count: How many hits to do, if not specified self.should_fight() is called. :type count: int - :param battle_zone: int Division number to fight in available choices - :type battle_zone: int + :param division: Division number to fight in available choices + :type division: BattleDivision :return: None if no errors while fighting, otherwise error count. :rtype: int """ if self.restricted_ip: self._report_action("IP_BLACKLISTED", "Fighting is not allowed from restricted IP!") return 1 - if not isinstance(battle_id, int): - self.report_error(f"WARNING! Parameter battle_id should be 'int', but it is '{type(battle_id).__name__}'") - battle_id = int(battle_id) - if battle_id not in self.all_battles: - self.update_war_info() - battle = self.all_battles.get(battle_id) - if not battle.div[battle_zone].is_air and self.config.boosters: + if not division.is_air and self.config.boosters: self.activate_dmg_booster() error_count = 0 ok_to_fight = True @@ -1628,10 +1634,9 @@ class CitizenMilitary(CitizenTravel): 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_id, side, battle_zone) + hits, error, damage = self._shoot(battle, division, side) count -= hits total_hits += hits total_damage += damage @@ -1642,17 +1647,15 @@ class CitizenMilitary(CitizenTravel): self.write_log("Hits: {:>4} | Damage: {}".format(total_hits, total_damage)) ok_to_fight = False if total_damage: - self.reporter.report_action("FIGHT", dict(battle=battle_id, side=side_id, dmg=total_damage, + self.reporter.report_action("FIGHT", dict(battle=battle, side=side, dmg=total_damage, air=battle.has_air, hits=total_hits)) return error_count - def _shoot(self, battle_id: int, inv_side: bool, zone_id: int): - battle = self.all_battles.get(battle_id) - side_id = battle.invader.id if inv_side else battle.defender.id - if battle.div[zone_id].is_air: - response = self._post_military_fight_air(battle_id, side_id, zone_id) + def _shoot(self, battle: Battle, division: BattleDivision, side: BattleSide): + if division.is_air: + response = self._post_military_fight_air(battle.id, side.id, division.id) else: - response = self._post_military_fight_ground(battle_id, side_id, zone_id) + response = self._post_military_fight_ground(battle.id, side.id, division.id) if "Zone is not meant for " in response.text: self.sleep(5) @@ -1668,16 +1671,16 @@ class CitizenMilitary(CitizenTravel): if r_json.get("message") == "SHOOT_LOCKOUT" or r_json.get("message") == "ZONE_INACTIVE": pass elif r_json.get("message") == "NOT_ENOUGH_WEAPONS": - self.set_default_weapon(battle_id, zone_id) + self.set_default_weapon(battle, division) elif r_json.get("message") == "FIGHT_DISABLED": self._post_main_profile_update('options', params='{"optionName":"enable_web_deploy","optionValue":"off"}') - self.set_default_weapon(battle_id, zone_id) + self.set_default_weapon(battle, division) else: if r_json.get("message") == "UNKNOWN_SIDE": - self._rw_choose_side(battle_id, side_id) + self._rw_choose_side(battle, side) elif r_json.get("message") == "CHANGE_LOCATION": - countries = [side_id] + battle.invader.deployed if inv_side else battle.defender.deployed - self.travel_to_battle(battle_id, countries) + countries = [side.country] + side.deployed + self.travel_to_battle(battle, countries) err = True elif r_json.get("message") == "ENEMY_KILLED": hits = (self.energy.recovered - r_json["details"]["wellness"]) // 10 @@ -1689,10 +1692,11 @@ class CitizenMilitary(CitizenTravel): return hits, err, damage - def deploy_bomb(self, battle_id: int, bomb_id: int, inv_side: bool = None, count: int = 1) -> int: + def deploy_bomb(self, battle: Battle, bomb_id: int, inv_side: bool = None, count: int = 1) -> int: """Deploy bombs in a battle for given side. - :param battle_id: int battle id + :param battle: Battle + :type battle: Battle :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 @@ -1702,25 +1706,24 @@ class CitizenMilitary(CitizenTravel): if not isinstance(count, int) or count < 1: count = 1 has_traveled = False - battle = self.all_battles.get(battle_id) if battle.is_rw: - has_traveled = self.travel_to_battle(battle_id, [battle.defender.id]) - self._rw_choose_side(battle.id, battle.invader.id if inv_side else battle.defender.id) + has_traveled = self.travel_to_battle(battle, [battle.defender.country]) + self._rw_choose_side(battle, battle.invader if inv_side else battle.defender) if inv_side: - good_countries = [battle.invader.id] + battle.invader.deployed + good_countries = [battle.invader.country] + battle.invader.deployed if self.details.current_country not in good_countries: - has_traveled = self.travel_to_battle(battle_id, good_countries) + has_traveled = self.travel_to_battle(battle, good_countries) elif inv_side is not None: - good_countries = [battle.defender.id] + battle.defender.deployed + good_countries = [battle.defender.country] + battle.defender.deployed if self.details.current_country not in good_countries: - has_traveled = self.travel_to_battle(battle_id, good_countries) + has_traveled = self.travel_to_battle(battle, good_countries) else: - involved = [battle.invader.id, battle.defender.id] + battle.invader.deployed + battle.defender.deployed + involved = [battle.invader.country, battle.defender.country] + 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() + r = self._post_military_deploy_bomb(battle.id, bomb_id).json() if not r.get('error'): deployed_count += 1 elif r.get('message') == 'LOCKED': @@ -1734,17 +1737,17 @@ class CitizenMilitary(CitizenTravel): self._report_action("MILITARY_BOMB", f"Deployed {deployed_count} bombs in battle {battle.id}") return deployed_count - def change_division(self, battle_id: int, division_to: int): + def change_division(self, battle: Battle, division: BattleDivision): """Change division. - :param battle_id: int battle id - :param division_to: int target division to switch to + :param battle: Battle + :type battle: Battle + :param division: int target division to switch to + :type division: BattleDivision :return: """ - battle = self.all_battles.get(battle_id) - division = battle.div[division_to] - self._post_main_battlefield_change_division(battle_id, division_to) - self._report_action("MILITARY_DIV_SWITCH", f"Switched to d{division.div} in battle {battle_id}") + self._post_main_battlefield_change_division(battle.id, division.id) + self._report_action("MILITARY_DIV_SWITCH", f"Switched to d{division.div} in battle {battle.id}") def get_ground_hit_dmg_value(self, rang: int = None, strength: float = None, elite: bool = None, ne: bool = False, booster_50: bool = False, booster_100: bool = False, tp: bool = True) -> Decimal: @@ -1801,8 +1804,8 @@ class CitizenMilitary(CitizenTravel): self._report_action('MILITARY_BOOSTER', 'Activated PrestigePoint booster') return self._post_military_fight_activate_booster(battle_id, 1, 180, "prestige_points") - def _rw_choose_side(self, battle_id: int, side_id: int) -> Response: - return self._post_main_battlefield_travel(side_id, battle_id) + def _rw_choose_side(self, battle: Battle, side: BattleSide) -> Response: + return self._post_main_battlefield_travel(side.id, battle.id) def should_travel_to_fight(self) -> bool: ret = False @@ -1875,16 +1878,15 @@ class CitizenMilitary(CitizenTravel): return (count if count > 0 else 0), msg, force_fight - def get_battle_round_data(self, battle_id: int, round_id: int, division: int = None) -> dict: - battle = self.all_battles.get(battle_id) - if not battle: - return {} + def get_battle_round_data(self, division: BattleDivision) -> Tuple[Any, Any]: + battle = division.battle - data = dict(zoneId=round_id, round=round_id, division=division, leftPage=1, rightPage=1, type="damage") - - r = self._post_military_battle_console(battle_id, "battleStatistics", 1, **data) - return {battle.invader.id: r.json().get(str(battle.invader.id)).get("fighterData"), - battle.defender.id: r.json().get(str(battle.defender.id)).get("fighterData")} + data = dict(zoneId=battle.zone_id, round_id=battle.zone_id, division=division.div, + battleZoneId=division.id, type="damage") + r = self._post_military_battle_console(battle.id, "battleStatistics", 1, **data) + r_json = r.json() + return (r_json.get(str(battle.invader.id)).get("fighterData"), + r_json.get(str(battle.defender.id)).get("fighterData")) def schedule_attack(self, war_id: int, region_id: int, region_name: str, at_time: datetime): if at_time: @@ -1892,8 +1894,8 @@ class CitizenMilitary(CitizenTravel): self.get_csrf_token() self.launch_attack(war_id, region_id, region_name) - def get_active_wars(self, country_id: int = None) -> List[int]: - r = self._get_country_military(utils.COUNTRY_LINK.get(country_id or self.details.citizenship)) + def get_active_wars(self, country: Country = None) -> List[int]: + r = self._get_country_military(country.link if country else self.details.citizenship.link) all_war_ids = re.findall(r'//www\.erepublik\.com/en/wars/show/(\d+)"', r.text) return [int(wid) for wid in all_war_ids] @@ -1909,31 +1911,32 @@ class CitizenMilitary(CitizenTravel): self._post_wars_attack_region(war_id, region_id, region_name) self._report_action("MILITARY_QUEUE_ATTACK", f"Battle for *{region_name}* queued") - def travel_to_battle(self, battle_id: int, allowed_countries: List[int]) -> bool: - data = self.get_travel_regions(battle_id=battle_id) + def travel_to_battle(self, battle: Battle, allowed_countries: List[Country]) -> bool: + data = self.get_travel_regions(battle=battle) regs = [] + countries: Dict[int, Country] = {c.id: c for c in allowed_countries} if data: for region in data.values(): - if region['countryId'] in allowed_countries: # Is not occupied by other country - regs.append((region['distanceInKm'], region['id'], region['countryId'])) + if region['countryId'] in countries: # Is not occupied by other country + regs.append((region['distanceInKm'], region['id'], countries[region['countryId']])) if regs: reg = min(regs, key=lambda _: int(_[0])) region_id = reg[1] - country_id = reg[2] - r = self._travel(country_id, region_id).json() + country = reg[2] + r = self._travel(country, region_id).json() if r.get('message', '') == 'success': - self._update_citizen_location(country_id, region_id) + self._update_citizen_location(country, region_id) return True return False - def get_country_mus(self, country_id: int) -> Dict[int, str]: + def get_country_mus(self, country: Country) -> Dict[int, str]: ret = {} - r = self._get_main_leaderboards_damage_rankings(country_id) + r = self._get_main_leaderboards_damage_rankings(country.id) for data in r.json()["mu_filter"]: if data["id"]: ret.update({data["id"]: data["name"]}) - r = self._get_main_leaderboards_damage_aircraft_rankings(country_id) + r = self._get_main_leaderboards_damage_aircraft_rankings(country.id) for data in r.json()["mu_filter"]: if data["id"]: ret.update({data["id"]: data["name"]}) @@ -1965,10 +1968,8 @@ class CitizenMilitary(CitizenTravel): class CitizenPolitics(BaseCitizen): - def get_country_parties(self, country_id: int = None) -> dict: - if country_id is None: - country_id = self.details.citizenship - r = self._get_main_rankings_parties(country_id) + def get_country_parties(self, country: Country = None) -> dict: + r = self._get_main_rankings_parties(country.id if country else self.details.citizenship.id) ret = {} for name, id_ in re.findall(r'', r.text): ret.update({int(id_): name}) @@ -1982,6 +1983,18 @@ class CitizenPolitics(BaseCitizen): self._report_action('POLITIC_PARTY_PRESIDENT', 'Applied for party president elections') return self._get_candidate_party(self.politics.party_slug) + def get_country_president_election_result(self, country: Country, year: int, month: int) -> Dict[str, int]: + timestamp = int(erep_tz.localize(datetime(year, month, 5)).timestamp()) + resp = self._get_presidential_elections(country.id, timestamp) + candidates = re.findall(r'class="candidate_info">(.*?)', resp.text, re.S | re.M) + ret = {} + for candidate in candidates: + name = re.search(r'', candidate) + name = name.group(1) + votes = re.search(r'(\d+) votes', candidate).group(1) + ret.update({name: int(votes)}) + return ret + class CitizenSocial(BaseCitizen): def send_mail_to_owner(self): @@ -2197,7 +2210,7 @@ class CitizenTasks(BaseCitizen): return ret def find_new_job(self) -> Response: - r = self._get_economy_job_market_json(self.details.current_country) + r = self._get_economy_job_market_json(self.details.current_country.id) jobs = r.json().get("jobs") data = dict(citizen=0, salary=10) for posting in jobs: @@ -2249,7 +2262,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade self.get_csrf_token() self.update_citizen_info() - self.reporter.do_init(self.name, self.config.email, self.details.citizen_id) + self.reporter.do_init() if self.config.telegram: # noinspection SpellCheckingInspection self.telegram.do_init(self.config.telegram_chat_id or 620981703, @@ -2462,8 +2475,8 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade self.post_market_offer(industry=self.available_industries[kind], amount=int(amount), quality=int(quality), price=price) - def _wam(self, holding_id: int) -> NoReturn: - response = self.work_as_manager_in_holding(holding_id) + def _wam(self, holding: Holding) -> NoReturn: + response = self.work_as_manager_in_holding(holding) if response is None: return if response.get("status"): @@ -2500,11 +2513,11 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade break else: if not start_place == (self.details.current_country, self.details.current_region): - self.travel_to_holding(holding_id) - return self._wam(holding_id) + self.travel_to_holding(holding) + return self._wam(holding) elif response.get("message") == "not_enough_health_food": self.buy_food() - self._wam(holding_id) + self._wam(holding) else: msg = "I was not able to wam and or employ because:\n{}".format(response) self._report_action("WORK_AS_MANAGER", f"Worked as manager failed: {msg}", kwargs=response) @@ -2523,19 +2536,19 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade self.update_companies() # Prevent messing up levelup with wam if not (self.is_levelup_close and self.config.fight) or self.config.force_wam: - regions = {} - for holding_id, holding in self.my_companies.holdings.items(): - if self.my_companies.get_holding_wam_companies(holding_id): - regions.update({holding.region: holding_id}) + regions: Dict[int, Holding] = {} + for holding in self.my_companies.holdings.values(): + if holding.wam_count: + regions.update({holding.region: holding}) # Check for current region if self.details.current_region in regions: self._wam(regions.pop(self.details.current_region)) self.update_companies() - for holding_id in regions.values(): - self.travel_to_holding(holding_id) - self._wam(holding_id) + for holding in regions.values(): + self.travel_to_holding(holding) + self._wam(holding) self.update_companies() wam_count = self.my_companies.get_total_wam_count() @@ -2551,9 +2564,9 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade self.update_companies() return bool(self.my_companies.get_total_wam_count()) - def sorted_battles(self, sort_by_time: bool = True) -> List[int]: - battles = self.reporter.fetch_battle_priorities(self.details.current_country) - return battles + super().sorted_battles(sort_by_time) + def sorted_battles(self, sort_by_time: bool = True, only_tp=False) -> List[Battle]: + battles: List[Battle] = self.reporter.fetch_battle_priorities(self.details.current_country) + return battles + super().sorted_battles(sort_by_time, only_tp) def command_central(self): while not self.stop_threads.is_set(): diff --git a/erepublik/classes.py b/erepublik/classes.py index c22cc12..c9e9881 100644 --- a/erepublik/classes.py +++ b/erepublik/classes.py @@ -16,6 +16,40 @@ INDUSTRIES = {1: "Food", 2: "Weapons", 4: "House", 23: "Aircraft", 24: "ARM q1", 25: "ARM q2", 26: "ARM q3", 27: "ARM q4", 28: "ARM q5", } +class Country: + id: int + name: str + link: str + iso: str + + def __init__(self, country_id: int, name: str, link: str, iso: str): + self.id = country_id + self.name = name + self.link = link + self.iso = iso + + def __repr__(self): + return f"Country({self.id}, '{self.name}', '{self.link}', '{self.iso}')" + + def __str__(self): + return f"#{self.id} {self.name}" + + def __format__(self, format_spec): + return self.iso + + def __int__(self): + return self.id + + def __eq__(self, other): + if isinstance(other, (int, float)): + return self.id == int(other) + else: + try: + return self.id == int(other) + except ValueError: + return self == other + + class ErepublikException(Exception): def __init__(self, message): super().__init__(message) @@ -32,7 +66,8 @@ class Holding: region: int companies: List["Company"] - def __init__(self, _id: int, region: int): + def __init__(self, _id: int, region: int, citizen): + self.citizen = citizen self.id: int = _id self.region: int = region self.companies: List["Company"] = list() @@ -63,6 +98,24 @@ class Holding: wrm += company.raw_usage return dict(frm=frm, wrm=wrm) + def get_wam_companies(self, raw_factory: bool = None): + raw = [] + factory = [] + for company in self.wam_companies: + if not company.already_worked and not company.cannot_wam_reason == "war": + if company.is_raw: + raw.append(company) + else: + factory.append(company) + if raw_factory is not None and not raw_factory: + return factory + elif raw_factory is not None and raw_factory: + return raw + elif raw_factory is None: + return raw + factory + else: + raise ErepublikException("raw_factory should be True/False/None") + def __str__(self): name = f"Holding (#{self.id}) with {len(self.companies)} " if len(self.companies) % 10 == 1: @@ -187,6 +240,15 @@ class Company: can_wam=self.can_wam, cannot_wam_reason=self.cannot_wam_reason, industry=self.industry, already_worked=self.already_worked, preset_works=self.preset_works) + def dissolve(self) -> Response: + self.holding.citizen.write_log(f"{self} dissolved!") + # noinspection PyProtectedMember + return self.holding.citizen._post_economy_sell_company(self.id, self.holding.citizen.details.pin, sell=False) + + def upgrade(self, level: int) -> Response: + # noinspection PyProtectedMember + return self.holding.citizen._post_economy_upgrade_company(self.id, level, self.holding.citizen.details.pin) + class MyCompanies: work_units: int = 0 @@ -196,7 +258,9 @@ class MyCompanies: holdings: Dict[int, Holding] companies: List[Company] - def __init__(self): + def __init__(self, citizen): + from erepublik import Citizen + self.citizen: Citizen = citizen self.holdings: Dict[int, Holding] = dict() self.companies: List[Company] = list() self.next_ot_time = utils.now() @@ -207,9 +271,9 @@ class MyCompanies: """ for holding in holdings.values(): if holding.get('id') not in self.holdings: - self.holdings.update({int(holding.get('id')): Holding(holding['id'], holding['region_id'])}) + self.holdings.update({int(holding.get('id')): Holding(holding['id'], holding['region_id'], self.citizen)}) if not self.holdings.get(0): - self.holdings.update({0: Holding(0, 0)}) # unassigned + self.holdings.update({0: Holding(0, 0, self.citizen)}) # unassigned def prepare_companies(self, companies: Dict[str, Dict[str, Any]]): """ @@ -240,43 +304,8 @@ class MyCompanies: def get_total_wam_count(self) -> int: return sum([holding.wam_count for holding in self.holdings.values()]) - def get_holding_wam_count(self, holding_id: int, raw_factory=None) -> int: - """ - Returns amount of wam enabled companies in the holding - :param holding_id: holding id - :param raw_factory: True - only raw, False - only factories, None - both - :return: int - """ - return len(self.get_holding_wam_companies(holding_id, raw_factory)) - - def get_holding_wam_companies(self, holding_id: int, raw_factory: bool = None) -> List[Company]: - """ - Returns WAM enabled companies in the holding, True - only raw, False - only factories, None - both - :param holding_id: holding id - :param raw_factory: bool or None - :return: list - """ - raw = [] - factory = [] - if holding_id in self.holdings: - for company in self.holdings[holding_id].wam_companies: - if not company.already_worked and not company.cannot_wam_reason == "war": - if company.is_raw: - raw.append(company) - else: - factory.append(company) - if raw_factory is not None and not raw_factory: - return factory - elif raw_factory is not None and raw_factory: - return raw - elif raw_factory is None: - return raw + factory - else: - raise ErepublikException("raw_factory should be True/False/None") - @staticmethod def get_needed_inventory_usage(companies: Union[Company, List[Company]]) -> Decimal: - if isinstance(companies, list): return sum([company.products_made * 100 if company.is_raw else 1 for company in companies]) else: @@ -428,11 +457,11 @@ class Details: gold = 0 next_pp: List[int] = None citizen_id = 0 - citizenship = 0 + citizenship: Country current_region = 0 - current_country = 0 + current_country: Country residence_region = 0 - residence_country = 0 + residence_country: Country daily_task_done = False daily_task_reward = False mayhem_skills = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0, 13: 0, 14: 0, } @@ -493,28 +522,35 @@ class House: class Reporter: __to_update: List[Dict[Any, Any]] = None - name: str = "" - email: str = "" - citizen_id: int = 0 key: str = "" allowed: bool = False + @property + def name(self) -> str: + return self.citizen.name + + @property + def email(self) -> str: + return self.citizen.config.email + + @property + def citizen_id(self) -> int: + return self.citizen.details.citizen_id + @property def __dict__(self): return dict(name=self.name, email=self.email, citizen_id=self.citizen_id, key=self.key, allowed=self.allowed, queue=self.__to_update) - def __init__(self): + def __init__(self, citizen): + self.citizen = citizen self._req = Session() self.url = "https://api.erep.lv" self._req.headers.update({"user-agent": "Bot reporter v2"}) self.__to_update = [] self.__registered: bool = False - def do_init(self, name: str = "", email: str = "", citizen_id: int = 0): - self.name: str = name - self.email: str = email - self.citizen_id: int = citizen_id + def do_init(self): self.key: str = "" self.__update_key() self.register_account() @@ -573,10 +609,10 @@ class Reporter: def report_promo(self, kind: str, time_until: datetime.datetime): self._req.post(f"{self.url}/promos/add/", data=dict(kind=kind, time_untill=time_until)) - def fetch_battle_priorities(self, country_id: int) -> List[int]: + def fetch_battle_priorities(self, country: Country) -> List["Battle"]: try: - battle_response = self._req.get(f'{self.url}/api/v1/battles/{country_id}') - return battle_response.json().get('battle_ids', []) + battle_response = self._req.get(f'{self.url}/api/v1/battles/{country.id}') + return [self.citizen.all_battles[bid] for bid in battle_response.json().get('battle_ids', []) if bid in self.citizen.all_battles] except: # noqa return [] @@ -617,26 +653,39 @@ class MyJSONEncoder(json.JSONEncoder): class BattleSide: - id: int points: int - deployed: List[int] = None - allies: List[int] = None + deployed: List[Country] + allies: List[Country] battle: "Battle" + country: Country + defender: bool - def __init__(self, battle: "Battle", country_id: int, points: int, allies: List[int], deployed: List[int]): + def __init__(self, battle: "Battle", country: Country, points: int, allies: List[Country], deployed: List[Country], defender: bool): self.battle = battle - self.id = country_id + self.country = country self.points = points - self.allies = [int(ally) for ally in allies] - self.deployed = [int(ally) for ally in deployed] + self.allies = allies + self.deployed = deployed + self.defender = defender + + @property + def id(self) -> int: + return self.country.id + + def __str__(self): + side_text = "Defender" if self.defender else "Invader " + return f"" + + def __format__(self, format_spec): + return self.country.iso class BattleDivision: + id: int end: datetime.datetime epic: bool dom_pts: Dict[str, int] wall: Dict[str, Union[int, float]] - battle_zone_id: int def_medal: Dict[str, int] inv_medal: Dict[str, int] terrain: int @@ -651,8 +700,8 @@ class BattleDivision: def div_end(self) -> bool: return utils.now() >= self.end - def __init__(self, battle: "Battle", div_id: int, end: datetime.datetime, epic: bool, div: int, wall_for: int, wall_dom: float, - terrain_id: int = 0): + def __init__(self, battle: "Battle", div_id: int, end: datetime.datetime, epic: bool, div: int, wall_for: int, + wall_dom: float, terrain_id: int = 0): """Battle division helper class :type div_id: int @@ -664,7 +713,7 @@ class BattleDivision: :type wall_dom: float """ self.battle = battle - self.battle_zone_id = div_id + self.id = div_id self.end = end self.epic = epic self.wall = dict({"for": wall_for, "dom": wall_dom}) @@ -672,8 +721,11 @@ class BattleDivision: self.div = div @property - def id(self): - return self.battle_zone_id + def terrain_display(self): + return _TERRAINS[self.terrain] + + def __str__(self): + return f"#{self.battle.id} {self.battle.invader} ({self.terrain_display})" class Battle: @@ -716,17 +768,15 @@ class Battle: self.start = datetime.datetime.fromtimestamp(int(battle.get('start', 0)), tz=utils.erep_tz) self.invader = BattleSide( - self, - battle.get('inv', {}).get('id'), battle.get('inv', {}).get('points'), + self, 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']] + [row.get('id') for row in battle.get('inv', {}).get('ally_list') if row['deployed']], False ) self.defender = BattleSide( - self, - battle.get('def', {}).get('id'), battle.get('def', {}).get('points'), + self, 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']] + [row.get('id') for row in battle.get('def', {}).get('ally_list') if row['deployed']], True ) self.div = {} @@ -754,7 +804,7 @@ class Battle: time_part = "-{}".format(self.start - now) return f"Battle {self.id} | " \ - f"{utils.ISO_CC[self.invader.id]} : {utils.ISO_CC[self.defender.id]} | " \ + f"{self.invader} : {self.defender} | " \ f"Round {self.zone_id:2} | " \ f"Round time {time_part}" @@ -841,22 +891,6 @@ class TelegramBot: return True - def report_free_bhs(self, battles: List[Tuple[int, int, int, int, datetime.timedelta]]): - battle_links = [] - for battle_id, side_id, against_id, damage, time_left in battles: - total_seconds = int(time_left.total_seconds()) - time_start = "" - hours, remainder = divmod(total_seconds, 3600) - if hours: - time_start = f"{hours}h " - minutes, seconds = divmod(remainder, 60) - time_start += f"{minutes:02}m {seconds:02}s" - damage = "{:,}".format(damage).replace(',', ' ') - battle_links.append(f"*{damage}*dmg bh for [{utils.COUNTRIES[side_id]} vs {utils.COUNTRIES[against_id]}]" - f"(https://www.erepublik.com/en/military/battlefield/{battle_id}) " - f"_time since start {time_start}_") - self.send_message("Free BHs:\n" + "\n".join(battle_links)) - def report_full_energy(self, available: int, limit: int, interval: int): if (utils.now() - self._last_full_energy_report).total_seconds() >= 30 * 60: self._last_full_energy_report = utils.now() @@ -886,7 +920,54 @@ class TelegramBot: class OfferItem(NamedTuple): price: float = 99_999. - country: int = 0 + country: Country = Country(0, "", "", "") amount: int = 0 offer_id: int = 0 citizen_id: int = 0 + + +COUNTRIES: Dict[int, Country] = { + 1: Country(1, 'Romania', 'Romania', 'ROU'), 9: Country(9, 'Brazil', 'Brazil', 'BRA'), + 10: Country(10, 'Italy', 'Italy', 'ITA'), 11: Country(11, 'France', 'France', 'FRA'), + 12: Country(12, 'Germany', 'Germany', 'DEU'), 13: Country(13, 'Hungary', 'Hungary', 'HUN'), + 14: Country(14, 'China', 'China', 'CHN'), 15: Country(15, 'Spain', 'Spain', 'ESP'), + 23: Country(23, 'Canada', 'Canada', 'CAN'), 24: Country(24, 'USA', 'USA', 'USA'), + 26: Country(26, 'Mexico', 'Mexico', 'MEX'), 27: Country(27, 'Argentina', 'Argentina', 'ARG'), + 28: Country(28, 'Venezuela', 'Venezuela', 'VEN'), 29: Country(29, 'United Kingdom', 'United-Kingdom', 'GBR'), + 30: Country(30, 'Switzerland', 'Switzerland', 'CHE'), 31: Country(31, 'Netherlands', 'Netherlands', 'NLD'), + 32: Country(32, 'Belgium', 'Belgium', 'BEL'), 33: Country(33, 'Austria', 'Austria', 'AUT'), + 34: Country(34, 'Czech Republic', 'Czech-Republic', 'CZE'), 35: Country(35, 'Poland', 'Poland', 'POL'), + 36: Country(36, 'Slovakia', 'Slovakia', 'SVK'), 37: Country(37, 'Norway', 'Norway', 'NOR'), + 38: Country(38, 'Sweden', 'Sweden', 'SWE'), 39: Country(39, 'Finland', 'Finland', 'FIN'), + 40: Country(40, 'Ukraine', 'Ukraine', 'UKR'), 41: Country(41, 'Russia', 'Russia', 'RUS'), + 42: Country(42, 'Bulgaria', 'Bulgaria', 'BGR'), 43: Country(43, 'Turkey', 'Turkey', 'TUR'), + 44: Country(44, 'Greece', 'Greece', 'GRC'), 45: Country(45, 'Japan', 'Japan', 'JPN'), + 47: Country(47, 'South Korea', 'South-Korea', 'KOR'), 48: Country(48, 'India', 'India', 'IND'), + 49: Country(49, 'Indonesia', 'Indonesia', 'IDN'), 50: Country(50, 'Australia', 'Australia', 'AUS'), + 51: Country(51, 'South Africa', 'South Africa', 'ZAF'), + 52: Country(52, 'Republic of Moldova', 'Republic-of-Moldova', 'MDA'), + 53: Country(53, 'Portugal', 'Portugal', 'PRT'), 54: Country(54, 'Ireland', 'Ireland', 'IRL'), + 55: Country(55, 'Denmark', 'Denmark', 'DNK'), 56: Country(56, 'Iran', 'Iran', 'IRN'), + 57: Country(57, 'Pakistan', 'Pakistan', 'PAK'), 58: Country(58, 'Israel', 'Israel', 'ISR'), + 59: Country(59, 'Thailand', 'Thailand', 'THA'), 61: Country(61, 'Slovenia', 'Slovenia', 'SVN'), + 63: Country(63, 'Croatia', 'Croatia', 'HRV'), 64: Country(64, 'Chile', 'Chile', 'CHL'), + 65: Country(65, 'Serbia', 'Serbia', 'SRB'), 66: Country(66, 'Malaysia', 'Malaysia', 'MYS'), + 67: Country(67, 'Philippines', 'Philippines', 'PHL'), 68: Country(68, 'Singapore', 'Singapore', 'SGP'), + 69: Country(69, 'Bosnia and Herzegovina', 'Bosnia-Herzegovina', 'BiH'), + 70: Country(70, 'Estonia', 'Estonia', 'EST'), 80: Country(80, 'Montenegro', 'Montenegro', 'MNE'), + 71: Country(71, 'Latvia', 'Latvia', 'LVA'), 72: Country(72, 'Lithuania', 'Lithuania', 'LTU'), + 73: Country(73, 'North Korea', 'North-Korea', 'PRK'), 74: Country(74, 'Uruguay', 'Uruguay', 'URY'), + 75: Country(75, 'Paraguay', 'Paraguay', 'PRY'), 76: Country(76, 'Bolivia', 'Bolivia', 'BOL'), + 77: Country(77, 'Peru', 'Peru', 'PER'), 78: Country(78, 'Colombia', 'Colombia', 'COL'), + 79: Country(79, 'Republic of Macedonia (FYROM)', 'Republic-of-Macedonia-FYROM', 'MKD'), + 81: Country(81, 'Republic of China (Taiwan)', 'Republic-of-China-Taiwan', 'TWN'), + 82: Country(82, 'Cyprus', 'Cyprus', 'CYP'), 167: Country(167, 'Albania', 'Albania', 'ALB'), + 83: Country(83, 'Belarus', 'Belarus', 'BLR'), 84: Country(84, 'New Zealand', 'New-Zealand', 'NZL'), + 164: Country(164, 'Saudi Arabia', 'Saudi-Arabia', 'SAU'), 165: Country(165, 'Egypt', 'Egypt', 'EGY'), + 166: Country(166, 'United Arab Emirates', 'United-Arab-Emirates', 'UAE'), + 168: Country(168, 'Georgia', 'Georgia', 'GEO'), 169: Country(169, 'Armenia', 'Armenia', 'ARM'), + 170: Country(170, 'Nigeria', 'Nigeria', 'NGA'), 171: Country(171, 'Cuba', 'Cuba', 'CUB') +} + +_TERRAINS = {0: "Standard", 1: 'Industrial', 2: 'Urban', 3: 'Suburbs', 4: 'Airport', 5: 'Plains', 6: 'Wasteland', + 7: 'Mountains', 8: 'Beach', 9: 'Swamp', 10: 'Mud', 11: 'Hills', 12: 'Jungle', 13: 'Forest', 14: 'Desert'} diff --git a/erepublik/utils.py b/erepublik/utils.py index 8ff3d2f..c4e8a9f 100644 --- a/erepublik/utils.py +++ b/erepublik/utils.py @@ -21,7 +21,7 @@ try: except ImportError: import json -__all__ = ["FOOD_ENERGY", "COMMIT_ID", "COUNTRIES", "erep_tz", 'COUNTRY_LINK', +__all__ = ["FOOD_ENERGY", "COMMIT_ID", "erep_tz", "now", "localize_dt", "localize_timestamp", "good_timedelta", "eday_from_date", "date_from_eday", "get_sleep_seconds", "interactive_sleep", "silent_sleep", "write_silent_log", "write_interactive_log", "get_file", "write_file", @@ -95,45 +95,6 @@ GROUND_RANK_POINTS: Dict[int, int] = { 86: 180000000000, 87: 190000000000, 88: 200000000000, 89: 210000000000 } -COUNTRIES: Dict[int, str] = { - 1: 'Romania', 9: 'Brazil', 10: 'Italy', 11: 'France', 12: 'Germany', 13: 'Hungary', 14: 'China', 15: 'Spain', - 23: 'Canada', 24: 'USA', 26: 'Mexico', 27: 'Argentina', 28: 'Venezuela', 29: 'United Kingdom', 30: 'Switzerland', - 31: 'Netherlands', 32: 'Belgium', 33: 'Austria', 34: 'Czech Republic', 35: 'Poland', 36: 'Slovakia', 37: 'Norway', - 38: 'Sweden', 39: 'Finland', 40: 'Ukraine', 41: 'Russia', 42: 'Bulgaria', 43: 'Turkey', 44: 'Greece', 45: 'Japan', - 47: 'South Korea', 48: 'India', 49: 'Indonesia', 50: 'Australia', 51: 'South Africa', 52: 'Republic of Moldova', - 53: 'Portugal', 54: 'Ireland', 55: 'Denmark', 56: 'Iran', 57: 'Pakistan', 58: 'Israel', 59: 'Thailand', - 61: 'Slovenia', 63: 'Croatia', 64: 'Chile', 65: 'Serbia', 66: 'Malaysia', 67: 'Philippines', 68: 'Singapore', - 69: 'Bosnia and Herzegovina', 70: 'Estonia', 71: 'Latvia', 72: 'Lithuania', 73: 'North Korea', 74: 'Uruguay', - 75: 'Paraguay', 76: 'Bolivia', 77: 'Peru', 78: 'Colombia', 79: 'Republic of Macedonia (FYROM)', 80: 'Montenegro', - 81: 'Republic of China (Taiwan)', 82: 'Cyprus', 83: 'Belarus', 84: 'New Zealand', 164: 'Saudi Arabia', 165: 'Egypt', - 166: 'United Arab Emirates', 167: 'Albania', 168: 'Georgia', 169: 'Armenia', 170: 'Nigeria', 171: 'Cuba' -} - -COUNTRY_LINK: Dict[int, str] = { - 1: 'Romania', 9: 'Brazil', 11: 'France', 12: 'Germany', 13: 'Hungary', 82: 'Cyprus', 168: 'Georgia', 15: 'Spain', - 23: 'Canada', 26: 'Mexico', 27: 'Argentina', 28: 'Venezuela', 80: 'Montenegro', 24: 'USA', 29: 'United-Kingdom', - 50: 'Australia', 47: 'South-Korea', 171: 'Cuba', 79: 'Republic-of-Macedonia-FYROM', 30: 'Switzerland', 165: 'Egypt', - 31: 'Netherlands', 32: 'Belgium', 33: 'Austria', 34: 'Czech-Republic', 35: 'Poland', 36: 'Slovakia', 37: 'Norway', - 38: 'Sweden', 39: 'Finland', 40: 'Ukraine', 41: 'Russia', 42: 'Bulgaria', 43: 'Turkey', 44: 'Greece', 45: 'Japan', - 48: 'India', 49: 'Indonesia', 78: 'Colombia', 68: 'Singapore', 51: 'South Africa', 52: 'Republic-of-Moldova', - 53: 'Portugal', 54: 'Ireland', 55: 'Denmark', 56: 'Iran', 57: 'Pakistan', 58: 'Israel', 59: 'Thailand', 10: 'Italy', - 61: 'Slovenia', 63: 'Croatia', 64: 'Chile', 65: 'Serbia', 66: 'Malaysia', 67: 'Philippines', 70: 'Estonia', - 77: 'Peru', 71: 'Latvia', 72: 'Lithuania', 73: 'North-Korea', 74: 'Uruguay', 75: 'Paraguay', 76: 'Bolivia', - 81: 'Republic-of-China-Taiwan', 166: 'United-Arab-Emirates', 167: 'Albania', 69: 'Bosnia-Herzegovina', 14: 'China', - 169: 'Armenia', 83: 'Belarus', 84: 'New-Zealand', 164: 'Saudi-Arabia', 170: 'Nigeria' -} - -ISO_CC: Dict[int, str] = { - 1: 'ROU', 9: 'BRA', 10: 'ITA', 11: 'FRA', 12: 'DEU', 13: 'HUN', 14: 'CHN', 15: 'ESP', 23: 'CAN', 24: 'USA', - 26: 'MEX', 27: 'ARG', 28: 'VEN', 29: 'GBR', 30: 'CHE', 31: 'NLD', 32: 'BEL', 33: 'AUT', 34: 'CZE', 35: 'POL', - 36: 'SVK', 37: 'NOR', 38: 'SWE', 39: 'FIN', 40: 'UKR', 41: 'RUS', 42: 'BGR', 43: 'TUR', 44: 'GRC', 45: 'JPN', - 47: 'KOR', 48: 'IND', 49: 'IDN', 50: 'AUS', 51: 'ZAF', 52: 'MDA', 53: 'PRT', 54: 'IRL', 55: 'DNK', 56: 'IRN', - 57: 'PAK', 58: 'ISR', 59: 'THA', 61: 'SVN', 63: 'HRV', 64: 'CHL', 65: 'SRB', 66: 'MYS', 67: 'PHL', 68: 'SGP', - 69: 'BiH', 70: 'EST', 71: 'LVA', 72: 'LTU', 73: 'PRK', 74: 'URY', 75: 'PRY', 76: 'BOL', 77: 'PER', 78: 'COL', - 79: 'MKD', 80: 'MNE', 81: 'TWN', 82: 'CYP', 83: 'BLR', 84: 'NZL', 164: 'SAU', 165: 'EGY', 166: 'UAE', 167: 'ALB', - 168: 'GEO', 169: 'ARM', 170: 'NGA', 171: 'CUB', -} - def now() -> datetime.datetime: return datetime.datetime.now(erep_tz).replace(microsecond=0)