eriks/erepublik-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Typehinting and battle/war stuff - last battle, attackable regions if CP, etc

Browse Source
This commit is contained in:
Eriks Karls 2019-08-01 15:05:44 +03:00
parent 4eccb339bb
commit 6bbc7a1f64
3 changed files with 78 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
erepublik/citizen.py
View File

@ -5,7 +5,7 @@ import sys
import threading
import time
from json import loads, dumps
from typing import Dict, List, Tuple, Any, Union
from typing import Dict, List, Tuple, Any, Union, Mapping
import requests
from requests import Response, RequestException
@ -21,9 +21,9 @@ class Citizen(classes.CitizenAPI):
all_battles: Dict[int, classes.Battle] = dict()
countries: Dict[int, Dict[str, Union[str, List[int]]]] = dict()
__last_war_update_data = {}
__last_full_update: datetime.datetime
__last_full_update: datetime.datetime = utils.now().min
active_fs = False
active_fs: bool = False
food = {"q1": 0, "q2": 0, "q3": 0, "q4": 0, "q5": 0, "q6": 0, "q7": 0, "total": 0}
inventory = {"used": 0, "total": 0}
@ -1669,18 +1669,45 @@ class Citizen(classes.CitizenAPI):
for resident in resp["widgets"]["residents"]["residents"]:
self.add_friend(resident["citizenId"])
def schedule_attack(self, war_id: int, region_id: int, at_time: datetime) -> None:
def schedule_attack(self, war_id: int, region_id: int, region_name: str, at_time: datetime) -> None:
if at_time:
self.sleep(utils.get_sleep_seconds(at_time))
self.get_csrf_token()
self._launch_battle(war_id, region_id)
self.launch_attack(war_id, region_id, region_name)
def get_active_wars_with_regions(self):
self._get_country_military(self.countries.get(self.details.citizen_id)["name"])
raise NotImplementedError
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]
def _launch_battle(self, war_id: int, region_id: int) -> Response:
return self._post_wars_attack_region(war_id, region_id)
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({str(reg[0]): reg[1]})
elif re.search(r'<a href="//www.erepublik.com/en/military/battlefield/(\d+)" '
r'class="join" title="Join"><span>Join</span></a>', html):
battle_id = re.search(r'<a href="//www.erepublik.com/en/military/battlefield/(\d+)" '
r'class="join" title="Join"><span>Join</span></a>', html).group(1)
ret.update(can_attack=False, battle_id=battle_id)
else:
ret.update(can_attack=False)
return ret
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'<a href="//www.erepublik.com/en/military/battlefield/(\d+)">', 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.datetime.strptime(battle.get('result').get('end'), "%Y-%m-%d %H:%M:%S"))
def launch_attack(self, war_id: int, region_id: int, region_name: str):
self._post_wars_attack_region(war_id, region_id, region_name)
def state_update_repeater(self):
try:

21
erepublik/classes.py
View File

@ -5,7 +5,7 @@ import random
import time
from collections import deque
from json import JSONDecodeError, loads, JSONEncoder
from typing import Any, Dict, List, Union
from typing import Any, Dict, List, Union, Mapping, Iterable
from requests import Response, Session
@ -41,7 +41,7 @@ class MyCompanies:
template = dict(id=0, num_factories=0, region_id=0, companies=[])
for holding_id, holding in holdings.items():
tmp: Dict[str, Union[List[Any], Any]] = {}
tmp: Dict[str, Union[Iterable[Any], Any]] = {}
for key in template:
if key == 'companies':
tmp.update({key: []})
@ -470,7 +470,7 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
def _get_citizen_daily_assistant(self) -> Response:
return self.get("{}/main/citizenDailyAssistant".format(self.url))
def _get_city_data_residents(self, city: int, page: int = 1, params: Dict[str, Any] = None) -> Response:
def _get_city_data_residents(self, city: int, page: int = 1, params: Mapping[str, Any] = None) -> Response:
if params is None:
params = {}
return self.get("{}/main/city-data/{}/residents".format(self.url, city), params={"currentPage": page, **params})
@ -549,6 +549,9 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
def _get_weekly_challenge_data(self) -> Response:
return self.get("{}/main/weekly-challenge-data".format(self.url))
def _get_wars_show(self, war_id: int) -> Response:
return self.get("{}/wars/show/{}".format(self.url, war_id))
def _post_activate_battle_effect(self, battle: int, kind: str, citizen_id: int) -> Response:
data = dict(battleId=battle, citizenId=citizen_id, type=kind, _token=self.token)
return self.post("{}/main/fight-activateBattleEffect".format(self.url), data=data)
@ -747,10 +750,14 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
citizen_subject=subject, _token=self.token, citizen_message=body)
return self.post("{}/main/messages-compose/{}}".format(self.url, url_pk), data=data)
def _post_military_battle_console(self, battle_id: int, round_id: int, division: int) -> Response:
data = dict(battleId=battle_id, zoneId=round_id, action="battleStatistics", round=round_id, division=division,
type="damage", leftPage=1, rightPage=1, _token=self.token)
return self.post("{}/military/battle-console".format(self.url, battle_id), data=data)
def _post_military_battle_console(self, battle_id: int, action: str, page: int = 1, **kwargs) -> Response:
data = dict(battleId=battle_id, action=action, _token=self.token)
if action == "battleStatistics":
data.update(round=kwargs["round_id"], zoneId=kwargs["round_id"], leftPage=page, rightPage=page,
division=kwargs["division"], type=kwargs.get("type", 'damage'),)
elif action == "warList":
data.update(page=page)
return self.post("{}/military/battle-console".format(self.url), data=data)
def _post_military_deploy_bomb(self, battle_id: int, bomb_id: int) -> Response:
data = dict(battleId=battle_id, bombId=bomb_id, _token=self.token)

36
erepublik/utils.py
View File

@ -11,13 +11,12 @@ from collections import deque
from decimal import Decimal
from json import JSONEncoder
from pathlib import Path
from typing import Union, Dict, Any, List
from typing import Union, Any, List, NoReturn, Mapping
import pytz
import requests
from requests import Response
__all__ = ["FOOD_ENERGY", "COMMIT_ID", "COUNTRIES", "erep_tz",
"now", "localize_dt", "localize_timestamp", "good_timedelta", "eday_from_date", "date_from_eday",
"get_sleep_seconds", "interactive_sleep", "silent_sleep",
@ -85,6 +84,19 @@ COUNTRIES = {1: 'Romania', 9: 'Brazil', 10: 'Italy', 11: 'France', 12: 'Germany'
82: 'Cyprus', 83: 'Belarus', 84: 'New Zealand', 164: 'Saudi Arabia', 165: 'Egypt',
166: 'United Arab Emirates', 167: 'Albania', 168: 'Georgia', 169: 'Armenia', 170: 'Nigeria', 171: 'Cuba'}
COUNTRY_LINK = {1: 'Romania', 9: 'Brazil', 11: 'France', 12: 'Germany', 13: 'Hungary', 82: 'Cyprus', 168: 'Georgia',
15: 'Spain', 23: 'Canada', 26: 'Mexico', 27: 'Argentina', 28: 'Venezuela', 80: 'Montenegro', 24: 'USA',
29: 'United-Kingdom', 50: 'Australia', 47: 'South-Korea',171: 'Cuba', 79: 'Republic-of-Macedonia-FYROM',
30: 'Switzerland', 31: 'Netherlands', 32: 'Belgium', 33: 'Austria', 34: 'Czech-Republic', 35: 'Poland',
36: 'Slovakia', 37: 'Norway', 38: 'Sweden', 39: 'Finland', 40: 'Ukraine', 41: 'Russia', 42: 'Bulgaria',
43: 'Turkey', 44: 'Greece', 45: 'Japan', 48: 'India', 49: 'Indonesia', 78: 'Colombia', 68: 'Singapore',
51: 'South Africa', 52: 'Republic-of-Moldova', 53: 'Portugal', 54: 'Ireland', 55: 'Denmark', 56: 'Iran',
57: 'Pakistan', 58: 'Israel', 59: 'Thailand', 61: 'Slovenia', 63: 'Croatia', 64: 'Chile', 65: 'Serbia',
66: 'Malaysia', 67: 'Philippines', 70: 'Estonia', 165: 'Egypt', 14: 'China', 77: 'Peru', 10: 'Italy',
71: 'Latvia', 72: 'Lithuania', 73: 'North-Korea', 74: 'Uruguay', 75: 'Paraguay', 76: 'Bolivia',
81: 'Republic-of-China-Taiwan', 166: 'United-Arab-Emirates', 167: 'Albania', 69: 'Bosnia-Herzegovina',
169: 'Armenia', 83: 'Belarus', 84: 'New-Zealand', 164: 'Saudi-Arabia', 170: 'Nigeria', }
class MyJSONEncoder(JSONEncoder):
def default(self, o):
@ -107,18 +119,22 @@ class MyJSONEncoder(JSONEncoder):
return super().default(o)
def now():
def now() -> datetime.datetime:
return datetime.datetime.now(erep_tz).replace(microsecond=0)
def localize_timestamp(timestamp: int):
def localize_timestamp(timestamp: int) -> datetime.datetime:
return datetime.datetime.fromtimestamp(timestamp, erep_tz)
def localize_dt(dt: Union[datetime.date, datetime.datetime]):
if isinstance(dt, datetime.date):
dt = datetime.datetime.combine(dt, datetime.time(0, 0, 0))
def localize_dt(dt: Union[datetime.date, datetime.datetime]) -> datetime.datetime:
try:
try:
return erep_tz.localize(dt)
except AttributeError:
return erep_tz.localize(datetime.datetime.combine(dt, datetime.time(0, 0, 0)))
except ValueError:
return dt.astimezone(erep_tz)
def good_timedelta(dt: datetime.datetime, td: datetime.timedelta) -> datetime.datetime:
@ -237,7 +253,7 @@ def write_request(response: requests.Response, is_error: bool = False):
"mimetype": "application/json" if ext == "json" else "text/html"}
def send_email(name: str, content: List[Any], player=None, local_vars: Dict[Any, Any] = None,
def send_email(name: str, content: List[Any], player=None, local_vars: Mapping[Any, Any] = None,
promo: bool = False, captcha: bool = False):
if local_vars is None:
local_vars = {}
@ -322,11 +338,11 @@ def process_error(log_info: str, name: str, exc_info: tuple, citizen=None, commi
send_email(name, bugtrace, citizen, local_vars=trace)
def report_promo(kind: str, time_untill: datetime.datetime):
def report_promo(kind: str, time_untill: datetime.datetime) -> NoReturn:
requests.post('https://api.erep.lv/promos/add/', data=dict(kind=kind, time_untill=time_untill))
def slugify(value, allow_unicode=False):
def slugify(value, allow_unicode=False) -> str:
"""
Function copied from Django2.2.1 django.utils.text.slugify
Convert to ASCII if 'allow_unicode' is False. Convert spaces to hyphens.