Compare commits

..

17 Commits

10 changed files with 87 additions and 93 deletions

View File

@ -2,6 +2,18 @@
History
=======
0.22.2 (2020-11-9)
-------------------
* Allow querying market offers for q2-q5 aircrafts
* Added "Ticket" industry
0.22.1 (2020-11-4)
-------------------
* Requirement update
* Unified product naming in inventory and other places based on `erepublik.constants.INDUSTRIES` values
* `erepublik.Citizen` parameter `auto_login` now defaults to `False`
* Continued work on more verbose action and result logging
0.22.0 (2020-10-22)
-------------------
* Ability to dump session and restore from file

View File

@ -4,9 +4,9 @@
__author__ = """Eriks Karls"""
__email__ = 'eriks@72.lv'
__version__ = '0.22.1'
__version__ = '0.22.2'
from erepublik import classes, utils, constants
from erepublik.citizen import Citizen
__all__ = ["classes", "utils", "Citizen", ]
__all__ = ["classes", "utils", "Citizen", 'constants']

View File

@ -346,11 +346,12 @@ class ErepublikEconomyAPI(CitizenBaseAPI):
def _post_economy_marketplace_actions(self, action: str, **kwargs) -> Response:
if action == 'buy':
data = dict(_token=self.token, offerId=kwargs['offer'], amount=kwargs['amount'], orderBy="price_asc", currentPage=1,
buyAction=1)
data = dict(_token=self.token, offerId=kwargs['offer'], amount=kwargs['amount'],
orderBy="price_asc", currentPage=1, buyAction=1)
elif action == 'sell':
data = dict(_token=self.token, countryId=kwargs["country_id"], price=kwargs["price"],
industryId=kwargs["industry"], quality=kwargs["quality"], amount=kwargs['amount'], sellAction='postOffer')
industryId=kwargs["industry"], quality=kwargs["quality"], amount=kwargs['amount'],
sellAction='postOffer')
elif action == 'delete':
data = dict(_token=self.token, offerId=kwargs["offer_id"], sellAction='deleteOffer')
else:
@ -359,16 +360,16 @@ class ErepublikEconomyAPI(CitizenBaseAPI):
class ErepublikLeaderBoardAPI(CitizenBaseAPI):
def _get_main_leaderboards_damage_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response:
def _get_main_leaderboards_damage_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-damage-aircraft-rankings/{country_id}/{weeks}/{mu_id}/0")
def _get_main_leaderboards_damage_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response:
def _get_main_leaderboards_damage_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-damage-rankings/{country_id}/{weeks}/{mu_id}/{div}")
def _get_main_leaderboards_kills_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response:
def _get_main_leaderboards_kills_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-kills-aircraft-rankings/{country_id}/{weeks}/{mu_id}/0")
def _get_main_leaderboards_kills_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response:
def _get_main_leaderboards_kills_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-kills-rankings/{country_id}/{weeks}/{mu_id}/{div}")

View File

@ -264,6 +264,8 @@ class BaseCitizen(access_points.CitizenAPI):
kind = re.sub(r'_q\d\d*', "", item_data.get('token'))
else:
kind = item_data.get('type')
if constants.INDUSTRIES[kind]:
kind = constants.INDUSTRIES[constants.INDUSTRIES[kind]]
if kind not in active_items:
active_items[kind] = {}
icon = item_data['icon'] if item_data[
@ -316,11 +318,15 @@ class BaseCitizen(access_points.CitizenAPI):
self.eb_small += amount
elif q == 15:
self.eb_small += amount
item_data.update(token='energy_bar')
kind = re.sub(r'_q\d\d*', "", item_data.get('token'))
if item_data.get('token', "") == "house_q100":
self.ot_points = item_data['amount']
if constants.INDUSTRIES[kind]:
kind = constants.INDUSTRIES[constants.INDUSTRIES[kind]]
if kind not in final_items:
final_items[kind] = {}
@ -339,10 +345,10 @@ class BaseCitizen(access_points.CitizenAPI):
icon = "/images/modules/pvp/ghost_boosters/icon_booster_30_60.png"
else:
icon = "//www.erepublik.net/images/modules/manager/tab_storage.png"
item_data = dict(kind=kind, quality=item_data.get('quality', 0), amount=item_data.get('amount', 0),
durability=item_data.get('duration', 0), icon=icon, name=name)
_item_data = dict(kind=kind, quality=item_data.get('quality', 0), amount=item_data.get('amount', 0),
durability=item_data.get('duration', 0), icon=icon, name=name)
if item_data.get('type') in ('damageBoosters', "aircraftDamageBoosters"):
item_data = {item_data['durability']: item_data}
_item_data = {_item_data['durability']: _item_data}
else:
if item_data.get('type') == 'bomb':
firepower = 0
@ -351,9 +357,9 @@ class BaseCitizen(access_points.CitizenAPI):
except AttributeError:
pass
finally:
item_data.update(fire_power=firepower)
item_data = {item_data['quality']: item_data}
final_items[kind].update(item_data)
_item_data.update(fire_power=firepower)
_item_data = {_item_data['quality']: _item_data}
final_items[kind].update(_item_data)
raw_materials: Dict[str, Dict[int, Dict[str, Union[str, int]]]] = {}
if data.get("rawMaterials", {}).get("items", {}):
@ -908,6 +914,7 @@ class CitizenCompanies(BaseCitizen):
if int(free_inventory * 0.75) < self.my_companies.get_needed_inventory_usage(wam_list):
self.update_inventory()
free_inventory = self.inventory_status["total"] - self.inventory_status["used"]
while wam_list and free_inventory < self.my_companies.get_needed_inventory_usage(wam_list):
wam_list.pop(-1)
@ -976,7 +983,7 @@ class CitizenEconomy(CitizenTravel):
def check_house_durability(self) -> Dict[int, datetime]:
ret = {}
inv = self.get_inventory()
for house_quality, active_house in inv['active'].get('house', {}).items():
for house_quality, active_house in inv['active'].get('House', {}).items():
till = utils.good_timedelta(self.now, timedelta(seconds=active_house['time_left']))
ret.update({house_quality: till})
return ret
@ -985,14 +992,14 @@ class CitizenEconomy(CitizenTravel):
original_region = self.details.current_country, self.details.current_region
ok_to_activate = False
inv = self.get_inventory()
if not inv['final'].get('house', {}).get(q, {}):
if not inv['final'].get('House', {}).get(q, {}):
countries = [self.details.citizenship, ]
if self.details.current_country != self.details.citizenship:
countries.append(self.details.current_country)
offers = [self.get_market_offers("house", q, country)[f"q{q}"] for country in countries]
offers = [self.get_market_offers("House", q, country)[f"q{q}"] for country in countries]
local_cheapest = sorted(offers, key=lambda o: o.price)[0]
global_cheapest = self.get_market_offers("house", q)[f"q{q}"]
global_cheapest = self.get_market_offers("House", q)[f"q{q}"]
if global_cheapest.price + 200 < local_cheapest.price:
if self.travel_to_country(global_cheapest.country):
buy = self.buy_from_market(global_cheapest.offer_id, 1)
@ -1026,10 +1033,10 @@ class CitizenEconomy(CitizenTravel):
return house_durability
def activate_house(self, quality: int) -> bool:
r = self._post_economy_activate_house(quality).json()
r: Dict[str, Any] = self._post_economy_activate_house(quality).json()
self._update_inventory_data(r)
if r.get("status") and not r.json().get("error"):
house: Dict[str, Union[int, str]] = self.get_inventory()['active']['house'][quality]
if r.get("status") and not r.get("error"):
house: Dict[str, Union[int, str]] = self.get_inventory()['active']['House'][quality]
time_left = timedelta(seconds=house["time_left"])
active_until = utils.good_timedelta(self.now, time_left)
self._report_action(
@ -1085,7 +1092,7 @@ class CitizenEconomy(CitizenTravel):
offers = self.get_my_market_offers()
for offer in offers:
if offer['id'] == offer_id:
industry = constants.INDUSTRIES[offer['industry']]
industry = constants.INDUSTRIES[offer['industryId']]
amount = offer['amount']
q = offer['quality']
price = offer['price']
@ -1097,7 +1104,7 @@ class CitizenEconomy(CitizenTravel):
self._report_action("ECONOMY_DELETE_OFFER",
f"Removed offer for {amount} x {industry} q{q} for {price}cc/each",
kwargs=offer)
return ret.get('error')
return not ret.get('error')
else:
self._report_action("ECONOMY_DELETE_OFFER", f"Unable to find offer id{offer_id}", kwargs={'offers': offers})
return False
@ -1128,25 +1135,24 @@ class CitizenEconomy(CitizenTravel):
message = (f"Posted market offer for {amount}q{quality} "
f"{constants.INDUSTRIES[industry]} for price {price}cc")
self._report_action("ECONOMY_SELL_PRODUCTS", message, kwargs=ret)
return bool(ret.get('error', True))
return not bool(ret.get('error', True))
def buy_from_market(self, offer: int, amount: int) -> dict:
ret = self._post_economy_marketplace_actions('buy', offer=offer, amount=amount)
json_ret = ret.json()
if json_ret.get('error'):
return json_ret
else:
if not json_ret.get('error', True):
self.details.cc = ret.json()['currency']
self.details.gold = ret.json()['gold']
json_ret.pop("offerUpdate", None)
self._report_action("BOUGHT_PRODUCTS", "", kwargs=json_ret)
self._report_action("BOUGHT_PRODUCTS", json_ret.get('message'), kwargs=json_ret)
return json_ret
def get_market_offers(
self, product_name: str, quality: int = None, country: constants.Country = None
) -> Dict[str, classes.OfferItem]:
raw_short_names = dict(frm="foodRaw", wrm="weaponRaw", hrm="houseRaw", arm="airplaneRaw")
q1_industries = ["aircraft"] + list(raw_short_names.values())
q1_industries = list(raw_short_names.values())
q5_industries = ['house', 'aircraft', 'ticket']
if product_name in raw_short_names:
quality = 1
product_name = raw_short_names[product_name]
@ -1160,7 +1166,7 @@ class CitizenEconomy(CitizenTravel):
if quality:
offers[f"q{quality}"] = classes.OfferItem()
else:
max_quality = 1 if product_name in q1_industries else 5 if product_name == 'house' else 7
max_quality = 1 if product_name in q1_industries else 5 if product_name.lower() in q5_industries else 7
for q in range(max_quality):
offers[f"q{q + 1}"] = classes.OfferItem()
@ -1494,51 +1500,6 @@ class CitizenMilitary(CitizenTravel):
f" new influence {influence}", kwargs=r.json())
return influence
# def check_epic_battles(self):
# active_fs = False
# for battle_id in self.sorted_battles(self.config.sort_battles_time):
# battle = self.all_battles.get(battle_id)
# if not battle.is_air:
# my_div: BattleDivision = battle.div.get(self.division)
# if my_div.epic and my_div.end > self.now:
# if self.energy.food_fights > 50:
# inv_allies = battle.invader.deployed + [battle.invader.id]
# def_allies = battle.defender.deployed + [battle.defender.id]
# all_allies = inv_allies + def_allies
# if self.details.current_country not in all_allies:
# if self.details.current_country in battle.invader.allies:
# allies = battle.invader.deployed
# side = battle.invader.id
# else:
# allies = battle.defender.deployed
# side = battle.defender.id
#
# self.travel_to_battle(battle.id, allies)
#
# else:
# if self.details.current_country in inv_allies:
# side = battle.invader.id
# elif self.details.current_country in def_allies:
# side = battle.defender.id
# else:
# self.write_log(
# f"Country {self.details.current_country} not in all allies list ({all_allies}) and "
# f"also not in inv allies ({inv_allies}) nor def allies ({def_allies})")
# break
# error_count = 0
# while self.energy.food_fights > 5 and error_count < 20:
# errors = self.fight(battle_id, side_id=side, count=self.energy.food_fights - 5)
# if errors:
# error_count += errors
# if self.config.epic_hunt_ebs:
# self._eat('orange')
# self.travel_to_residence()
# break
# elif bool(my_div.epic):
# active_fs = True
#
# self.active_fs = active_fs
def sorted_battles(self, sort_by_time: bool = True, only_tp=False) -> List[classes.Battle]:
cs_battles_priority_air: List[classes.Battle] = []
cs_battles_priority_ground: List[classes.Battle] = []
@ -2609,6 +2570,10 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
raw_kind = raw_kind.group(1)
result = response.get("result", {})
amount_needed = round(result.get("consume", 0) - result.get("stock", 0) + 0.49)
self._report_action(
'WORK_AS_MANAGER', f"Unable to wam! Missing {amount_needed} {raw_kind}, will try to buy.",
kwargs=response
)
start_place = (self.details.current_country, self.details.current_region)
while amount_needed > 0:
amount = amount_needed
@ -2617,20 +2582,31 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
if not best_offer.country == self.details.current_country:
self.travel_to_country(best_offer.country)
self._report_action("ECONOMY_BUY", f"Attempting to buy {amount} {raw_kind} for {best_offer.price*amount}cc")
rj = self.buy_from_market(amount=amount, offer=best_offer.offer_id)
if not rj.get('error'):
amount_needed -= amount
else:
self.write_log(rj.get('message', ""))
self._report_action(
"ECONOMY_BUY", f"Unable to buy products! Reason: {rj.get('message')}", kwargs=rj
)
break
else:
if not start_place == (self.details.current_country, self.details.current_region):
self.travel_to_holding(holding)
self._wam(holding)
return
if not start_place == (self.details.current_country, self.details.current_region):
self.travel_to_residence()
return
elif response.get("message") == "not_enough_health_food":
self.buy_food()
self._wam(holding)
elif response.get("message") == "tax_money":
self._report_action("WORK_AS_MANAGER", "Not enough money to work as manager!", kwargs=response)
self.write_log("Not enough money to work as manager!")
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)

View File

@ -3,7 +3,7 @@ import hashlib
import threading
import weakref
from decimal import Decimal
from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union, NoReturn, Generator, Iterable
from typing import Any, Dict, List, NamedTuple, Tuple, Union, NoReturn, Generator, Iterable
from requests import Response, Session, post
@ -314,7 +314,9 @@ class MyCompanies:
self._companies.clear()
@property
def as_dict(self) -> Dict[str, Union[str, int, datetime.datetime, Dict[str, Dict[str, Union[str, int, List[Dict[str, Union[str, int, bool, float, Decimal]]]]]]]]:
def as_dict(self) -> Dict[str, Union[str, int, datetime.datetime, Dict[str, Dict[str, Union[
str, int, List[Dict[str, Union[str, int, bool, float, Decimal]]]]
]]]]:
return dict(name=str(self), work_units=self.work_units, next_ot_time=self.next_ot_time,
ff_lockdown=self.ff_lockdown,
holdings={str(hi): h.as_dict for hi, h in self.holdings.items()},
@ -857,13 +859,15 @@ class Battle:
self.invader = BattleSide(
self, constants.COUNTRIES[battle.get('inv', {}).get('id')], battle.get('inv', {}).get('points'),
[constants.COUNTRIES[row.get('id')] for row in battle.get('inv', {}).get('ally_list')],
[constants.COUNTRIES[row.get('id')] for row in battle.get('inv', {}).get('ally_list') if row['deployed']], False
[constants.COUNTRIES[row.get('id')] for row in battle.get('inv', {}).get('ally_list') if row['deployed']],
False
)
self.defender = BattleSide(
self, constants.COUNTRIES[battle.get('def', {}).get('id')], battle.get('def', {}).get('points'),
[constants.COUNTRIES[row.get('id')] for row in battle.get('def', {}).get('ally_list')],
[constants.COUNTRIES[row.get('id')] for row in battle.get('def', {}).get('ally_list') if row['deployed']], True
[constants.COUNTRIES[row.get('id')] for row in battle.get('def', {}).get('ally_list') if row['deployed']],
True
)
self.div = {}
@ -890,7 +894,8 @@ class Battle:
else:
time_part = "-{}".format(self.start - time_now)
return f"Battle {self.id} for {self.region_name[:16]} | {self.invader} : {self.defender} | Round time {time_part}"
return (f"Battle {self.id} for {self.region_name[:16]} | "
f"{self.invader} : {self.defender} | Round time {time_part}")
def __repr__(self):
return f"<Battle #{self.id} {self.invader}:{self.defender} R{self.zone_id}>"

View File

@ -53,14 +53,14 @@ class Country:
class Industries:
__by_name = {'food': 1, 'weapon': 2, 'house': 4, 'aircraft': 23,
__by_name = {'food': 1, 'weapon': 2, 'ticket':3, 'house': 4, 'aircraft': 23,
'foodraw': 7, 'weaponraw': 12, 'houseraw': 18, 'aircraftraw': 24,
'frm': 7, 'wrm': 12, 'hrm': 18, 'arm': 24,
'frm q1': 7, 'frm q2': 8, 'frm q3': 9, 'frm q4': 10, 'frm q5': 11,
'wrm q1': 12, 'wrm q2': 13, 'wrm q3': 14, 'wrm q4': 15, 'wrm q5': 16,
'hrm q1': 18, 'hrm q2': 19, 'hrm q3': 20, 'hrm q4': 21, 'hrm q5': 22,
'arm q1': 24, 'arm q2': 25, 'arm q3': 26, 'arm q4': 27, 'arm q5': 28}
__by_id = {1: "Food", 2: "Weapon", 4: "House", 23: "Aircraft",
__by_id = {1: "Food", 2: "Weapon", 3: "Ticket", 4: "House", 23: "Aircraft",
7: "foodRaw", 8: "FRM q2", 9: "FRM q3", 10: "FRM q4", 11: "FRM q5",
12: "weaponRaw", 13: "WRM q2", 14: "WRM q3", 15: "WRM q4", 16: "WRM q5",
17: "houseRaw", 18: "houseRaw", 19: "HRM q2", 20: "HRM q3", 21: "HRM q4", 22: "HRM q5",

View File

@ -2,15 +2,15 @@ bump2version==1.0.1
coverage==5.3
edx-sphinx-theme==1.5.0
flake8==3.8.4
ipython>=7.18.1
ipython>=7.19.0
isort==5.6.4
pip==20.2.4
PyInstaller==4.0
pytz==2020.1
pytest==6.1.1
pytz==2020.4
pytest==6.1.2
responses==0.12.0
setuptools==50.3.2
Sphinx==3.2.1
Sphinx==3.3.0
requests==2.24.0
PySocks==1.7.1
tox==3.20.1

View File

@ -1,5 +1,5 @@
[bumpversion]
current_version = 0.22.1
current_version = 0.22.2
commit = True
tag = True
parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)\.?(?P<dev>\d+)?

View File

@ -12,7 +12,7 @@ with open('HISTORY.rst') as history_file:
history = history_file.read()
requirements = [
'pytz==2020.1',
'pytz==2020.4',
'requests==2.24.0',
'PySocks==1.7.1'
]
@ -20,7 +20,7 @@ requirements = [
setup_requirements = []
test_requirements = [
"pytest==6.1.0",
"pytest==6.1.2",
"responses==0.12.0"
]
@ -50,6 +50,6 @@ setup(
test_suite='tests',
tests_require=test_requirements,
url='https://github.com/eeriks/erepublik/',
version='0.22.1',
version='0.22.2',
zip_safe=False,
)

View File

@ -2,7 +2,6 @@
# -*- coding: utf-8 -*-
"""Tests for `erepublik` package."""
from typing import Callable
from erepublik import Citizen
@ -66,7 +65,8 @@ class TestErepublik(unittest.TestCase):
self.assertEqual(self.citizen.next_reachable_energy, 0)
def test_should_fight(self):
is_wc_close: Callable[[], bool] = lambda: self.citizen.max_time_till_full_ff > self.citizen.time_till_week_change
def is_wc_close():
return self.citizen.max_time_till_full_ff > self.citizen.time_till_week_change
self.citizen.config.fight = False
self.assertEqual(self.citizen.should_fight(), (0, "Fighting not allowed!", False))