diff --git a/erepublik/citizen.py b/erepublik/citizen.py
index 4587f31..2c4a4ab 100644
--- a/erepublik/citizen.py
+++ b/erepublik/citizen.py
@@ -8,128 +8,35 @@ from typing import Any, Dict, List, Optional, Set, Tuple, Union
from requests import RequestException, Response
+from erepublik import utils
from erepublik.classes import (Battle, BattleDivision, CitizenAPI, Config, Details, Energy, ErepublikException,
MyCompanies, MyJSONEncoder, Politics, Reporter, TelegramBot)
-from erepublik.utils import *
-
-try:
- import simplejson as json
-except ImportError:
- import json
-class Citizen(CitizenAPI):
- division = 0
-
- all_battles: Dict[int, Battle] = None
- countries: Dict[int, Dict[str, Union[str, List[int]]]] = None
- __last_war_update_data = None
- __last_full_update: datetime = now().min
-
- active_fs: bool = False
+class BaseCitizen(CitizenAPI):
+ __last_full_update: datetime = utils.now().min
+ promos: Dict[str, datetime] = None
food: Dict[str, int] = {"q1": 0, "q2": 0, "q3": 0, "q4": 0, "q5": 0, "q6": 0, "q7": 0, "total": 0}
- inventory: Dict[str, int] = {"used": 0, "total": 0}
- boosters: Dict[int, Dict[int, int]] = {100: {}, 50: {}}
-
eb_normal = 0
eb_double = 0
eb_small = 0
-
- work_units = 0
- ot_points = 0
-
- tg_contract = None
- promos: Dict[str, datetime] = None
+ division = 0
eday = 0
+ config: Config = None
energy: Energy = None
details: Details = None
politics: Politics = None
- my_companies: MyCompanies = None
reporter: Reporter = None
stop_threads: Event = None
telegram: TelegramBot = None
r: Response = None
- name = "Not logged in!"
- debug = False
- __registered = False
- logged_in = False
-
- def __init__(self, email: str = "", password: str = "", auto_login: bool = True):
- super().__init__()
- self.commit_id = COMMIT_ID
- self.config = Config()
- self.config.email = email
- self.config.password = password
- self.energy = Energy()
- self.details = Details()
- self.politics = Politics()
- self.my_companies = MyCompanies()
- self.set_debug(True)
- self.reporter = Reporter()
- self.stop_threads = Event()
- self.telegram = TelegramBot(stop_event=self.stop_threads)
- if auto_login:
- self.login()
-
- def config_setup(self, **kwargs):
- self.config.reset()
- for key, value in kwargs.items():
- if hasattr(self.config, key):
- setattr(self.config, key, value)
- else:
- self.write_log(f"Unknown config parameter! ({key}={value})")
-
- def login(self):
- self.get_csrf_token()
-
- self.update_citizen_info()
- self.reporter.do_init(self.name, self.config.email, self.details.citizen_id)
- if self.config.telegram:
- self.telegram.do_init(self.config.telegram_chat_id or 620981703,
- self.config.telegram_token or "864251270:AAFzZZdjspI-kIgJVk4gF3TViGFoHnf8H4o",
- "" if self.config.telegram_chat_id or self.config.telegram_token else self.name)
- self.telegram.send_message(f"*Started* {now():%F %T}")
-
- self.__last_full_update = good_timedelta(self.now, - timedelta(minutes=5))
-
- def write_log(self, *args, **kwargs):
- if self.config.interactive:
- write_interactive_log(*args, **kwargs)
- else:
- write_silent_log(*args, **kwargs)
-
- def sleep(self, seconds: int):
- if seconds < 0:
- seconds = 0
- if self.config.interactive:
- interactive_sleep(seconds)
- else:
- sleep(seconds)
-
- def __str__(self) -> str:
- return f"Citizen {self.name}"
-
- def __repr__(self):
- return self.__str__()
-
- @property
- def __dict__(self):
- ret = super().__dict__.copy()
- ret.pop('stop_threads', None)
- ret.pop('_Citizen__last_war_update_data', None)
-
- return ret
-
- def set_debug(self, debug: bool):
- self.debug = debug
- self._req.debug = debug
-
- def set_pin(self, pin: int):
- self.details.pin = pin
+ name: str = "Not logged in!"
+ logged_in: bool = False
+ commit_id: str = ""
def get_csrf_token(self):
"""
@@ -144,7 +51,7 @@ class Citizen(CitizenAPI):
return
html = resp.text
- self.check_for_medals(html)
+ self._check_response_for_medals(html)
re_token = re.search(r'var csrfToken = \'(\w{32})\'', html)
re_login_token = re.search(r'', html)
if re_token:
@@ -156,7 +63,7 @@ class Citizen(CitizenAPI):
raise ErepublikException("Something went wrong! Can't find token in page! Exiting!")
try:
self.update_citizen_info(resp.text)
- except (AttributeError, json.JSONDecodeError, ValueError, KeyError):
+ except (AttributeError, utils.json.JSONDecodeError, ValueError, KeyError):
pass
def _login(self):
@@ -183,7 +90,7 @@ class Citizen(CitizenAPI):
if j['error'] and j["message"] == 'Too many requests':
self.write_log("Made too many requests! Sleeping for 30 seconds.")
self.sleep(30)
- except (json.JSONDecodeError, KeyError, TypeError):
+ except (utils.json.JSONDecodeError, KeyError, TypeError):
pass
if response.status_code >= 400:
self.r = response
@@ -213,14 +120,14 @@ class Citizen(CitizenAPI):
try:
self.update_citizen_info(response.text)
- except (AttributeError, json.JSONDecodeError, ValueError, KeyError):
+ except (AttributeError, utils.json.JSONDecodeError, ValueError, KeyError):
pass
if self._errors_in_response(response):
self.get_csrf_token()
self.get(url, **kwargs)
else:
- self.check_for_medals(response.text)
+ self._check_response_for_medals(response.text)
self.r = response
return response
@@ -247,8 +154,8 @@ class Citizen(CitizenAPI):
try:
resp_data = response.json()
if (resp_data.get("error") or not resp_data.get("status")) and resp_data.get("message", "") == "captcha":
- send_email(self.name, [response.text, ], player=self, captcha=True)
- except:
+ utils.send_email(self.name, [response.text, ], player=self, captcha=True)
+ except (AttributeError, utils.json.JSONDecodeError, ValueError, KeyError):
pass
if self._errors_in_response(response):
@@ -259,12 +166,80 @@ class Citizen(CitizenAPI):
json.update({"_token": self.token})
response = self.post(url, data=data, json=json, **kwargs)
else:
- self.check_for_medals(response.text)
+ self._check_response_for_medals(response.text)
self.r = response
return response
- def check_for_medals(self, html: str):
+ def update_citizen_info(self, html: str = None):
+ """
+ Gets main page and updates most information about player
+ """
+ if html is None:
+ self._get_main()
+ return
+ ugly_js = re.search(r'"promotions":\s*(\[{?.*?}?])', html).group(1)
+ promos = utils.json.loads(utils.normalize_html_json(ugly_js))
+ if self.promos is None:
+ self.promos = {}
+ else:
+ self.promos = {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")
+ expire = utils.localize_timestamp(int(promo.get("expiresAt")))
+ if promo_name not in self.promos:
+ send_mail = True
+ self.promos.update({promo_name: expire})
+ if send_mail:
+ active_promos = []
+ for kind, time_until in self.promos.items():
+ active_promos.append(f"{kind} active until {time_until}")
+ utils.report_promo(kind, time_until)
+ utils.send_email(self.name, active_promos, player=self, promo=True)
+
+ new_date = re.search(r"var new_date = '(\d*)';", html)
+ if new_date:
+ self.energy.set_reference_time(
+ utils.good_timedelta(self.now, timedelta(seconds=int(new_date.group(1))))
+ )
+
+ ugly_js = re.search(r"var erepublik = ({.*}),\s+", html).group(1)
+ citizen_js = utils.json.loads(ugly_js)
+ citizen = citizen_js.get("citizen", {})
+
+ self.eday = citizen_js.get("settings").get("eDay")
+ self.division = int(citizen.get("division", 0))
+
+ self.energy.interval = citizen.get("energyPerInterval", 0)
+ self.energy.limit = citizen.get("energyToRecover", 0)
+ self.energy.recovered = citizen.get("energy", 0)
+ self.energy.recoverable = citizen.get("energyFromFoodRemaining", 0)
+
+ self.details.current_region = citizen.get("regionLocationId", 0)
+ self.details.current_country = citizen.get("countryLocationId", 0) # country where citizen is located
+ self.details.residence_region = citizen.get("residence", {}).get("regionId", 0)
+ self.details.residence_country = citizen.get("residence", {}).get("countryId", 0)
+ self.details.citizen_id = citizen.get("citizenId", 0)
+ self.details.citizenship = int(citizen.get("country", 0))
+ self.details.xp = citizen.get("currentExperiencePoints", 0)
+ self.details.daily_task_done = citizen.get("dailyTasksDone", False)
+ self.details.daily_task_reward = citizen.get("hasReward", False)
+ if citizen.get("dailyOrderDone", False) and not citizen.get("hasDailyOrderReward", False):
+ self._post_military_group_missions()
+
+ self.details.next_pp.sort()
+ for skill in citizen.get("mySkills", {}).values():
+ self.details.mayhem_skills.update({int(skill["terrain_id"]): int(skill["skill_points"])})
+
+ if citizen.get('party', []):
+ party = citizen.get('party')
+ self.politics.is_party_member = True
+ self.politics.party_id = party.get('party_id')
+ self.politics.is_party_president = bool(party.get('is_party_president'))
+ self.politics.party_slug = f"{party.get('stripped_title')}-{party.get('party_id')}"
+
+ def _check_response_for_medals(self, html: str):
new_medals = re.findall(r'()',
html, re.M | re.S | re.I)
data: Dict[Tuple[str, Union[float, str]], Dict[str, Union[int, str, float]]] = {}
@@ -293,12 +268,12 @@ class Citizen(CitizenAPI):
except AttributeError:
continue
if data:
-
msgs = ["{count} x {kind}, totaling {} {currency}\n"
"{about}".format(d["count"] * d["reward"], **d) for d in data.values()]
msgs = "\n".join(msgs)
- self.telegram.report_medal(msgs)
+ if self.config.telegram:
+ self.telegram.report_medal(msgs)
self.write_log(f"Found awards:\n{msgs}")
for info in data.values():
self.reporter.report_action("NEW_MEDAL", info)
@@ -308,171 +283,413 @@ class Citizen(CitizenAPI):
level = levelup.group(1)
msg = f"Level up! Current level {level}"
self.write_log(msg)
- self.telegram.report_medal(f"Level *{level}*")
+ if self.config.telegram:
+ self.telegram.report_medal(f"Level *{level}*")
self.reporter.report_action("LEVEL_UP", value=level)
- def check_for_notification_medals(self):
- notifications = self._get_main_citizen_daily_assistant().json()
- data: Dict[Tuple[str, Union[float, str]], Dict[str, Union[int, str, float]]] = {}
- for medal in notifications.get('notifications', []):
- if medal.get('details', {}).get('type') == "citizenAchievement":
- params = medal.get('details', {}).get('achievement')
- about = medal.get('details').get('description')
- title = medal.get('title')
- # award_id = re.search(r'"wall_enable_alerts_(\d+)', medal)
- # if award_id:
- # self._post_main_wall_post_automatic(**{'message': title, 'awardId': award_id.group(1)})
-
- if params.get('ccValue'):
- reward = params.get('ccValue')
- currency = "Currency"
- elif params.get('goldValue'):
- reward = params.get('goldValue')
- currency = "Gold"
- else:
- reward = params.get('energyValue')
- currency = "Energy"
-
- if (title, reward) not in data:
- data[(title, reward)] = {'about': about, 'kind': title, 'reward': reward, "count": 1,
- "currency": currency, "params": params}
- else:
- data[(title, reward)]['count'] += 1
- self._post_main_global_alerts_close(medal.get('id'))
- if data:
- msgs = ["{count} x {kind},"
- " totaling {} {currency}".format(d["count"] * d["reward"], **d) for d in data.values()]
-
- msgs = "\n".join(msgs)
- self.telegram.report_medal(msgs)
- self.write_log(f"Found awards:\n{msgs}")
- for info in data.values():
- self.reporter.report_action("NEW_MEDAL", info)
-
- def update_all(self, force_update=False):
- # Do full update max every 5 min
- if good_timedelta(self.__last_full_update, timedelta(minutes=5)) > self.now and not force_update:
- return
+ def write_log(self, *args, **kwargs):
+ if self.config.interactive:
+ utils.write_interactive_log(*args, **kwargs)
else:
- self.__last_full_update = self.now
+ utils.write_silent_log(*args, **kwargs)
+
+ def report_error(self, msg: str = "", is_warning: bool = False):
+ if is_warning:
+ utils.process_warning(msg, self.name, sys.exc_info(), self, self.commit_id)
+ else:
+ utils.process_error(msg, self.name, sys.exc_info(), self, self.commit_id, None)
+
+ def sleep(self, seconds: int):
+ if seconds < 0:
+ seconds = 0
+ if self.config.interactive:
+ utils.interactive_sleep(seconds)
+ else:
+ sleep(seconds)
+
+ def __str__(self) -> str:
+ return f"Citizen {self.name}"
+
+ def __repr__(self):
+ return self.__str__()
+
+ @property
+ def __dict__(self):
+ ret = super().__dict__.copy()
+ ret.pop('stop_threads', None)
+ ret.pop('_Citizen__last_war_update_data', None)
+
+ return ret
+
+ def _travel(self, country_id: int, region_id: int = 0) -> Response:
+ data = {
+ "toCountryId": country_id,
+ "inRegionId": region_id,
+ }
+ return self._post_main_travel("moveAction", **data)
+
+ @property
+ def health_info(self):
+ ret = f"{self.energy.recovered}/{self.energy.limit} + {self.energy.recoverable}, " \
+ f"{self.energy.interval}hp/6m. {self.details.xp_till_level_up}xp until level up"
+ return ret
+
+ @property
+ def now(self) -> datetime:
+ """
+ Returns aware datetime object localized to US/Pacific (eRepublik time)
+ :return: datetime
+ """
+ return utils.now()
+
+ def to_json(self, indent: bool = False) -> str:
+ return utils.json.dumps(self.__dict__, cls=MyJSONEncoder, indent=4 if indent else None, sort_keys=True)
+
+ @property
+ def next_reachable_energy(self) -> int:
+ # Return pps for furthest __reachable__ +1 energy else 0
+ max_pp = 0
+ for pp_milestone in self.details.next_pp:
+ pp_milestone = int(pp_milestone)
+ if self.details.pp + self.energy.food_fights > pp_milestone: # if reachable set max pp
+ max_pp = pp_milestone
+ else: # rest are only bigger no need
+ break
+ return max_pp - self.details.pp if max_pp else 0
+
+ @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 utils.good_timedelta(self.now.replace(hour=0, minute=0, second=0, microsecond=0), timedelta(days=days))
+
+ @property
+ def time_till_week_change(self) -> timedelta:
+ return self.next_wc_start - self.now
+
+ @property
+ def time_till_full_ff(self) -> timedelta:
+ energy = self.energy.recoverable + self.energy.recovered
+ if energy >= self.energy.limit * 2:
+ return timedelta(0)
+ minutes_needed = round((self.energy.limit * 2 - energy) / self.energy.interval) * 6
+ return (self.energy.reference_time - self.now) + timedelta(minutes=minutes_needed)
+
+ @property
+ def max_time_till_full_ff(self) -> timedelta:
+ """
+ Max required time for 0 to full energy (0/0 -> limit/limit) (last interval rounded up)
+ :return:
+ """
+ return timedelta(minutes=round((self.energy.limit * 2 / self.energy.interval) + 0.49) * 6)
+
+ @property
+ def is_levelup_close(self) -> bool:
+ """
+ If Energy limit * 2 >= xp till levelup * 10
+ :return: bool
+ """
+ return self.energy.limit * 2 >= self.details.xp_till_level_up * 10
+
+ @property
+ def is_levelup_reachable(self) -> bool:
+ """
+ If Energy limit >= xp till levelup * 10
+ :return: bool
+ """
+ return self.energy.limit >= self.details.xp_till_level_up * 10
+
+ @property
+ def should_do_levelup(self) -> bool:
+ """
+ If Energy limit >= xp till levelup * 10
+ :return: bool
+ """
+ return (self.energy.recovered >= self.details.xp_till_level_up * 10 and # can reach next level
+ self.energy.recoverable + 2 * self.energy.interval >= self.energy.limit) # can do max amount of dmg
+
+ @property
+ def available_industries(self) -> Dict[str, int]:
+ """
+ Returns currently available industries as dict(name: id)
+ :return: dict
+ """
+ return {"food": 1, "weapon": 2, "house": 4, "aircraft": 23,
+ "foodRaw": 7, "weaponRaw": 12, "houseRaw": 17, "airplaneRaw": 24}
+
+ @property
+ def factories(self) -> Dict[int, str]:
+ """Returns factory industries as dict(id: name)
+ :return: Factory id:name dict
+ ":rtype: Dict[int, str]
+ """
+ return {1: "Food", 2: "Weapons", 4: "House", 23: "Aircraft",
+ 7: "FRM q1", 8: "FRM q2", 9: "FRM q3", 10: "FRM q4", 11: "FRM q5",
+ 12: "WRM q1", 13: "WRM q2", 14: "WRM q3", 15: "WRM q4", 16: "WRM q5",
+ 18: "HRM q1", 19: "HRM q2", 20: "HRM q3", 21: "HRM q4", 22: "HRM q5",
+ 24: "ARM q1", 25: "ARM q2", 26: "ARM q3", 27: "ARM q4", 28: "ARM q5", }
+
+ def _get_main_party_members(self, party_id: int) -> Dict[int, str]:
+ ret = {}
+ r = super()._get_main_party_members(party_id)
+ for id_, name in re.findall(r'', r.text):
+ ret.update({id_: name})
+ return ret
+
+ def _eat(self, colour: str = "blue") -> Response:
+ response = self._post_eat(colour)
+ r_json = response.json()
+ next_recovery = r_json.get("food_remaining_reset").split(":")
+ self.energy.set_reference_time(
+ utils.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")
+ return response
+
+
+class CitizenTravel(BaseCitizen):
+ def _update_citizen_location(self, country_id: int, region_id: int):
+ self.details.current_region = region_id
+ self.details.current_country = country_id
+
+ def get_country_travel_region(self, country_id: int) -> int:
+ regions = self.get_travel_regions(country_id=country_id)
+ regs = []
+ if regions:
+ for region in regions.values():
+ 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]
+ else:
+ return 0
+
+ def travel_to_residence(self) -> bool:
self.update_citizen_info()
- self.update_war_info()
- self.update_inventory()
- self.update_companies()
- self.update_money()
- self.update_weekly_challenge()
- self.send_state_update()
- self.check_for_notification_medals()
+ res_r = self.details.residence_region
+ if self.details.residence_country and res_r and not res_r == self.details.current_region:
+ r = self._travel(self.details.residence_country, self.details.residence_region)
+ if r.json().get('message', '') == 'success':
+ self._update_citizen_location(self.details.residence_country, self.details.current_region)
+ return True
+ return False
+ return True
- def update_citizen_info(self, html: str = None):
- """
- Gets main page and updates most information about player
- """
- if html is None:
- self.check_for_notification_medals()
- self._get_main()
- return
- ugly_js = re.search(r'"promotions":\s*(\[{?.*?}?])', html).group(1)
- promos = json.loads(normalize_html_json(ugly_js))
- if self.promos is None:
- self.promos = {}
+ def travel_to_region(self, region_id: int) -> bool:
+ data = self._post_main_travel_data(region_id=region_id).json()
+ if data.get('alreadyInRegion'):
+ return True
else:
- self.promos = {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")
- expire = localize_timestamp(int(promo.get("expiresAt")))
- if promo_name not in self.promos:
- send_mail = True
- self.promos.update({promo_name: expire})
- if promo_name == "trainingContract":
- if not self.tg_contract:
+ r = self._travel(data.get('preselectCountryId'), region_id).json()
+ if r.get('message', '') == 'success':
+ self._update_citizen_location(data.get('preselectCountryId'), region_id)
+ 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()
+
+ regs = []
+ if data.get('regions'):
+ for region in data.get('regions').values():
+ if region['countryId'] == country_id: # Is not occupied by other country
+ regs.append((region['id'], region['distanceInKm']))
+ if regs:
+ region_id = min(regs, key=lambda _: int(_[1]))[0]
+ r = self._travel(country_id, region_id).json()
+ if r.get('message', '') == 'success':
+ self._update_citizen_location(country_id, region_id)
+ return True
+ return False
+
+ def travel_to_holding(self, holding_id: int) -> bool:
+ data = self._post_main_travel_data(holdingId=holding_id).json()
+ if data.get('alreadyInRegion'):
+ return True
+ else:
+ r = self._travel(data.get('preselectCountryId'), data.get('preselectRegionId')).json()
+ if r.get('message', '') == 'success':
+ self._update_citizen_location(data.get('preselectCountryId'), data.get('preselectRegionId'))
+ return True
+ return False
+
+ def get_travel_regions(self, holding_id: int = 0, battle_id: int = 0, country_id: int = 0
+ ) -> 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', [])
+
+ def get_travel_countries(self) -> Set[int]:
+ response_json = self._post_main_travel_data().json()
+ return_list = {*[]}
+ for country_data in response_json['countries'].values():
+ if country_data['currentRegions']:
+ return_list.add(country_data['id'])
+ return return_list
+
+
+class CitizenTasks(BaseCitizen):
+ tg_contract: dict = {}
+ ot_points: int = 0
+ next_ot_time: datetime = None
+
+ def _eat(self, colour: str = "blue") -> Response:
+ resp = super()._eat(colour)
+ for q, amount in resp.json().get("units_consumed").items():
+ if f"q{q}" in self.food:
+ self.food[f"q{q}"] -= amount
+ elif q == "10":
+ self.eb_normal -= amount
+ elif q == "11":
+ self.eb_double -= amount
+ elif q == "12":
+ self.eb_small -= amount
+ return resp
+
+ def work(self):
+ if self.energy.food_fights >= 1:
+ response = self._post_economy_work("work")
+ js = response.json()
+ good_msg = ["already_worked", "captcha"]
+ if not js.get("status") and not js.get("message") in good_msg:
+ if js.get('message') == 'employee':
+ self.find_new_job()
+ self.update_citizen_info()
+ self.work()
+ else:
+ self.reporter.report_action("WORK", json_val=js)
+ else:
+ self._eat("blue")
+ if self.energy.food_fights < 1:
+ seconds = (self.energy.reference_time - self.now).total_seconds()
+ self.write_log("I don't have energy to work. Will sleep for {}s".format(seconds))
+ self.sleep(seconds)
+ self._eat("blue")
+ self.work()
+
+ def train(self):
+ r = self._get_main_training_grounds_json()
+ tg_json = r.json()
+ self.details.gold = tg_json["page_details"]["gold"]
+ self.tg_contract.update(free_train=tg_json["hasFreeTrain"])
+ if tg_json["contracts"]:
+ self.tg_contract.update(**tg_json["contracts"][0])
+
+ tgs = []
+ for data in sorted(tg_json["grounds"], key=lambda k: k["cost"]):
+ if data["default"] and not data["trained"]:
+ tgs.append(data["id"])
+ if tgs:
+ if self.energy.food_fights >= len(tgs):
+ response = self._post_economy_train(tgs)
+ if not response.json().get("status"):
+ self.update_citizen_info()
self.train()
- if not self.tg_contract["free_train"] and self.tg_contract.get("active", False):
- if self.details.gold >= 54:
- self.buy_tg_contract()
- else:
- self.write_log(f"Training ground contract active but "
- f"don't have enough gold ({self.details.gold}g {self.details.cc}cc)")
- if send_mail:
- active_promos = []
- for kind, time_until in self.promos.items():
- active_promos.append(f"{kind} active until {time_until}")
- report_promo(kind, time_until)
- send_email(self.name, active_promos, player=self, promo=True)
+ else:
+ self.reporter.report_action("TRAIN", response.json())
+ else:
+ self._eat("blue")
+ if self.energy.food_fights < len(tgs):
+ large = max(self.energy.reference_time, self.now)
+ small = min(self.energy.reference_time, self.now)
+ self.write_log("I don't have energy to train. Will sleep for {} seconds".format(
+ (large - small).seconds))
+ self.sleep(int((large - small).total_seconds()))
+ self._eat("blue")
+ self.train()
- new_date = re.search(r"var new_date = '(\d*)';", html)
- if new_date:
- self.energy.set_reference_time(
- good_timedelta(self.now, timedelta(seconds=int(new_date.group(1))))
- )
+ def work_ot(self):
+ # I"m not checking for 1h cooldown. Beware of nightshift work, if calling more than once every 60min
+ self.update_job_info()
+ if self.ot_points >= 24 and self.energy.food_fights > 1:
+ r = self._post_economy_work_overtime()
+ if not r.json().get("status") and r.json().get("message") == "money":
+ self.resign_from_employer()
+ self.find_new_job()
+ else:
+ if r.json().get('message') == 'employee':
+ self.find_new_job()
+ self.reporter.report_action("WORK_OT", r.json())
+ elif self.energy.food_fights < 1 and self.ot_points >= 24:
+ self._eat("blue")
+ if self.energy.food_fights < 1:
+ large = max(self.energy.reference_time, self.now)
+ small = min(self.energy.reference_time, self.now)
+ self.write_log("I don't have energy to work OT. Will sleep for {}s".format((large - small).seconds))
+ self.sleep(int((large - small).total_seconds()))
+ self._eat("blue")
+ self.work_ot()
- ugly_js = re.search(r"var erepublik = ({.*}),\s+", html).group(1)
- citizen_js = json.loads(ugly_js)
- citizen = citizen_js.get("citizen", {})
+ def resign_from_employer(self) -> bool:
+ self.update_job_info()
+ if self.r.json().get("isEmployee"):
+ self.reporter.report_action("RESIGN", self.r.json())
+ self._post_economy_resign()
+ return True
+ return False
- self.eday = citizen_js.get("settings").get("eDay")
- self.division = int(citizen.get("division", 0))
+ def buy_tg_contract(self) -> Response:
+ ret = self._post_main_buy_gold_items('gold', "TrainingContract2", 1)
+ self.reporter.report_action("BUY_TG_CONTRACT", ret.json())
+ return ret
- self.energy.interval = citizen.get("energyPerInterval", 0)
- self.energy.limit = citizen.get("energyToRecover", 0)
- self.energy.recovered = citizen.get("energy", 0)
- self.energy.recoverable = citizen.get("energyFromFoodRemaining", 0)
- if self.energy.is_energy_full:
- self.telegram.report_full_energy(self.energy.available, self.energy.limit, self.energy.interval)
+ def find_new_job(self) -> Response:
+ r = self._get_economy_job_market_json(self.details.current_country)
+ jobs = r.json().get("jobs")
+ data = dict(citizen=0, salary=10)
+ for posting in jobs:
+ salary = posting.get("salary")
+ limit = posting.get("salaryLimit", 0)
+ userid = posting.get("citizen").get("id")
- self.details.current_region = citizen.get("regionLocationId", 0)
- self.details.current_country = citizen.get("countryLocationId", 0) # country where citizen is located
- self.details.residence_region = citizen.get("residence", {}).get("regionId", 0)
- self.details.residence_country = citizen.get("residence", {}).get("countryId", 0)
- self.details.citizen_id = citizen.get("citizenId", 0)
- self.details.citizenship = int(citizen.get("country", 0))
- self.details.xp = citizen.get("currentExperiencePoints", 0)
- self.details.daily_task_done = citizen.get("dailyTasksDone", False)
- self.details.daily_task_reward = citizen.get("hasReward", False)
- if citizen.get("dailyOrderDone", False) and not citizen.get("hasDailyOrderReward", False):
- self._post_military_group_missions()
+ if (not limit or salary * 3 < limit) and salary > data["salary"]:
+ data.update({"citizen": userid, "salary": salary})
+ self.reporter.report_action("APPLYING_FOR_JOB", jobs, str(data['citizen']))
+ return self._post_economy_job_market_apply(**data)
- self.details.next_pp.sort()
- for skill in citizen.get("mySkills", {}).values():
- self.details.mayhem_skills.update({int(skill["terrain_id"]): int(skill["skill_points"])})
-
- if citizen.get('party', []):
- party = citizen.get('party')
- self.politics.is_party_member = True
- self.politics.party_id = party.get('party_id')
- self.politics.is_party_president = bool(party.get('is_party_president'))
- self.politics.party_slug = f"{party.get('stripped_title')}-{party.get('party_id')}"
-
- def update_money(self, page: int = 0, currency: int = 62) -> Dict[str, Any]:
- """
- Gets monetary market offers to get exact amount of CC and Gold available
- """
- if currency not in [1, 62]:
- currency = 62
- resp = self._post_economy_exchange_retrieve(False, page, currency)
- resp_data = resp.json()
- self.details.cc = float(resp_data.get("ecash").get("value"))
- self.details.gold = float(resp_data.get("gold").get("value"))
- return resp_data
+ def apply_to_employer(self, employer_id: int, salary: float) -> bool:
+ data = dict(citizen=0, salary=10)
+ self.reporter.report_action("APPLYING_FOR_JOB", data, str(data['citizen']))
+ r = self._post_economy_job_market_apply(**data)
+ return bool(r.json().get('status'))
def update_job_info(self):
ot = self._get_main_job_data().json().get("overTime", {})
if ot:
- self.my_companies.next_ot_time = localize_timestamp(int(ot.get("nextOverTime", 0)))
+ self.next_ot_time = utils.localize_timestamp(int(ot.get("nextOverTime", 0)))
self.ot_points = ot.get("points", 0)
- def update_companies(self):
- html = self._get_economy_my_companies().text
- page_details = json.loads(re.search(r"var pageDetails\s+= ({.*});", html).group(1))
- self.my_companies.work_units = int(page_details.get("total_works", 0))
- have_holdings = re.search(r"var holdingCompanies\s+= ({.*}});", html)
- have_companies = re.search(r"var companies\s+= ({.*}});", html)
- if have_holdings and have_companies:
- self.my_companies.prepare_companies(json.loads(have_companies.group(1)))
- self.my_companies.prepare_holdings(json.loads(have_holdings.group(1)))
- self.my_companies.update_holding_companies()
+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)
+ ret = {}
+ for name, id_ in re.findall(r'', r.text):
+ ret.update({int(id_): name})
+ return ret
+
+ def candidate_for_congress(self, presentation: str = "") -> Response:
+ return self._post_candidate_for_congress(presentation)
+
+ def candidate_for_party_presidency(self) -> Response:
+ return self._get_candidate_party(self.politics.party_slug)
+
+
+class CitizenEconomy(CitizenTravel):
+ food: Dict[str, int] = {"q1": 0, "q2": 0, "q3": 0, "q4": 0, "q5": 0, "q6": 0, "q7": 0, "total": 0}
+ inventory: Dict[str, int] = {"used": 0, "total": 0}
+ boosters: Dict[int, Dict[int, int]] = {100: {}, 50: {}}
+
+ work_units = 0
+ ot_points = 0
+
+ my_companies: MyCompanies = None
+
+ def __init__(self):
+ super().__init__()
+ self.my_companies = MyCompanies()
def update_inventory(self) -> Dict[str, Any]:
"""
@@ -608,30 +825,581 @@ class Citizen(CitizenAPI):
"total": j.get("inventoryStatus").get("totalStorage")})
inventory = dict(items=dict(active=active_items, final=final_items,
raw=raw_materials, offers=offers), status=self.inventory)
- self.food["total"] = sum([self.food[q] * FOOD_ENERGY[q] for q in FOOD_ENERGY])
+ self.food["total"] = sum([self.food[q] * utils.FOOD_ENERGY[q] for q in utils.FOOD_ENERGY])
return inventory
- def update_weekly_challenge(self):
- data = self._get_main_weekly_challenge_data().json()
- self.details.pp = data.get("player", {}).get("prestigePoints", 0)
- self.details.next_pp.clear()
- for reward in data.get("rewards", {}).get("normal", {}):
- status = reward.get("status", "")
- if status == "rewarded":
+ def work_employees(self) -> bool:
+ self.update_companies()
+ ret = True
+ work_units_needed = 0
+ employee_companies = self.my_companies.get_employable_factories()
+ for c_id, preset_count in employee_companies.items():
+ work_units_needed += preset_count
+
+ if work_units_needed:
+ if work_units_needed <= self.my_companies.work_units:
+ self._do_wam_and_employee_work(employee_companies=employee_companies)
+ self.update_companies()
+ if self.my_companies.get_employable_factories():
+ ret = False
+ else:
+ ret = True
+
+ return ret
+
+ def work_wam(self) -> bool:
+ self.update_citizen_info()
+ self.update_companies()
+ # Prevent messing up levelup with wam
+ if not (self.is_levelup_close and self.config.fight) or self.config.force_wam:
+ # Check for current region
+ 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_id"]: holding_id})
+
+ if self.details.current_region in regions:
+ self._do_wam_and_employee_work(regions.pop(self.details.current_region, None))
+
+ for holding_id in regions.values():
+ self._do_wam_and_employee_work(holding_id)
+
+ self.travel_to_residence()
+ else:
+ self.write_log("Did not wam because I would mess up levelup!")
+
+ self.update_companies()
+ return not self.my_companies.get_total_wam_count()
+
+ def _do_wam_and_employee_work(self, wam_holding_id: int = 0, employee_companies: dict = None) -> bool:
+ self.update_citizen_info()
+ if employee_companies is None:
+ employee_companies = {}
+ 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
+
+ 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:
+ inv_w = len(str(self.inventory["total"]))
+ self.write_log(
+ "Inv: {:{inv_w}}/{:{inv_w}} ({:4.2f}), Energy: {}/{} + {} (+{}hp/6min) WAM count {:3}".format(
+ self.inventory["used"], self.inventory["total"], extra_needed,
+ self.energy.recovered, self.energy.limit, self.energy.recoverable, self.energy.interval,
+ len(wam_list), inv_w=inv_w
+ ))
+ wam_list.pop(-1)
+
+ if wam_list or employee_companies:
+ 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_id']:
+ if not self.travel_to_region(wam_holding['region_id']):
+ return False
+ response = self._post_economy_work("production", wam=wam_list, employ=employee_companies).json()
+ if response.get("status"):
+ self.reporter.report_action("WORK_WAM_EMPLOYEES", response)
+ if self.config.auto_sell:
+ for kind, data in response.get("result", {}).get("production", {}).items():
+ if kind in self.config.auto_sell and data:
+ if kind in ["food", "weapon", "house", "airplane"]:
+ for quality, amount in data.items():
+ self.sell_produced_product(kind, quality)
+
+ elif kind.endswith("Raw"):
+ self.sell_produced_product(kind, 1)
+
+ else:
+ raise ErepublikException("Unknown kind produced '{kind}'".format(kind=kind))
+ elif self.config.auto_buy_raw and re.search(r"not_enough_[^_]*_raw", response.get("message")):
+ raw_kind = re.search(r"not_enough_(\w+)_raw", response.get("message"))
+ if raw_kind:
+ raw_kind = raw_kind.group(1)
+ result = response.get("result", {})
+ amount_remaining = round(result.get("consume") + 0.49) - round(result.get("stock") - 0.49)
+ industry = "{}Raw".format(raw_kind)
+ while amount_remaining > 0:
+ amount = amount_remaining
+ best_offer = self.get_market_offers(self.details.citizenship, industry, 1)
+ amount = best_offer['amount'] if amount >= best_offer['amount'] else amount
+ rj = self.buy_from_market(amount=best_offer['amount'], offer=best_offer['offer_id'])
+ if not rj.get('error'):
+ amount_remaining -= amount
+ else:
+ self.write_log(rj.get('message', ""))
+ break
+ else:
+ return self._do_wam_and_employee_work(wam_holding_id, employee_companies)
+ elif response.get("message") == "not_enough_health_food":
+ self.buy_food()
+ return self._do_wam_and_employee_work(wam_holding_id, employee_companies)
+ else:
+ msg = "I was not able to wam and or employ because:\n{}".format(response)
+ self.reporter.report_action("WORK_WAM_EMPLOYEES", response, msg)
+ self.write_log(msg)
+ wam_count = self.my_companies.get_total_wam_count()
+ if wam_count:
+ self.write_log("Wam ff lockdown is now {}, was {}".format(wam_count, self.my_companies.ff_lockdown))
+ self.my_companies.ff_lockdown = wam_count
+ return bool(wam_count)
+
+ def update_money(self, page: int = 0, currency: int = 62) -> Dict[str, Any]:
+ """
+ Gets monetary market offers to get exact amount of CC and Gold available
+ """
+ if currency not in [1, 62]:
+ currency = 62
+ resp = self._post_economy_exchange_retrieve(False, page, currency)
+ resp_data = resp.json()
+ self.details.cc = float(resp_data.get("ecash").get("value"))
+ self.details.gold = float(resp_data.get("gold").get("value"))
+ return resp_data
+
+ def update_companies(self):
+ html = self._get_economy_my_companies().text
+ page_details = utils.json.loads(re.search(r"var pageDetails\s+= ({.*});", html).group(1))
+ self.my_companies.work_units = int(page_details.get("total_works", 0))
+
+ have_holdings = re.search(r"var holdingCompanies\s+= ({.*}});", html)
+ have_companies = re.search(r"var companies\s+= ({.*}});", html)
+ if have_holdings and have_companies:
+ self.my_companies.prepare_companies(utils.json.loads(have_companies.group(1)))
+ self.my_companies.prepare_holdings(utils.json.loads(have_holdings.group(1)))
+ self.my_companies.update_holding_companies()
+
+ def check_house_durability(self) -> Dict[int, datetime]:
+ ret = {}
+ inv = self.update_inventory()
+ for house_quality, active_house in inv['items']['active'].get('house', {}).items():
+ till = utils.good_timedelta(self.now, timedelta(seconds=active_house['time_left']))
+ ret.update({house_quality: till})
+ return ret
+
+ def buy_and_activate_house(self, q: int) -> Dict[int, datetime]:
+ inventory = self.update_inventory()
+ ok_to_activate = False
+ if not inventory['items']['final'].get('house', {}).get(q, {}):
+ offers = []
+ countries = [self.details.citizenship, ]
+ if self.details.current_country != self.details.citizenship:
+ countries.append(self.details.current_country)
+ for country in countries:
+ offers += [self.get_market_offers(country, "house", q)]
+ global_cheapest = self.get_market_offers(product_name="house", quality=q)
+ cheapest_offer = sorted(offers, key=lambda o: o["price"])[0]
+ region = self.get_country_travel_region(global_cheapest['country'])
+ if global_cheapest['price'] + 200 < cheapest_offer['price'] and region:
+ self._travel(global_cheapest['country'], region)
+ buy = self.buy_from_market(global_cheapest['offer_id'], 1)
+ else:
+ buy = self.buy_from_market(cheapest_offer['offer_id'], 1)
+ if buy["error"]:
+ msg = f"Unable to buy q{q} house! \n{buy['message']}"
+ self.write_log(msg)
+ else:
+ ok_to_activate = True
+ else:
+ ok_to_activate = True
+ if ok_to_activate:
+ self.activate_house(q)
+ return self.check_house_durability()
+
+ def renew_houses(self, forced: bool = False) -> Dict[int, datetime]:
+ """
+ Renew all houses which endtime is in next 48h
+ :param forced: if true - renew all houses
+ :return:
+ """
+ house_durability = self.check_house_durability()
+ for q, active_till in house_durability.items():
+ if utils.good_timedelta(active_till, - timedelta(hours=48)) <= self.now or forced:
+ house_durability = self.buy_and_activate_house(q)
+ return house_durability
+
+ def activate_house(self, quality: int) -> datetime:
+ active_until = self.now
+ r = self._post_economy_activate_house(quality)
+ if r.json().get("status") and not r.json().get("error"):
+ house = r.json()["inventoryItems"]["activeEnhancements"]["items"]["4_%i_active" % quality]
+ active_until = utils.good_timedelta(active_until, timedelta(seconds=house["active"]["time_left"]))
+ return active_until
+
+ def get_game_token_offers(self):
+ r = self._post_economy_game_tokens_market('retrieve').json()
+ return {v.get('id'): dict(amount=v.get('amount'), price=v.get('price')) for v in r.get("topOffers")}
+
+ def fetch_organisation_account(self, org_id: int):
+ r = self._get_economy_citizen_accounts(org_id)
+ table = re.search(r'()', r.text, re.I | re.M | re.S)
+ if table:
+ account = re.findall(r'>\s*(\d+.\d+)\s*', table.group(1))
+ if account:
+ return {"gold": account[0], "cc": account[1], 'ok': True}
+
+ return {"gold": 0, "cc": 0, 'ok': False}
+
+ def accept_money_donations(self):
+ for notification in self._get_main_notifications_ajax_system():
+ don_id = re.search(r"erepublik.functions.acceptRejectDonation\(\"accept\", (\d+)\)", notification)
+ if don_id:
+ self._get_main_money_donation_accept(int(don_id.group(1)))
+ self.sleep(5)
+
+ def reject_money_donations(self) -> int:
+ r = self._get_main_notifications_ajax_system()
+ count = 0
+ donation_ids = re.findall(r"erepublik.functions.acceptRejectDonation\(\"reject\", (\d+)\)", r.text)
+ while donation_ids:
+ for don_id in donation_ids:
+ self._get_main_money_donation_reject(int(don_id))
+ count += 1
+ self.sleep(5)
+ r = self._get_main_notifications_ajax_system()
+ donation_ids = re.findall(r"erepublik.functions.acceptRejectDonation\(\"reject\", (\d+)\)", r.text)
+ return count
+
+ def get_my_market_offers(self) -> List[Dict[str, Union[int, float, str]]]:
+ ret = []
+ for offer in self._get_economy_my_market_offers().json():
+ line = offer.copy()
+ line.pop('icon', None)
+ ret.append(line)
+ return ret
+
+ def post_market_offer(self, industry: int, quality: int, amount: int, price: float) -> Response:
+ if industry not in self.available_industries.values():
+ self.write_log(f"Trying to sell unsupported industry {industry}")
+
+ data = {
+ "country": self.details.citizenship,
+ "industry": industry,
+ "quality": quality,
+ "amount": amount,
+ "price": price,
+ "buy": False,
+ }
+ ret = self._post_economy_marketplace_actions(**data)
+ self.reporter.report_action("SELL_PRODUCT", ret.json())
+ return ret
+
+ def buy_from_market(self, offer: int, amount: int) -> dict:
+ ret = self._post_economy_marketplace_actions(amount, True, offer=offer)
+ json_ret = ret.json()
+ if json_ret.get('error'):
+ return json_ret
+ else:
+ self.details.cc = ret.json()['currency']
+ self.details.gold = ret.json()['gold']
+ r_json = ret.json()
+ r_json.pop("offerUpdate", None)
+ self.reporter.report_action("BUY_PRODUCT", ret.json())
+ return json_ret
+
+ def assign_factory_to_holding(self, factory_id: int, holding_id: int) -> Response:
+ """
+ Assigns factory to new holding
+ """
+ company = self.my_companies.companies[factory_id]
+ company_name = self.factories[company['industry_id']]
+ if not company['is_raw']:
+ company_name += f" q{company['quality']}"
+ self.write_log(f"{company_name} 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)
+
+ def create_factory(self, industry_id: int, building_type: int = 1) -> Response:
+ """
+ param industry_ids: FRM={q1:7, q2:8, q3:9, q4:10, q5:11} WRM={q1:12, q2:13, q3:14, q4:15, q5:16}
+ HRM={q1:18, q2:19, q3:20, q4:21, q5:22} ARM={q1:24, q2:25, q3:26, q4:27, q5:28}
+ Factories={Food:1, Weapons:2, House:4, Aircraft:23} <- Building_type 1
+
+ Storage={1000: 1, 2000: 2} <- Building_type 2
+ """
+ company_name = self.factories[industry_id]
+ if building_type == 2:
+ company_name = f"Storage"
+ 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]
+ company_name = self.factories[company['industry_id']]
+ if not company['is_raw']:
+ company_name += f" q{company['quality']}"
+ self.write_log(f"{company_name} dissolved!")
+ return self._post_economy_sell_company(factory_id, self.details.pin, sell=False)
+
+ def get_industry_id(self, industry_name: str) -> int:
+ """Returns industry id
+
+ :type industry_name: str
+ :return: int
+ """
+ return self.available_industries.get(industry_name, 0)
+
+ def get_industry_name(self, industry_id: int) -> str:
+ """Returns industry name from industry ID
+
+ :type industry_id: int
+ :return: industry name
+ :rtype: str
+ """
+ for iname, iid in self.available_industries.items():
+ if iid == industry_id:
+ return iname
+ return ""
+
+ def get_market_offers(self, country_id: int = None, product_name: str = None, quality: int = None) -> dict:
+ raw_short_names = dict(frm="foodRaw", wrm="weaponRaw", hrm="houseRaw", arm="airplaneRaw")
+ q1_industries = ["aircraft"] + list(raw_short_names.values())
+ if product_name:
+ if product_name not in self.available_industries and product_name not in raw_short_names:
+ self.write_log(f"Industry '{product_name}' not implemented")
+ raise ErepublikException(f"Industry '{product_name}' not implemented")
+ elif product_name in raw_short_names:
+ quality = 1
+ product_name = raw_short_names[product_name]
+ product_name = [product_name]
+ elif quality:
+ raise ErepublikException("Quality without product not allowed")
+
+ item_data = dict(price=999999., country=0, amount=0, offer_id=0, citizen_id=0)
+
+ items = {"food": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
+ q5=item_data.copy(), q6=item_data.copy(), q7=item_data.copy()),
+ "weapon": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
+ q5=item_data.copy(), q6=item_data.copy(), q7=item_data.copy()),
+ "house": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
+ q5=item_data.copy()), "aircraft": dict(q1=item_data.copy()),
+ "foodRaw": dict(q1=item_data.copy()), "weaponRaw": dict(q1=item_data.copy()),
+ "houseRaw": dict(q1=item_data.copy()), "airplaneRaw": dict(q1=item_data.copy())}
+
+ if country_id:
+ countries = [country_id]
+ else:
+ countries = self.get_travel_countries()
+
+ start_dt = self.now
+ iterable = [countries, product_name or items, [quality] if quality else range(1, 8)]
+ for country, industry, q in product(*iterable):
+ if (q > 1 and industry in q1_industries) or (q > 5 and industry == "house"):
continue
- elif status == "completed":
- self._post_main_weekly_challenge_reward(reward.get("id", 0))
- elif reward.get("icon", "") == "energy_booster":
- pps = re.search(r"Reach (\d+) Prestige Points to unlock the following reward: \+1 Energy",
- reward.get("tooltip", ""))
- if pps:
- self.details.next_pp.append(int(pps.group(1)))
+
+ r = self._post_economy_marketplace(country, self.available_industries[industry], q).json()
+ obj = items[industry][f"q{q}"]
+ if not r.get("error", False):
+ for offer in r["offers"]:
+ if obj["price"] > float(offer["priceWithTaxes"]):
+ obj["price"] = float(offer["priceWithTaxes"])
+ obj["country"] = int(offer["country_id"])
+ obj["amount"] = int(offer["amount"])
+ obj["offer_id"] = int(offer["id"])
+ obj["citizen_id"] = int(offer["citizen_id"])
+ elif obj["price"] == float(offer["priceWithTaxes"]) and obj["amount"] < int(offer["amount"]):
+ obj["country"] = int(offer["country_id"])
+ obj["amount"] = int(offer["amount"])
+ obj["offer_id"] = int(offer["id"])
+ self.write_log(f"Scraped market in {self.now - start_dt}!")
+
+ if quality:
+ ret = items[product_name[0]]["q%i" % quality]
+ elif product_name:
+ if product_name[0] in raw_short_names.values():
+ ret = items[product_name[0]]["q1"]
+ else:
+ ret = items[product_name[0]]
+ else:
+ ret = items
+ return ret
+
+ def buy_food(self):
+ hp_per_quality = {"q1": 2, "q2": 4, "q3": 6, "q4": 8, "q5": 10, "q6": 12, "q7": 20}
+ hp_needed = 48 * self.energy.interval * 10 - self.food["total"]
+ local_offers = self.get_market_offers(country_id=self.details.current_country, product_name="food")
+
+ cheapest_q, cheapest = sorted(local_offers.items(), key=lambda v: v[1]["price"] / hp_per_quality[v[0]])[0]
+
+ if cheapest["amount"] * hp_per_quality[cheapest_q] < hp_needed:
+ amount = cheapest["amount"]
+ else:
+ amount = hp_needed // hp_per_quality[cheapest_q]
+
+ if amount * cheapest["price"] < self.details.cc:
+ data = dict(offer=cheapest["offer_id"], amount=amount, price=cheapest["price"],
+ cost=amount * cheapest["price"], quality=cheapest_q, energy=amount * hp_per_quality[cheapest_q])
+ self.reporter.report_action("BUY_FOOD", json_val=data)
+ self.buy_from_market(cheapest["offer_id"], amount)
+ self.update_inventory()
+ else:
+ s = f"Don't have enough money! Needed: {amount * cheapest['price']}cc, Have: {self.details.cc}cc"
+ self.write_log(s)
+ self.reporter.report_action("BUY_FOOD", value=s)
+
+ def get_monetary_offers(self, currency: int = 62) -> List[Dict[str, Union[int, float]]]:
+ if currency not in [1, 62]:
+ currency = 62
+ resp = self._post_economy_exchange_retrieve(False, 0, currency).json()
+ ret = []
+ offers = re.findall(r"id='purchase_(\d+)' data-i18n='Buy for' data-currency='GOLD' "
+ r"data-price='(\d+\.\d+)' data-max='(\d+\.\d+)' trigger='purchase'",
+ resp["buy_mode"], re.M | re.I | re.S)
+
+ for offer_id, price, amount in offers:
+ ret.append({"offer_id": int(offer_id), "price": float(price), "amount": float(amount)})
+
+ return sorted(ret, key=lambda o: (o["price"], -o["amount"]))
+
+ def buy_monetary_market_offer(self, offer: int, amount: float, currency: int) -> bool:
+ response = self._post_economy_exchange_purchase(amount, currency, offer)
+ self.details.cc = float(response.json().get("ecash").get("value"))
+ self.details.gold = float(response.json().get("gold").get("value"))
+ self.reporter.report_action("BUY_GOLD", json_val=response.json(),
+ value=f"New amount {self.details.cc}cc, {self.details.gold}g")
+ return not response.json().get("error", False)
+
+ def donate_money(self, citizen_id: int = 1620414, amount: float = 0.0, currency: int = 62) -> bool:
+ """ currency: gold = 62, cc = 1 """
+ resp = self._post_economy_donate_money_action(citizen_id, amount, currency)
+ r = re.search('You do not have enough money in your account to make this donation', resp.text)
+ return not bool(r)
+
+ def donate_items(self, citizen_id: int = 1620414, amount: int = 0, industry_id: int = 1, quality: int = 1) -> int:
+ if amount < 1:
+ return 0
+ ind = {v: k for k, v in self.available_industries.items()}
+ self.write_log(f"Donate: {amount:4d}q{quality} {ind[industry_id]} to {citizen_id}")
+ response = self._post_economy_donate_items_action(citizen_id, amount, industry_id, quality)
+ if re.search(rf"Successfully transferred {amount} item\(s\) to", response.text):
+ return amount
+ else:
+ if re.search(r"You do not have enough items in your inventory to make this donation", response.text):
+ return 0
+ available = re.search(rf"Cannot transfer the items because the user has only (\d+) free slots in (his|her) "
+ rf"storage.", response.text).group(1)
+ return self.donate_items(citizen_id, int(available), industry_id, quality)
+
+ def sell_produced_product(self, kind: str, quality: int = 1, amount: int = 0):
+ if not amount:
+ inv_resp = self._get_economy_inventory_items().json()
+ category = "rawMaterials" if kind.endswith("Raw") else "finalProducts"
+ item = "{}_{}".format(self.available_industries[kind], quality)
+ amount = inv_resp.get("inventoryItems").get(category).get("items").get(item).get("amount", 0)
+
+ if amount >= 1:
+ lowest_price = self.get_market_offers(country_id=self.details.citizenship,
+ product_name=kind, quality=int(quality))
+
+ if lowest_price["citizen_id"] == self.details.citizen_id:
+ price = lowest_price["price"]
+ else:
+ price = lowest_price["price"] - 0.01
+
+ self.post_market_offer(industry=self.available_industries[kind], amount=int(amount),
+ quality=int(quality), price=price)
+
+ def contribute_cc_to_country(self, amount=0., country_id: int = 71) -> 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)
+ self.reporter.report_action("CONTRIBUTE_CC", data, str(amount))
+ r = self._post_main_country_donate(**data)
+ return r.json().get('status') or not r.json().get('error')
+
+ def contribute_food_to_country(self, amount: int = 0, quality: int = 1, country_id: int = 71) -> 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)
+ self.reporter.report_action("CONTRIBUTE_FOOD", data, utils.FOOD_ENERGY[quality] * amount)
+ r = self._post_main_country_donate(**data)
+ return r.json().get('status') or not r.json().get('error')
+
+ def contribute_gold_to_country(self, amount: int, country_id: int = 71) -> bool:
+ self.update_money()
+
+ if self.details.cc < amount:
+ return False
+ 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)
+ return r.json().get('status') or not r.json().get('error')
+
+
+class CitizenMedia(BaseCitizen):
+ def endorse_article(self, article_id: int, amount: int) -> bool:
+ if amount in (5, 50, 100):
+ resp = self._post_main_donate_article(article_id, amount).json()
+ return not bool(resp.get('error'))
+ else:
+ return False
+
+ def vote_article(self, article_id: int) -> bool:
+ resp = self._post_main_vote_article(article_id).json()
+ return not bool(resp.get('error'))
+
+ def get_article_comments(self, article_id: int, page_id: int = 1) -> Response:
+ return self._post_main_article_comments(article_id, page_id)
+
+ def comment_article(self, article_id: int = 2645676, msg: str = None) -> Response:
+ if msg is None:
+ msg = self.eday
+ r = self.get_article_comments(article_id, 2)
+ r = self.get_article_comments(article_id, r.json()["pages"])
+ comments = r.json()["comments"]
+ if not comments[max(comments.keys())]["isMyComment"]:
+ r = self.write_article_comment(msg, article_id)
+ return r
+
+ def write_article_comment(self, message: str, article_id: int, parent_id: int = None) -> Response:
+ return self._post_main_article_comments_create(message, article_id, parent_id)
+
+ def publish_article(self, title: str, content: str, kind: int) -> Response:
+ kinds = {1: "First steps in eRepublik", 2: "Battle orders", 3: "Warfare analysis",
+ 4: "Political debates and analysis", 5: "Financial business",
+ 6: "Social interactions and entertainment"}
+ if kind in kinds:
+ return self._post_main_write_article(title, content, self.details.citizenship, kind)
+ else:
+ raise ErepublikException("Article kind must be one of:\n{}\n'{}' is not supported".format(
+ "\n".join(["{}: {}".format(k, v) for k, v in kinds.items()]), kind
+ ))
+
+
+class CitizenMilitary(CitizenTravel, CitizenTasks):
+ all_battles: Dict[int, Battle] = None
+ countries: Dict[int, Dict[str, Union[str, List[int]]]] = None
+ __last_war_update_data = None
+
+ active_fs: bool = False
+ boosters: Dict[int, Dict[int, int]] = {100: {}, 50: {}}
def update_war_info(self):
if not self.details.current_country:
self.update_citizen_info()
- if self.__last_war_update_data and self.__last_war_update_data.get('last_updated', 0) + 30 > self.now.timestamp():
+ if self.__last_war_update_data and self.__last_war_update_data.get('last_updated',
+ 0) + 30 > self.now.timestamp():
resp_json = self.__last_war_update_data
else:
resp_json = self._get_military_campaigns_json_list().json()
@@ -658,64 +1426,55 @@ class Citizen(CitizenAPI):
for battle_data in resp_json.get("battles", {}).values():
self.all_battles[battle_data.get('id')] = Battle(battle_data)
- def eat(self):
- """
- Try to eat food
- """
- if self.food["total"] > self.energy.interval:
- if self.energy.limit - self.energy.recovered > self.energy.interval or not self.energy.recoverable % 2:
- self._eat("blue")
- else:
- self.write_log("I don't want to eat right now!")
+ def get_battle_for_war(self, war_id: int) -> Optional[Battle]:
+ self.update_war_info()
+ war_info = self.get_war_status(war_id)
+ return self.all_battles.get(war_info.get("battle_id"), None)
+
+ def get_war_status(self, war_id: int) -> Dict[str, Union[bool, Dict[int, str]]]:
+ r = self._get_wars_show(war_id)
+ html = r.text
+ ret = {}
+ reg_re = re.compile(fr'data-war-id="{war_id}" data-region-id="(\d+)" data-region-name="([- \w]+)"')
+ if reg_re.findall(html):
+ ret.update(regions={}, can_attack=True)
+ for reg in reg_re.findall(html):
+ ret["regions"].update({int(reg[0]): reg[1]})
+ elif re.search(r'Join', html):
+ battle_id = re.search(r'Join', html).group(1)
+ ret.update(can_attack=False, battle_id=int(battle_id))
+ elif re.search(r'This war is no longer active.', html):
+ ret.update(can_attack=False, ended=True)
else:
- self.write_log(f"I'm out of food! But I'll try to buy some!\n{self.food}")
- self.buy_food()
- if self.food["total"] > self.energy.interval:
- self.eat()
- else:
- self.write_log("I failed to buy food")
- self.write_log(self.health_info)
-
- def eat_ebs(self):
- self.write_log("Eating energy bar")
- if self.energy.recoverable:
- self._eat("blue")
- self._eat("orange")
- self.write_log(self.health_info)
-
- def _eat(self, colour: str = "blue") -> Response:
- response = self._post_eat(colour)
- 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])))
- )
- self.energy.recovered = r_json.get("health")
- self.energy.recoverable = r_json.get("food_remaining")
- for q, amount in r_json.get("units_consumed").items():
- if f"q{q}" in self.food:
- self.food[f"q{q}"] -= amount
- elif q == "10":
- self.eb_normal -= amount
- elif q == "11":
- self.eb_double -= amount
- elif q == "12":
- self.eb_small -= amount
- return response
-
- @property
- def health_info(self):
- ret = f"{self.energy.recovered}/{self.energy.limit} + {self.energy.recoverable}, " \
- f"{self.energy.interval}hp/6m. {self.details.xp_till_level_up}xp until level up"
+ ret.update(can_attack=False)
return ret
- @property
- def now(self) -> datetime:
- """
- Returns aware datetime object localized to US/Pacific (eRepublik time)
- :return: datetime
- """
- return now()
+ 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) -> int:
+ battle = self.all_battles.get(battle_id)
+ 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()
+ weapon_quality = -1
+ weapon_damage = 0
+ if not battle.is_air:
+ for weapon in available_weapons:
+ try:
+ if weapon['weaponQuantity'] > 30 and weapon['damage'] > weapon_damage:
+ weapon_quality = int(weapon['weaponId'])
+ except ValueError:
+ pass
+ return self.change_weapon(battle_id, weapon_quality)
+
+ def change_weapon(self, battle_id: int, weapon_quality: int) -> int:
+ battle = self.all_battles.get(battle_id)
+ battle_zone = battle.div[11 if battle.is_air else self.division].battle_zone_id
+ r = self._post_military_change_weapon(battle_id, battle_zone, weapon_quality)
+ return r.json().get('weaponInfluence')
def check_epic_battles(self):
active_fs = False
@@ -872,7 +1631,7 @@ class Citizen(CitizenAPI):
continue
if battle.start > self.now:
- self.sleep(get_sleep_seconds(battle.start))
+ self.sleep(utils.get_sleep_seconds(battle.start))
if travel_needed:
if battle.is_rw:
@@ -889,7 +1648,6 @@ class Citizen(CitizenAPI):
self.set_default_weapon(battle_id)
self.fight(battle_id, side_id)
self.travel_to_residence()
- self.collect_weekly_reward()
break
def fight(self, battle_id: int, side_id: int = None, count: int = None) -> int:
@@ -1023,474 +1781,31 @@ class Citizen(CitizenAPI):
battle = self.all_battles.get(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
- self.update_job_info()
- if self.ot_points >= 24 and self.energy.food_fights > 1:
- r = self._post_economy_work_overtime()
- if not r.json().get("status") and r.json().get("message") == "money":
- self.resign()
- self.find_new_job()
- else:
- if r.json().get('message') == 'employee':
- self.find_new_job()
- self.reporter.report_action("WORK_OT", r.json())
- elif self.energy.food_fights < 1 and self.ot_points >= 24:
- self._eat("blue")
- if self.energy.food_fights < 1:
- large = max(self.energy.reference_time, self.now)
- small = min(self.energy.reference_time, self.now)
- self.write_log("I don't have energy to work OT. Will sleep for {}s".format((large - small).seconds))
- self.sleep(int((large - small).total_seconds()))
- self._eat("blue")
- self.work_ot()
+ 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) -> float:
+ if not rang or strength or elite is None:
+ r = self._get_main_citizen_profile_json(self.details.citizen_id).json()
+ if not rang:
+ rang = r['military']['militaryData']['ground']['rankNumber']
+ if not strength:
+ strength = r['military']['militaryData']['ground']['strength']
+ if elite is None:
+ elite = r['citizenAttributes']['level'] > 100
+ if ne:
+ tp = True
- def work(self):
- if self.energy.food_fights >= 1:
- response = self._post_economy_work("work")
- js = response.json()
- good_msg = ["already_worked", "captcha"]
- if not js.get("status") and not js.get("message") in good_msg:
- if js.get('message') == 'employee':
- self.find_new_job()
- self.update_citizen_info()
- self.work()
- else:
- self.reporter.report_action("WORK", json_val=js)
- else:
- self._eat("blue")
- if self.energy.food_fights < 1:
- seconds = (self.energy.reference_time - self.now).total_seconds()
- self.write_log("I don't have energy to work. Will sleep for {}s".format(seconds))
- self.sleep(seconds)
- self._eat("blue")
- self.work()
+ return utils.calculate_hit(strength, rang, tp, elite, ne, 50 if booster_50 else 100 if booster_100 else 0)
- def train(self):
- r = self._get_main_training_grounds_json()
- tg_json = r.json()
- self.details.gold = tg_json["page_details"]["gold"]
- self.tg_contract = {"free_train": tg_json["hasFreeTrain"]}
- if tg_json["contracts"]:
- self.tg_contract.update(**tg_json["contracts"][0])
+ def get_air_hit_dmg_value(self, rang: int = None, elite: bool = None, ne: bool = False,
+ weapon: bool = False) -> float:
+ if not rang or elite is None:
+ r = self._get_main_citizen_profile_json(self.details.citizen_id).json()
+ if not rang:
+ rang = r['military']['militaryData']['air']['rankNumber']
+ if elite is None:
+ elite = r['citizenAttributes']['level'] > 100
- tgs = []
- for data in sorted(tg_json["grounds"], key=lambda k: k["cost"]):
- if data["default"] and not data["trained"]:
- tgs.append(data["id"])
- if tgs:
- if self.energy.food_fights >= len(tgs):
- response = self._post_economy_train(tgs)
- if not response.json().get("status"):
- self.update_citizen_info()
- self.train()
- else:
- self.reporter.report_action("TRAIN", response.json())
- else:
- self._eat("blue")
- if self.energy.food_fights < len(tgs):
- large = max(self.energy.reference_time, self.now)
- small = min(self.energy.reference_time, self.now)
- self.write_log("I don't have energy to train. Will sleep for {} seconds".format(
- (large - small).seconds))
- self.sleep(int((large - small).total_seconds()))
- self._eat("blue")
- self.train()
-
- def work_employees(self) -> bool:
- self.update_companies()
- ret = True
- work_units_needed = 0
- employee_companies = self.my_companies.get_employable_factories()
- for c_id, preset_count in employee_companies.items():
- work_units_needed += preset_count
-
- if work_units_needed:
- if work_units_needed <= self.my_companies.work_units:
- self._do_wam_and_employee_work(employee_companies=employee_companies)
- self.update_companies()
- if self.my_companies.get_employable_factories():
- ret = False
- else:
- ret = True
-
- return ret
-
- def work_wam(self) -> bool:
- self.update_citizen_info()
- self.update_companies()
- # Prevent messing up levelup with wam
- if not (self.is_levelup_close and self.config.fight) or self.config.force_wam:
- # Check for current region
- 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_id"]: holding_id})
-
- if self.details.current_region in regions:
- self._do_wam_and_employee_work(regions.pop(self.details.current_region, None))
-
- for holding_id in regions.values():
- self._do_wam_and_employee_work(holding_id)
-
- self.travel_to_residence()
- else:
- self.write_log("Did not wam because I would mess up levelup!")
-
- self.update_companies()
- return not self.my_companies.get_total_wam_count()
-
- def _do_wam_and_employee_work(self, wam_holding_id: int = 0, employee_companies: dict = None) -> bool:
- self.update_citizen_info()
- if employee_companies is None:
- employee_companies = {}
- 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
-
- 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:
- inv_w = len(str(self.inventory["total"]))
- self.write_log(
- "Inv: {:{inv_w}}/{:{inv_w}} ({:4.2f}), Energy: {}/{} + {} (+{}hp/6min) WAM count {:3}".format(
- self.inventory["used"], self.inventory["total"], extra_needed,
- self.energy.recovered, self.energy.limit, self.energy.recoverable, self.energy.interval,
- len(wam_list), inv_w=inv_w
- ))
- wam_list.pop(-1)
-
- if wam_list or employee_companies:
- 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_id']:
- if not self.travel_to_region(wam_holding['region_id']):
- return False
- response = self._post_economy_work("production", wam=wam_list, employ=employee_companies).json()
- if response.get("status"):
- self.reporter.report_action("WORK_WAM_EMPLOYEES", response)
- if self.config.auto_sell:
- for kind, data in response.get("result", {}).get("production", {}).items():
- if kind in self.config.auto_sell and data:
- if kind in ["food", "weapon", "house", "airplane"]:
- for quality, amount in data.items():
- self.sell_produced_product(kind, quality)
-
- elif kind.endswith("Raw"):
- self.sell_produced_product(kind, 1)
-
- else:
- raise ErepublikException("Unknown kind produced '{kind}'".format(kind=kind))
- elif self.config.auto_buy_raw and re.search(r"not_enough_[^_]*_raw", response.get("message")):
- raw_kind = re.search(r"not_enough_(\w+)_raw", response.get("message"))
- if raw_kind:
- raw_kind = raw_kind.group(1)
- result = response.get("result", {})
- amount_remaining = round(result.get("consume") + 0.49) - round(result.get("stock") - 0.49)
- industry = "{}Raw".format(raw_kind)
- while amount_remaining > 0:
- amount = amount_remaining
- best_offer = self.get_market_offers(self.details.citizenship, industry, 1)
- amount = best_offer['amount'] if amount >= best_offer['amount'] else amount
- rj = self.buy_from_market(amount=best_offer['amount'], offer=best_offer['offer_id'])
- if not rj.get('error'):
- amount_remaining -= amount
- else:
- self.write_log(rj.get('message', ""))
- break
- else:
- return self._do_wam_and_employee_work(wam_holding_id, employee_companies)
- elif response.get("message") == "not_enough_health_food":
- self.buy_food()
- return self._do_wam_and_employee_work(wam_holding_id, employee_companies)
- else:
- msg = "I was not able to wam and or employ because:\n{}".format(response)
- self.reporter.report_action("WORK_WAM_EMPLOYEES", response, msg)
- self.write_log(msg)
- wam_count = self.my_companies.get_total_wam_count()
- if wam_count:
- self.write_log("Wam ff lockdown is now {}, was {}".format(wam_count, self.my_companies.ff_lockdown))
- self.my_companies.ff_lockdown = wam_count
- return bool(wam_count)
-
- def sell_produced_product(self, kind: str, quality: int = 1, amount: int = 0):
- if not amount:
- inv_resp = self._get_economy_inventory_items().json()
- category = "rawMaterials" if kind.endswith("Raw") else "finalProducts"
- item = "{}_{}".format(self.available_industries[kind], quality)
- amount = inv_resp.get("inventoryItems").get(category).get("items").get(item).get("amount", 0)
-
- if amount >= 1:
- lowest_price = self.get_market_offers(country_id=self.details.citizenship,
- product_name=kind, quality=int(quality))
-
- if lowest_price["citizen_id"] == self.details.citizen_id:
- price = lowest_price["price"]
- else:
- price = lowest_price["price"] - 0.01
-
- self.post_market_offer(industry=self.available_industries[kind], amount=int(amount),
- quality=int(quality), price=price)
-
- def get_country_travel_region(self, country_id: int) -> int:
- regions = self.get_travel_regions(country_id=country_id)
- regs = []
- if regions:
- for region in regions.values():
- 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]
- else:
- return 0
-
- def _update_citizen_location(self, country_id: int, region_id: int):
- self.details.current_region = region_id
- self.details.current_country = country_id
-
- def travel_to_residence(self) -> bool:
- self.update_citizen_info()
- res_r = self.details.residence_region
- if self.details.residence_country and res_r and not res_r == self.details.current_region:
- r = self._travel(self.details.residence_country, self.details.residence_region)
- if r.json().get('message', '') == 'success':
- self._update_citizen_location(self.details.residence_country, self.details.current_region)
- return True
- return False
- return True
-
- def travel_to_region(self, region_id: int) -> bool:
- data = self._post_main_travel_data(region_id=region_id).json()
- if data.get('alreadyInRegion'):
- return True
- else:
- r = self._travel(data.get('preselectCountryId'), region_id).json()
- if r.get('message', '') == 'success':
- self._update_citizen_location(data.get('preselectCountryId'), region_id)
- 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()
-
- regs = []
- if data.get('regions'):
- for region in data.get('regions').values():
- if region['countryId'] == country_id: # Is not occupied by other country
- regs.append((region['id'], region['distanceInKm']))
- if regs:
- region_id = min(regs, key=lambda _: int(_[1]))[0]
- r = self._travel(country_id, region_id).json()
- if r.get('message', '') == 'success':
- self._update_citizen_location(country_id, region_id)
- return True
- return False
-
- def travel_to_holding(self, holding_id: int) -> bool:
- data = self._post_main_travel_data(holdingId=holding_id).json()
- if data.get('alreadyInRegion'):
- return True
- else:
- r = self._travel(data.get('preselectCountryId'), data.get('preselectRegionId')).json()
- if r.get('message', '') == 'success':
- self._update_citizen_location(data.get('preselectCountryId'), data.get('preselectRegionId'))
- return True
- return False
-
- def travel_to_battle(self, battle_id: int, allowed_countries: List[int]) -> bool:
- data = self.get_travel_regions(battle_id=battle_id)
-
- regs = []
- 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 regs:
- reg = min(regs, key=lambda _: int(_[0]))
- region_id = reg[1]
- country_id = reg[2]
- r = self._travel(country_id, region_id).json()
- if r.get('message', '') == 'success':
- self._update_citizen_location(country_id, region_id)
- return True
- return False
-
- def _travel(self, country_id: int, region_id: int = 0) -> Response:
- data = {
- "toCountryId": country_id,
- "inRegionId": region_id,
- }
- return self._post_main_travel("moveAction", **data)
-
- def get_travel_regions(self, holding_id: int = 0, battle_id: int = 0, country_id: int = 0
- ) -> 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', [])
-
- def get_travel_countries(self) -> Set[int]:
- response_json = self._post_main_travel_data().json()
- return_list = {*[]}
- for country_data in response_json['countries'].values():
- if country_data['currentRegions']:
- return_list.add(country_data['id'])
- return return_list
-
- def parse_notifications(self, page: int = 1) -> list:
- community = self._get_main_notifications_ajax_community(page).json()
- system = self._get_main_notifications_ajax_system(page).json()
- return community['alertsList'] + system['alertsList']
-
- def delete_notifications(self):
- response = self._get_main_notifications_ajax_community().json()
- while response['totalAlerts']:
- self._post_main_messages_alert([_['id'] for _ in response['alertList']])
- response = self._get_main_notifications_ajax_community().json()
-
- response = self._get_main_notifications_ajax_system().json()
- while response['totalAlerts']:
- self._post_main_messages_alert([_['id'] for _ in response['alertList']])
- response = self._get_main_notifications_ajax_system().json()
-
- def collect_weekly_reward(self):
- self.update_weekly_challenge()
-
- def collect_daily_task(self) -> None:
- self.update_citizen_info()
- if self.details.daily_task_done and not self.details.daily_task_reward:
- self._post_main_daily_task_reward()
-
- def send_mail_to_owner(self) -> None:
- if not self.details.citizen_id == 1620414:
- self.send_mail("Started", "time {}".format(self.now.strftime("%Y-%m-%d %H-%M-%S")), [1620414, ])
- self.sleep(1)
- msg_id = re.search(r"", self.r.text).group(1)
- self._post_delete_message([msg_id])
-
- def get_market_offers(self, country_id: int = None, product_name: str = None, quality: int = None) -> dict:
- raw_short_names = dict(frm="foodRaw", wrm="weaponRaw", hrm="houseRaw", arm="airplaneRaw")
- q1_industries = ["aircraft"] + list(raw_short_names.values())
- if product_name:
- if product_name not in self.available_industries and product_name not in raw_short_names:
- self.write_log(f"Industry '{product_name}' not implemented")
- raise ErepublikException(f"Industry '{product_name}' not implemented")
- elif product_name in raw_short_names:
- quality = 1
- product_name = raw_short_names[product_name]
- product_name = [product_name]
- elif quality:
- raise ErepublikException("Quality without product not allowed")
-
- item_data = dict(price=999999., country=0, amount=0, offer_id=0, citizen_id=0)
-
- items = {"food": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
- q5=item_data.copy(), q6=item_data.copy(), q7=item_data.copy()),
- "weapon": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
- q5=item_data.copy(), q6=item_data.copy(), q7=item_data.copy()),
- "house": dict(q1=item_data.copy(), q2=item_data.copy(), q3=item_data.copy(), q4=item_data.copy(),
- q5=item_data.copy()), "aircraft": dict(q1=item_data.copy()),
- "foodRaw": dict(q1=item_data.copy()), "weaponRaw": dict(q1=item_data.copy()),
- "houseRaw": dict(q1=item_data.copy()), "airplaneRaw": dict(q1=item_data.copy())}
-
- if country_id:
- countries = [country_id]
- else:
- countries = self.get_travel_countries()
-
- start_dt = self.now
- iterable = [countries, product_name or items, [quality] if quality else range(1, 8)]
- for country, industry, q in product(*iterable):
- if (q > 1 and industry in q1_industries) or (q > 5 and industry == "house"):
- continue
-
- r = self._post_economy_marketplace(country, self.available_industries[industry], q).json()
- obj = items[industry][f"q{q}"]
- if not r.get("error", False):
- for offer in r["offers"]:
- if obj["price"] > float(offer["priceWithTaxes"]):
- obj["price"] = float(offer["priceWithTaxes"])
- obj["country"] = int(offer["country_id"])
- obj["amount"] = int(offer["amount"])
- obj["offer_id"] = int(offer["id"])
- obj["citizen_id"] = int(offer["citizen_id"])
- elif obj["price"] == float(offer["priceWithTaxes"]) and obj["amount"] < int(offer["amount"]):
- obj["country"] = int(offer["country_id"])
- obj["amount"] = int(offer["amount"])
- obj["offer_id"] = int(offer["id"])
- self.write_log(f"Scraped market in {self.now - start_dt}!")
-
- if quality:
- ret = items[product_name[0]]["q%i" % quality]
- elif product_name:
- if product_name[0] in raw_short_names.values():
- ret = items[product_name[0]]["q1"]
- else:
- ret = items[product_name[0]]
- else:
- ret = items
- return ret
-
- def buy_food(self):
- hp_per_quality = {"q1": 2, "q2": 4, "q3": 6, "q4": 8, "q5": 10, "q6": 12, "q7": 20}
- hp_needed = 48 * self.energy.interval * 10 - self.food["total"]
- local_offers = self.get_market_offers(country_id=self.details.current_country, product_name="food")
-
- cheapest_q, cheapest = sorted(local_offers.items(), key=lambda v: v[1]["price"] / hp_per_quality[v[0]])[0]
-
- if cheapest["amount"] * hp_per_quality[cheapest_q] < hp_needed:
- amount = cheapest["amount"]
- else:
- amount = hp_needed // hp_per_quality[cheapest_q]
-
- if amount * cheapest["price"] < self.details.cc:
- data = dict(offer=cheapest["offer_id"], amount=amount, price=cheapest["price"],
- cost=amount * cheapest["price"], quality=cheapest_q, energy=amount * hp_per_quality[cheapest_q])
- self.reporter.report_action("BUY_FOOD", json_val=data)
- self.buy_from_market(cheapest["offer_id"], amount)
- self.update_inventory()
- else:
- s = f"Don't have enough money! Needed: {amount * cheapest['price']}cc, Have: {self.details.cc}cc"
- self.write_log(s)
- self.reporter.report_action("BUY_FOOD", value=s)
-
- def get_monetary_offers(self, currency: int = 62) -> List[Dict[str, Union[int, float]]]:
- if currency not in [1, 62]:
- currency = 62
- resp = self._post_economy_exchange_retrieve(False, 0, currency).json()
- ret = []
- offers = re.findall(r"id='purchase_(\d+)' data-i18n='Buy for' data-currency='GOLD' "
- r"data-price='(\d+\.\d+)' data-max='(\d+\.\d+)' trigger='purchase'",
- resp["buy_mode"], re.M | re.I | re.S)
-
- for offer_id, price, amount in offers:
- ret.append({"offer_id": int(offer_id), "price": float(price), "amount": float(amount)})
-
- return sorted(ret, key=lambda o: (o["price"], -o["amount"]))
-
- def buy_monetary_market_offer(self, offer: int, amount: float, currency: int) -> bool:
- response = self._post_economy_exchange_purchase(amount, currency, offer)
- self.details.cc = float(response.json().get("ecash").get("value"))
- self.details.gold = float(response.json().get("gold").get("value"))
- self.reporter.report_action("BUY_GOLD", json_val=response.json(),
- value=f"New amount {self.details.cc}cc, {self.details.gold}g")
- return not response.json().get("error", False)
+ return utils.calculate_hit(0, rang, True, elite, ne, 0, 20 if weapon else 0)
def activate_dmg_booster(self):
if self.config.boosters:
@@ -1518,53 +1833,6 @@ class Citizen(CitizenAPI):
def activate_pp_booster(self, battle_id: int) -> Response:
return self._post_military_fight_activate_booster(battle_id, 1, 180, "prestige_points")
- def donate_money(self, citizen_id: int = 1620414, amount: float = 0.0, currency: int = 62) -> bool:
- """ currency: gold = 62, cc = 1 """
- resp = self._post_economy_donate_money_action(citizen_id, amount, currency)
- r = re.search('You do not have enough money in your account to make this donation', resp.text)
- return not bool(r)
-
- def donate_items(self, citizen_id: int = 1620414, amount: int = 0, industry_id: int = 1, quality: int = 1) -> int:
- if amount < 1:
- return 0
- ind = {v: k for k, v in self.available_industries.items()}
- self.write_log(f"Donate: {amount:4d}q{quality} {ind[industry_id]} to {citizen_id}")
- response = self._post_economy_donate_items_action(citizen_id, amount, industry_id, quality)
- if re.search(rf"Successfully transferred {amount} item\(s\) to", response.text):
- return amount
- else:
- if re.search(r"You do not have enough items in your inventory to make this donation", response.text):
- return 0
- available = re.search(rf"Cannot transfer the items because the user has only (\d+) free slots in (his|her) "
- rf"storage.", response.text).group(1)
- return self.donate_items(citizen_id, int(available), industry_id, quality)
-
- def candidate_for_congress(self, presentation: str = "") -> Response:
- return self._post_candidate_for_congress(presentation)
-
- def candidate_for_party_presidency(self) -> Response:
- return self._get_candidate_party(self.politics.party_slug)
-
- def accept_money_donations(self):
- for notification in self.parse_notifications():
- don_id = re.search(r"erepublik.functions.acceptRejectDonation\(\"accept\", (\d+)\)", notification)
- if don_id:
- self._get_main_money_donation_accept(int(don_id.group(1)))
- self.sleep(5)
-
- def reject_money_donations(self) -> int:
- r = self._get_main_notifications_ajax_system()
- count = 0
- donation_ids = re.findall(r"erepublik.functions.acceptRejectDonation\(\"reject\", (\d+)\)", r.text)
- while donation_ids:
- for don_id in donation_ids:
- self._get_main_money_donation_reject(int(don_id))
- count += 1
- self.sleep(5)
- r = self._get_main_notifications_ajax_system()
- donation_ids = re.findall(r"erepublik.functions.acceptRejectDonation\(\"reject\", (\d+)\)", r.text)
- return count
-
def _rw_choose_side(self, battle_id: int, side_id: int) -> Response:
return self._post_main_battlefield_travel(side_id, battle_id)
@@ -1653,262 +1921,58 @@ class Citizen(CitizenAPI):
return count if count > 0 else 0
- @property
- def next_reachable_energy(self) -> int:
- # Return pps for furthest __reachable__ +1 energy else 0
- max_pp = 0
- for pp_milestone in self.details.next_pp:
- pp_milestone = int(pp_milestone)
- if self.details.pp + self.energy.food_fights > pp_milestone: # if reachable set max pp
- max_pp = pp_milestone
- else: # rest are only bigger no need
- break
- return max_pp - self.details.pp if max_pp else 0
+ 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 {}
- @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))
+ data = dict(zoneId=round_id, round=round_id, division=division, leftPage=1, rightPage=1, type="damage")
- @property
- def time_till_week_change(self) -> timedelta:
- return self.next_wc_start - self.now
+ 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")}
- @property
- def time_till_full_ff(self) -> timedelta:
- energy = self.energy.recoverable + self.energy.recovered
- if energy >= self.energy.limit * 2:
- return timedelta(0)
- minutes_needed = round((self.energy.limit * 2 - energy) / self.energy.interval) * 6
- return (self.energy.reference_time - self.now) + timedelta(minutes=minutes_needed)
+ def schedule_attack(self, war_id: int, region_id: int, region_name: str, at_time: datetime):
+ if at_time:
+ self.sleep(utils.get_sleep_seconds(at_time))
+ self.get_csrf_token()
+ self.launch_attack(war_id, region_id, region_name)
- @property
- def max_time_till_full_ff(self) -> timedelta:
- """
- Max required time for 0 to full energy (0/0 -> limit/limit) (last interval rounded up)
- :return:
- """
- return timedelta(minutes=round((self.energy.limit * 2 / self.energy.interval) + 0.49) * 6)
+ 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))
+ all_war_ids = re.findall(r'//www\.erepublik\.com/en/wars/show/(\d+)"', r.text)
+ return [int(wid) for wid in all_war_ids]
- @property
- def is_levelup_close(self) -> bool:
- """
- If Energy limit * 2 >= xp till levelup * 10
- :return: bool
- """
- return self.energy.limit * 2 >= self.details.xp_till_level_up * 10
+ def get_last_battle_of_war_end_time(self, war_id: int) -> datetime:
+ r = self._get_wars_show(war_id)
+ html = r.text
+ last_battle_id = int(re.search(r'', html).group(1))
+ console = self._post_military_battle_console(last_battle_id, 'warList', 1).json()
+ battle = console.get('list')[0]
+ return utils.localize_dt(datetime.strptime(battle.get('result').get('end'), "%Y-%m-%d %H:%M:%S"))
- @property
- def is_levelup_reachable(self) -> bool:
- """
- If Energy limit >= xp till levelup * 10
- :return: bool
- """
- return self.energy.limit >= self.details.xp_till_level_up * 10
+ def launch_attack(self, war_id: int, region_id: int, region_name: str):
+ self._post_wars_attack_region(war_id, region_id, region_name)
+ self.telegram.send_message(f"Battle for *{region_name}* queued")
- @property
- def should_do_levelup(self) -> bool:
- """
- If Energy limit >= xp till levelup * 10
- :return: bool
- """
- return (self.energy.recovered >= self.details.xp_till_level_up * 10 and # can reach next level
- self.energy.recoverable + 2 * self.energy.interval >= self.energy.limit) # can do max amount of dmg
+ def travel_to_battle(self, battle_id: int, allowed_countries: List[int]) -> bool:
+ data = self.get_travel_regions(battle_id=battle_id)
- def get_article_comments(self, article_id: int = 2645676, page_id: int = 1) -> Response:
- return self._post_main_article_comments(article_id, page_id)
-
- def comment_article(self, article_id: int = 2645676, msg: str = None) -> Response:
- if msg is None:
- msg = self.eday
- r = self.get_article_comments(article_id, 2)
- r = self.get_article_comments(article_id, r.json()["pages"])
- comments = r.json()["comments"]
- if not comments[max(comments.keys())]["isMyComment"]:
- r = self.write_article_comment(msg, article_id)
- return r
-
- def write_article_comment(self, message: str, article_id: int, parent_id: int = None) -> Response:
- return self._post_main_article_comments_create(message, article_id, parent_id)
-
- def publish_article(self, title: str, content: str, kind: int) -> Response:
- kinds = {1: "First steps in eRepublik", 2: "Battle orders", 3: "Warfare analysis",
- 4: "Political debates and analysis", 5: "Financial business",
- 6: "Social interactions and entertainment"}
- if kind in kinds:
- return self._post_main_write_article(title, content, self.details.citizenship, kind)
- else:
- raise ErepublikException("Article kind must be one of:\n{}\n'{}' is not supported".format(
- "\n".join(["{}: {}".format(k, v) for k, v in kinds.items()]), kind
- ))
-
- def get_my_market_offers(self) -> List[Dict[str, Union[int, float, str]]]:
- ret = []
- for offer in self._get_economy_my_market_offers().json():
- line = offer.copy()
- line.pop('icon', None)
- ret.append(line)
- return ret
-
- def post_market_offer(self, industry: int, quality: int, amount: int, price: float) -> Response:
- if industry not in self.available_industries.values():
- self.write_log(f"Trying to sell unsupported industry {industry}")
-
- data = {
- "country": self.details.citizenship,
- "industry": industry,
- "quality": quality,
- "amount": amount,
- "price": price,
- "buy": False,
- }
- ret = self._post_economy_marketplace_actions(**data)
- self.reporter.report_action("SELL_PRODUCT", ret.json())
- return ret
-
- def buy_from_market(self, offer: int, amount: int) -> dict:
- ret = self._post_economy_marketplace_actions(amount, True, offer=offer)
- json_ret = ret.json()
- if json_ret.get('error'):
- return json_ret
- else:
- self.details.cc = ret.json()['currency']
- self.details.gold = ret.json()['gold']
- r_json = ret.json()
- r_json.pop("offerUpdate", None)
- self.reporter.report_action("BUY_PRODUCT", ret.json())
- return json_ret
-
- def assign_factory_to_holding(self, factory_id: int, holding_id: int) -> Response:
- """
- Assigns factory to new holding
- """
- company = self.my_companies.companies[factory_id]
- company_name = self.factories[company['industry_id']]
- if not company['is_raw']:
- company_name += f" q{company['quality']}"
- self.write_log(f"{company_name} 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)
-
- def create_factory(self, industry_id: int, building_type: int = 1) -> Response:
- """
- param industry_ids: FRM={q1:7, q2:8, q3:9, q4:10, q5:11} WRM={q1:12, q2:13, q3:14, q4:15, q5:16}
- HRM={q1:18, q2:19, q3:20, q4:21, q5:22} ARM={q1:24, q2:25, q3:26, q4:27, q5:28}
- Factories={Food:1, Weapons:2, House:4, Aircraft:23} <- Building_type 1
-
- Storage={1000: 1, 2000: 2} <- Building_type 2
- """
- company_name = self.factories[industry_id]
- if building_type == 2:
- company_name = f"Storage"
- 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]
- company_name = self.factories[company['industry_id']]
- if not company['is_raw']:
- company_name += f" q{company['quality']}"
- self.write_log(f"{company_name} dissolved!")
- return self._post_economy_sell_company(factory_id, self.details.pin, sell=False)
-
- @property
- def available_industries(self) -> Dict[str, int]:
- """
- Returns currently available industries as dict(name: id)
- :return: dict
- """
- return {"food": 1, "weapon": 2, "house": 4, "aircraft": 23,
- "foodRaw": 7, "weaponRaw": 12, "houseRaw": 17, "airplaneRaw": 24}
-
- @property
- def factories(self) -> Dict[int, str]:
- """Returns factory industries as dict(id: name)
- :return: Factory id:name dict
- ":rtype: Dict[int, str]
- """
- return {1: "Food", 2: "Weapons", 4: "House", 23: "Aircraft",
- 7: "FRM q1", 8: "FRM q2", 9: "FRM q3", 10: "FRM q4", 11: "FRM q5",
- 12: "WRM q1", 13: "WRM q2", 14: "WRM q3", 15: "WRM q4", 16: "WRM q5",
- 18: "HRM q1", 19: "HRM q2", 20: "HRM q3", 21: "HRM q4", 22: "HRM q5",
- 24: "ARM q1", 25: "ARM q2", 26: "ARM q3", 27: "ARM q4", 28: "ARM q5", }
-
- def get_industry_id(self, industry_name: str) -> int:
- """Returns industry id
-
- :type industry_name: str
- :return: int
- """
- return self.available_industries.get(industry_name, 0)
-
- def get_industry_name(self, industry_id: int) -> str:
- """Returns industry name from industry ID
-
- :type industry_id: int
- :return: industry name
- :rtype: str
- """
- for iname, iid in self.available_industries.items():
- if iid == industry_id:
- return iname
- return ""
-
- def buy_tg_contract(self) -> Response:
- ret = self._post_main_buy_gold_items('gold', "TrainingContract2", 1)
- self.reporter.report_action("BUY_TG_CONTRACT", ret.json())
- return ret
-
- def resign(self) -> bool:
- self.update_job_info()
- if self.r.json().get("isEmployee"):
- self.reporter.report_action("RESIGN", self.r.json())
- self._post_economy_resign()
- return True
+ regs = []
+ 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 regs:
+ reg = min(regs, key=lambda _: int(_[0]))
+ region_id = reg[1]
+ country_id = reg[2]
+ r = self._travel(country_id, region_id).json()
+ if r.get('message', '') == 'success':
+ self._update_citizen_location(country_id, region_id)
+ return True
return False
- def find_new_job(self) -> Response:
- r = self._get_economy_job_market_json(self.details.current_country)
- jobs = r.json().get("jobs")
- data = dict(citizen=0, salary=10)
- for posting in jobs:
- salary = posting.get("salary")
- limit = posting.get("salaryLimit", 0)
- userid = posting.get("citizen").get("id")
-
- if (not limit or salary * 3 < limit) and salary > data["salary"]:
- data.update({"citizen": userid, "salary": salary})
- self.reporter.report_action("APPLYING_FOR_JOB", jobs, str(data['citizen']))
- return self._post_economy_job_market_apply(**data)
-
- def add_friend(self, player_id: int) -> Response:
- resp = self._get_main_citizen_hovercard(player_id)
- rjson = resp.json()
- if not any([rjson["isBanned"], rjson["isDead"], rjson["isFriend"], rjson["isOrg"], rjson["isSelf"]]):
- r = self._post_main_citizen_add_remove_friend(int(player_id), True)
- self.write_log(f"{rjson['name']:<64} (id:{player_id:>11}) added as friend")
- return r
- return resp
-
- 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)
- ret = {}
- for name, id_ in re.findall(r'', r.text):
- ret.update({int(id_): name})
- return ret
-
- def _get_main_party_members(self, party_id: int) -> Dict[int, str]:
- ret = {}
- r = super()._get_main_party_members(party_id)
- for id_, name in re.findall(r'', r.text):
- ret.update({id_: name})
- return ret
-
def get_country_mus(self, country_id: int) -> Dict[int, str]:
ret = {}
r = self._get_main_leaderboards_damage_rankings(country_id)
@@ -1932,12 +1996,56 @@ class Citizen(CitizenAPI):
ret.update({user["citizenId"]: user["name"]})
return ret
+
+class CitizenAnniversary(BaseCitizen):
+ def collect_anniversary_reward(self) -> Response:
+ return self._post_main_collect_anniversary_reward()
+
+ def get_anniversary_quest_data(self):
+ return self._get_anniversary_quest_data().json()
+
+ def start_unlocking_map_quest_node(self, node_id: int):
+ return self._post_map_rewards_unlock(node_id)
+
+ def collect_map_quest_node(self, node_id: int):
+ return self._post_map_rewards_claim(node_id)
+
+ def speedup_map_quest_node(self, node_id: int):
+ node = self.get_anniversary_quest_data().get('cities', {}).get(str(node_id), {})
+ return self._post_map_rewards_speedup(node_id, node.get("skipCost", 0))
+
+
+class CitizenSocial(BaseCitizen):
+ def send_mail_to_owner(self):
+ if not self.details.citizen_id == 1620414:
+ self.send_mail("Started", "time {}".format(self.now.strftime("%Y-%m-%d %H-%M-%S")), [1620414, ])
+ self.sleep(1)
+ msg_id = re.search(r"", self.r.text).group(1)
+ self._post_delete_message([msg_id])
+
def send_mail(self, subject: str, msg: str, ids: List[int] = None):
if ids is None:
ids = [1620414, ]
for player_id in ids:
self._post_main_messages_compose(subject, msg, [player_id])
+ def write_on_country_wall(self, message: str) -> bool:
+ self._get_main()
+ post_to_wall_as = re.findall(r'id="post_to_country_as".*?.*',
+ self.r.text, re.S | re.M)
+ r = self._post_main_country_post_create(message, max(post_to_wall_as, key=int) if post_to_wall_as else 0)
+ return r.json()
+
+ def add_friend(self, player_id: int) -> Response:
+ resp = self._get_main_citizen_hovercard(player_id)
+ rjson = resp.json()
+ if not any([rjson["isBanned"], rjson["isDead"], rjson["isFriend"], rjson["isOrg"], rjson["isSelf"]]):
+ r = self._post_main_citizen_add_remove_friend(int(player_id), True)
+ self.write_log(f"{rjson['name']:<64} (id:{player_id:>11}) added as friend")
+ return r
+ return resp
+
def add_every_player_as_friend(self):
cities = []
cities_dict = {}
@@ -1959,48 +2067,170 @@ class Citizen(CitizenAPI):
for resident in resp["widgets"]["residents"]["residents"]:
self.add_friend(resident["citizenId"])
- def schedule_attack(self, war_id: int, region_id: int, region_name: str, at_time: datetime):
- if at_time:
- self.sleep(get_sleep_seconds(at_time))
+
+class Citizen(CitizenMilitary, CitizenAnniversary, CitizenEconomy, CitizenSocial, CitizenPolitics):
+ debug: bool = False
+
+ def __init__(self, email: str = "", password: str = "", auto_login: bool = True):
+ super().__init__()
+ self.__last_full_update = utils.good_timedelta(self.now, - timedelta(minutes=5))
+ self.commit_id = utils.COMMIT_ID
+ self.config = Config()
+ self.config.email = email
+ self.config.password = password
+ self.energy = Energy()
+ self.details = Details()
+ self.politics = Politics()
+ self.my_companies = MyCompanies()
+ self.set_debug(True)
+ self.reporter = Reporter()
+ self.stop_threads = Event()
+ self.telegram = TelegramBot(stop_event=self.stop_threads)
+ if auto_login:
+ self.login()
+
+ def config_setup(self, **kwargs):
+ self.config.reset()
+ for key, value in kwargs.items():
+ if hasattr(self.config, key):
+ setattr(self.config, key, value)
+ else:
+ self.write_log(f"Unknown config parameter! ({key}={value})")
+
+ def login(self):
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(COUNTRY_LINK.get(country_id or self.details.citizenship))
- all_war_ids = re.findall(r'//www\.erepublik\.com/en/wars/show/(\d+)"', r.text)
- return [int(wid) for wid in all_war_ids]
+ self.update_citizen_info()
+ self.reporter.do_init(self.name, self.config.email, self.details.citizen_id)
+ if self.config.telegram:
+ self.telegram.do_init(self.config.telegram_chat_id or 620981703,
+ self.config.telegram_token or "864251270:AAFzZZdjspI-kIgJVk4gF3TViGFoHnf8H4o",
+ "" if self.config.telegram_chat_id or self.config.telegram_token else self.name)
+ self.telegram.send_message(f"*Started* {utils.now():%F %T}")
- def get_war_status(self, war_id: int) -> Dict[str, Union[bool, Dict[int, str]]]:
- r = self._get_wars_show(war_id)
- html = r.text
- ret = {}
- reg_re = re.compile(fr'data-war-id="{war_id}" data-region-id="(\d+)" data-region-name="([- \w]+)"')
- if reg_re.findall(html):
- ret.update(regions={}, can_attack=True)
- for reg in reg_re.findall(html):
- ret["regions"].update({int(reg[0]): reg[1]})
- elif re.search(r'Join', html):
- battle_id = re.search(r'Join', html).group(1)
- ret.update(can_attack=False, battle_id=int(battle_id))
- elif re.search(r'This war is no longer active.', html):
- ret.update(can_attack=False, ended=True)
+ self.__last_full_update = utils.good_timedelta(self.now, - timedelta(minutes=5))
+
+ def update_citizen_info(self, html: str = None):
+ """
+ Gets main page and updates most information about player
+ """
+ if html is None:
+ self._get_main()
+ return
+ super().update_citizen_info(html)
+
+ if self.promos.get("trainingContract"):
+ if not self.tg_contract:
+ self.train()
+ if not self.tg_contract["free_train"] and self.tg_contract.get("active", False):
+ if self.details.gold >= 54:
+ self.buy_tg_contract()
+ else:
+ self.write_log(f"Training ground contract active but "
+ f"don't have enough gold ({self.details.gold}g {self.details.cc}cc)")
+ if self.energy.is_energy_full and self.config.telegram:
+ self.telegram.report_full_energy(self.energy.available, self.energy.limit, self.energy.interval)
+
+ def check_for_notification_medals(self):
+ notifications = self._get_main_citizen_daily_assistant().json()
+ data: Dict[Tuple[str, Union[float, str]], Dict[str, Union[int, str, float]]] = {}
+ for medal in notifications.get('notifications', []):
+ if medal.get('details', {}).get('type') == "citizenAchievement":
+ params: dict = medal.get('details', {}).get('achievement')
+ about: str = medal.get('details').get('description')
+ title: str = medal.get('title')
+
+ award_id: int = medal.get('id')
+ if award_id and title:
+ self._post_main_wall_post_automatic(title.lower(), award_id)
+
+ if params.get('ccValue'):
+ reward = params.get('ccValue')
+ currency = "Currency"
+ elif params.get('goldValue'):
+ reward = params.get('goldValue')
+ currency = "Gold"
+ else:
+ reward = params.get('energyValue')
+ currency = "Energy"
+
+ if (title, reward) not in data:
+ data[(title, reward)] = {'about': about, 'kind': title, 'reward': reward, "count": 1,
+ "currency": currency, "params": params}
+ else:
+ data[(title, reward)]['count'] += 1
+ self._post_main_global_alerts_close(medal.get('id'))
+ if data:
+ msgs = ["{count} x {kind},"
+ " totaling {} {currency}".format(d["count"] * d["reward"], **d) for d in data.values()]
+
+ msgs = "\n".join(msgs)
+ self.telegram.report_medal(msgs)
+ self.write_log(f"Found awards:\n{msgs}")
+ for info in data.values():
+ self.reporter.report_action("NEW_MEDAL", info)
+
+ def set_debug(self, debug: bool):
+ self.debug = debug
+ self._req.debug = debug
+
+ def set_pin(self, pin: int):
+ self.details.pin = pin
+
+ def update_all(self, force_update=False):
+ # Do full update max every 5 min
+ if utils.good_timedelta(self.__last_full_update, timedelta(minutes=5)) > self.now and not force_update:
+ return
else:
- ret.update(can_attack=False)
- return ret
+ self.__last_full_update = self.now
+ self.update_citizen_info()
+ self.update_war_info()
+ self.update_inventory()
+ self.update_companies()
+ self.update_money()
+ self.update_weekly_challenge()
+ self.send_state_update()
+ self.check_for_notification_medals()
- def get_last_battle_of_war_end_time(self, war_id: int) -> datetime:
- r = self._get_wars_show(war_id)
- html = r.text
- last_battle_id = int(re.search(r'', html).group(1))
- console = self._post_military_battle_console(last_battle_id, 'warList', 1).json()
- battle = console.get('list')[0]
- return localize_dt(datetime.strptime(battle.get('result').get('end'), "%Y-%m-%d %H:%M:%S"))
+ def update_weekly_challenge(self):
+ data = self._get_main_weekly_challenge_data().json()
+ self.details.pp = data.get("player", {}).get("prestigePoints", 0)
+ self.details.next_pp.clear()
+ for reward in data.get("rewards", {}).get("normal", {}):
+ status = reward.get("status", "")
+ if status == "rewarded":
+ continue
+ elif status == "completed":
+ self._post_main_weekly_challenge_reward(reward.get("id", 0))
+ elif reward.get("icon", "") == "energy_booster":
+ pps = re.search(r"Reach (\d+) Prestige Points to unlock the following reward: \+1 Energy",
+ reward.get("tooltip", ""))
+ if pps:
+ self.details.next_pp.append(int(pps.group(1)))
- def launch_attack(self, war_id: int, region_id: int, region_name: str):
- self._post_wars_attack_region(war_id, region_id, region_name)
- self.telegram.send_message(f"Battle for *{region_name}* queued")
+ def parse_notifications(self, page: int = 1) -> list:
+ community = self._get_main_notifications_ajax_community(page).json()
+ system = self._get_main_notifications_ajax_system(page).json()
+ return community['alertsList'] + system['alertsList']
+
+ def delete_notifications(self):
+ response = self._get_main_notifications_ajax_community().json()
+ while response['totalAlerts']:
+ self._post_main_messages_alert([_['id'] for _ in response['alertList']])
+ response = self._get_main_notifications_ajax_community().json()
+
+ response = self._get_main_notifications_ajax_system().json()
+ while response['totalAlerts']:
+ self._post_main_messages_alert([_['id'] for _ in response['alertList']])
+ response = self._get_main_notifications_ajax_system().json()
+
+ def collect_weekly_reward(self):
+ self.update_weekly_challenge()
+
+ def collect_daily_task(self):
+ self.update_citizen_info()
+ if self.details.daily_task_done and not self.details.daily_task_reward:
+ self._post_main_daily_task_reward()
def state_update_repeater(self):
try:
@@ -2008,10 +2238,10 @@ class Citizen(CitizenAPI):
if start_time.minute <= 30:
start_time = start_time.replace(minute=30)
else:
- start_time = good_timedelta(start_time.replace(minute=0), timedelta(hours=1))
+ start_time = utils.good_timedelta(start_time.replace(minute=0), timedelta(hours=1))
while not self.stop_threads.is_set():
self.update_citizen_info()
- start_time = good_timedelta(start_time, timedelta(minutes=30))
+ start_time = utils.good_timedelta(start_time, timedelta(minutes=30))
self.send_state_update()
self.send_inventory_update()
sleep_seconds = (start_time - self.now).total_seconds()
@@ -2028,230 +2258,3 @@ class Citizen(CitizenAPI):
def send_inventory_update(self):
to_report = self.update_inventory()
self.reporter.report_action("INVENTORY", json_val=to_report)
-
- def check_house_durability(self) -> Dict[int, datetime]:
- ret = {}
- inv = self.update_inventory()
- for house_quality, active_house in inv['items']['active'].get('house', {}).items():
- till = good_timedelta(self.now, timedelta(seconds=active_house['time_left']))
- ret.update({house_quality: till})
- return ret
-
- def buy_and_activate_house(self, q: int) -> Dict[int, datetime]:
- inventory = self.update_inventory()
- ok_to_activate = False
- if not inventory['items']['final'].get('house', {}).get(q, {}):
- offers = []
- countries = [self.details.citizenship, ]
- if self.details.current_country != self.details.citizenship:
- countries.append(self.details.current_country)
- for country in countries:
- offers += [self.get_market_offers(country, "house", q)]
- global_cheapest = self.get_market_offers(product_name="house", quality=q)
- cheapest_offer = sorted(offers, key=lambda o: o["price"])[0]
- region = self.get_country_travel_region(global_cheapest['country'])
- if global_cheapest['price'] + 200 < cheapest_offer['price'] and region:
- self._travel(global_cheapest['country'], region)
- buy = self.buy_from_market(global_cheapest['offer_id'], 1)
- else:
- buy = self.buy_from_market(cheapest_offer['offer_id'], 1)
- if buy["error"]:
- msg = f"Unable to buy q{q} house! \n{buy['message']}"
- self.write_log(msg)
- else:
- ok_to_activate = True
- else:
- ok_to_activate = True
- if ok_to_activate:
- self.activate_house(q)
- return self.check_house_durability()
-
- def renew_houses(self, forced: bool = False) -> Dict[int, datetime]:
- """
- Renew all houses which endtime is in next 48h
- :param forced: if true - renew all houses
- :return:
- """
- house_durability = self.check_house_durability()
- for q, active_till in house_durability.items():
- if good_timedelta(active_till, - timedelta(hours=48)) <= self.now or forced:
- house_durability = self.buy_and_activate_house(q)
- self.travel_to_residence()
- return house_durability
-
- def activate_house(self, quality: int) -> datetime:
- active_until = self.now
- r = self._post_economy_activate_house(quality)
- if r.json().get("status") and not r.json().get("error"):
- house = r.json()["inventoryItems"]["activeEnhancements"]["items"]["4_%i_active" % quality]
- active_until = good_timedelta(active_until, timedelta(seconds=house["active"]["time_left"]))
- return active_until
-
- def collect_anniversary_reward(self) -> Response:
- return self._post_main_collect_anniversary_reward()
-
- 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 {}
-
- 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")}
-
- def contribute_cc_to_country(self, amount=0., country_id: int = 71) -> 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)
- self.reporter.report_action("CONTRIBUTE_CC", data, str(amount))
- r = self._post_main_country_donate(**data)
- return r.json().get('status') or not r.json().get('error')
-
- def contribute_food_to_country(self, amount: int = 0, quality: int = 1, country_id: int = 71) -> 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)
- self.reporter.report_action("CONTRIBUTE_FOOD", data, FOOD_ENERGY[quality] * amount)
- r = self._post_main_country_donate(**data)
- return r.json().get('status') or not r.json().get('error')
-
- def contribute_gold_to_country(self, amount: int, country_id: int = 71) -> bool:
- self.update_money()
-
- if self.details.cc < amount:
- return False
- 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)
- return r.json().get('status') or not r.json().get('error')
-
- def write_on_country_wall(self, message: str) -> bool:
- self._get_main()
- post_to_wall_as = re.findall(r'id="post_to_country_as".*?.*',
- self.r.text, re.S | re.M)
- r = self._post_main_country_post_create(message, max(post_to_wall_as, key=int) if post_to_wall_as else 0)
- return r.json()
-
- def report_error(self, msg: str = "", is_warning: bool = False):
- if is_warning:
- process_warning(msg, self.name, sys.exc_info(), self, self.commit_id)
- else:
- process_error(msg, self.name, sys.exc_info(), self, self.commit_id, None)
-
- def get_battle_top_10(self, battle_id: int) -> Dict[int, List[Tuple[int, int]]]:
- 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 json.dumps(self.__dict__, cls=MyJSONEncoder, indent=4 if indent else None, sort_keys=True)
-
- def get_game_token_offers(self):
- r = self._post_economy_game_tokens_market('retrieve').json()
- return {v.get('id'): dict(amount=v.get('amount'), price=v.get('price')) for v in r.get("topOffers")}
-
- def fetch_organisation_account(self, org_id: int):
- r = self._get_economy_citizen_accounts(org_id)
- table = re.search(r'()', r.text, re.I | re.M | re.S)
- if table:
- account = re.findall(r'>\s*(\d+.\d+)\s*', table.group(1))
- if account:
- return {"gold": account[0], "cc": account[1], 'ok': True}
-
- return {"gold": 0, "cc": 0, 'ok': False}
-
- 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) -> float:
- if not rang or strength or elite is None:
- r = self._get_main_citizen_profile_json(self.details.citizen_id).json()
- if not rang:
- rang = r['military']['militaryData']['ground']['rankNumber']
- if not strength:
- strength = r['military']['militaryData']['ground']['strength']
- if elite is None:
- elite = r['citizenAttributes']['level'] > 100
- if ne:
- tp = True
-
- return calculate_hit(strength, rang, tp, elite, ne, 50 if booster_50 else 100 if booster_100 else 0)
-
- def get_air_hit_dmg_value(self, rang: int = None, elite: bool = None, ne: bool = False,
- weapon: bool = False) -> float:
- if not rang or elite is None:
- r = self._get_main_citizen_profile_json(self.details.citizen_id).json()
- if not rang:
- rang = r['military']['militaryData']['air']['rankNumber']
- if elite is None:
- elite = r['citizenAttributes']['level'] > 100
-
- return calculate_hit(0, rang, True, elite, ne, 0, 20 if weapon else 0)
-
- def endorse_article(self, article_id: int, amount: int) -> bool:
- if amount in (5, 50, 100):
- resp = self._post_main_donate_article(article_id, amount).json()
- return not bool(resp.get('error'))
- else:
- return False
-
- def vote_article(self, article_id: int) -> bool:
- resp = self._post_main_vote_article(article_id).json()
- return not bool(resp.get('error'))
-
- def get_anniversary_quest_data(self):
- return self._get_anniversary_quest_data().json()
-
- def start_unlocking_map_quest_node(self, node_id: int):
- return self._post_map_rewards_unlock(node_id)
-
- def collect_map_quest_node(self, node_id: int):
- return self._post_map_rewards_claim(node_id)
-
- def speedup_map_quest_node(self, node_id: int):
- node = self.get_anniversary_quest_data().get('cities', {}).get(str(node_id), {})
- return self._post_map_rewards_speedup(node_id, node.get("skipCost", 0))
-
- 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) -> int:
- battle = self.all_battles.get(battle_id)
- 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()
- weapon_quality = -1
- weapon_damage = 0
- if not battle.is_air:
- for weapon in available_weapons:
- try:
- if weapon['weaponQuantity'] > 30 and weapon['damage'] > weapon_damage:
- weapon_quality = int(weapon['weaponId'])
- except ValueError:
- pass
- return self.change_weapon(battle_id, weapon_quality)
-
- def change_weapon(self, battle_id: int, weapon_quality: int) -> int:
- battle = self.all_battles.get(battle_id)
- battle_zone = battle.div[11 if battle.is_air else self.division].battle_zone_id
- r = self._post_military_change_weapon(battle_id, battle_zone, weapon_quality)
- return r.json().get('weaponInfluence')
-
- def get_battle_for_war(self, war_id: int) -> Optional[Battle]:
- self.update_war_info()
- war_info = self.get_war_status(war_id)
- return self.all_battles.get(war_info.get("battle_id"), None)