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

Black formatting

Browse Source 
Style config update

Makefile update

Test fix
This commit is contained in:
KEriks 2021-07-21 14:19:31 +03:00
parent 0b07ee8b8f
commit ade0f9b11a
11 changed files with 2169 additions and 1377 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Makefile
View File

@ -52,12 +52,13 @@ clean-test: ## remove test and coverage artifacts
rm -fr .pytest_cache
lint: ## check style with flake8
black erepublik tests
flake8 erepublik tests
test: ## run tests quickly with the default Python
python setup.py test
python -m unittest
coverage: ## check code coverage quickly with the default Python
coverage: lint ## check code coverage quickly with the default Python
coverage run --source erepublik setup.py test
coverage report -m
coverage html

6
erepublik/__init__.py
View File

@ -3,9 +3,9 @@
"""Top-level package for eRepublik script."""
__author__ = """Eriks Karls"""
__email__ = 'eriks@72.lv'
__version__ = '0.25.1.2'
__email__ = "eriks@72.lv"
__version__ = "0.25.1.2"
from erepublik.citizen import Citizen
__all__ = ['Citizen', '__version__']
__all__ = ["Citizen", "__version__"]

81
erepublik/_logging.py
View File

@ -19,13 +19,13 @@ from erepublik.utils import json, json_dumps, json_loads, slugify
class ErepublikFileHandler(handlers.TimedRotatingFileHandler):
_file_path: Path
def __init__(self, filename: str = 'log/erepublik.log', *args, **kwargs):
def __init__(self, filename: str = "log/erepublik.log", *args, **kwargs):
log_path = Path(filename)
self._file_path = log_path
log_path.parent.mkdir(parents=True, exist_ok=True)
at_time = erep_tz.localize(datetime.datetime.now()).replace(hour=0, minute=0, second=0, microsecond=0)
kwargs.update(atTime=at_time)
super().__init__(filename, when='d', *args, **kwargs)
super().__init__(filename, when="d", *args, **kwargs)
def doRollover(self) -> None:
self._file_path.parent.mkdir(parents=True, exist_ok=True)
@ -88,7 +88,7 @@ class ErepublikFormatter(logging.Formatter):
if datefmt:
s = dt.strftime(datefmt)
else:
s = dt.strftime('%Y-%m-%d %H:%M:%S')
s = dt.strftime("%Y-%m-%d %H:%M:%S")
return s
def usesTime(self):
@ -99,9 +99,9 @@ class ErepublikErrorHTTTPHandler(handlers.HTTPHandler):
def __init__(self, reporter: Reporter):
logging.Handler.__init__(self, level=logging.ERROR)
self._reporter = weakref.ref(reporter)
self.host = 'erep.lv'
self.url = '/ebot/error/'
self.method = 'POST'
self.host = "erep.lv"
self.url = "/ebot/error/"
self.method = "POST"
self.secure = True
self.credentials = (str(reporter.citizen_id), reporter.key)
self.context = None
@ -120,26 +120,34 @@ class ErepublikErrorHTTTPHandler(handlers.HTTPHandler):
try:
json_loads(html)
ext = 'json'
ext = "json"
except json.decoder.JSONDecodeError:
ext = 'html'
ext = "html"
try:
resp_time = datetime.datetime.strptime(
response.headers.get('date'), '%a, %d %b %Y %H:%M:%S %Z'
).replace(tzinfo=datetime.timezone.utc).astimezone(erep_tz).strftime('%F_%H-%M-%S')
except:
resp_time = slugify(response.headers.get('date'))
return dict(name=f"{resp_time}_{name}.{ext}", content=html.encode('utf-8'),
mimetype="application/json" if ext == 'json' else "text/html")
resp_time = (
datetime.datetime.strptime(response.headers.get("date"), "%a, %d %b %Y %H:%M:%S %Z")
.replace(tzinfo=datetime.timezone.utc)
.astimezone(erep_tz)
.strftime("%F_%H-%M-%S")
)
except: # noqa
resp_time = slugify(response.headers.get("date"))
return dict(
name=f"{resp_time}_{name}.{ext}", content=html.encode("utf-8"), mimetype="application/json" if ext == "json" else "text/html"
)
def _get_local_vars(self) -> str:
trace = inspect.trace()
local_vars = {}
if trace:
local_vars = trace[-1][0].f_locals
if local_vars.get('__name__') == '__main__':
local_vars.update(commit_id=local_vars.get('COMMIT_ID'), interactive=local_vars.get('INTERACTIVE'),
version=local_vars.get('__version__'), config=local_vars.get('CONFIG'))
if local_vars.get("__name__") == "__main__":
local_vars.update(
commit_id=local_vars.get("COMMIT_ID"),
interactive=local_vars.get("INTERACTIVE"),
version=local_vars.get("__version__"),
config=local_vars.get("CONFIG"),
)
else:
stack = inspect.stack()
report_error_caller_found = False
@ -147,18 +155,19 @@ class ErepublikErrorHTTTPHandler(handlers.HTTPHandler):
if report_error_caller_found:
local_vars = frame.frame.f_locals
break
if 'report_error' in str(frame.frame):
if "report_error" in str(frame.frame):
report_error_caller_found = True
if 'state_thread' in local_vars:
local_vars.pop('state_thread', None)
if "state_thread" in local_vars:
local_vars.pop("state_thread", None)
from erepublik import Citizen
if isinstance(local_vars.get('self'), Citizen):
local_vars['self'] = repr(local_vars['self'])
if isinstance(local_vars.get('player'), Citizen):
local_vars['player'] = repr(local_vars['player'])
if isinstance(local_vars.get('citizen'), Citizen):
local_vars['citizen'] = repr(local_vars['citizen'])
if isinstance(local_vars.get("self"), Citizen):
local_vars["self"] = repr(local_vars["self"])
if isinstance(local_vars.get("player"), Citizen):
local_vars["player"] = repr(local_vars["player"])
if isinstance(local_vars.get("citizen"), Citizen):
local_vars["citizen"] = repr(local_vars["citizen"])
return json_dumps(local_vars)
def _get_instance_json(self) -> str:
@ -171,15 +180,17 @@ class ErepublikErrorHTTTPHandler(handlers.HTTPHandler):
# Log last response
resp = self._get_last_response()
files = [('file', (resp.get('name'), resp.get('content'), resp.get('mimetype'))), ]
files = [
("file", (resp.get("name"), resp.get("content"), resp.get("mimetype"))),
]
files += list(('file', (f, open(f'log/{f}', 'rb'))) for f in os.listdir('log') if f.endswith('.log'))
files += list(("file", (f, open(f"log/{f}", "rb"))) for f in os.listdir("log") if f.endswith(".log"))
local_vars_json = self._get_local_vars()
if local_vars_json:
files.append(('file', ('local_vars.json', local_vars_json, "application/json")))
files.append(("file", ("local_vars.json", local_vars_json, "application/json")))
instance_json = self._get_instance_json()
if instance_json:
files.append(('file', ('instance.json', instance_json, "application/json")))
files.append(("file", ("instance.json", instance_json, "application/json")))
data.update(files=files)
return data
@ -190,12 +201,12 @@ class ErepublikErrorHTTTPHandler(handlers.HTTPHandler):
Send the record to the Web server as a percent-encoded dictionary
"""
try:
proto = 'https' if self.secure else 'http'
proto = "https" if self.secure else "http"
u, p = self.credentials
s = 'Basic ' + base64.b64encode(f'{u}:{p}'.encode('utf-8')).strip().decode('ascii')
headers = {'Authorization': s}
s = "Basic " + base64.b64encode(f"{u}:{p}".encode("utf-8")).strip().decode("ascii")
headers = {"Authorization": s}
data = self.mapLogRecord(record)
files = data.pop('files') if 'files' in data else None
files = data.pop("files") if "files" in data else None
requests.post(f"{proto}://{self.host}{self.url}", headers=headers, data=data, files=files)
except Exception:
self.handleError(record)

412
erepublik/access_points.py
View File

@ -9,7 +9,7 @@ from requests_toolbelt.utils import dump
from erepublik import constants, utils
__all__ = ['SlowRequests', 'CitizenAPI']
__all__ = ["SlowRequests", "CitizenAPI"]
class SlowRequests(Session):
@ -25,13 +25,20 @@ class SlowRequests(Session):
user_agent = random.choice(self.get_random_user_agent())
self.request_log_name = utils.get_file(utils.now().strftime("debug/requests_%Y-%m-%d.log"))
self.last_time = utils.now()
self.headers.update({'User-Agent': user_agent})
self.headers.update({"User-Agent": user_agent})
self.hooks["response"] = [self._log_response]
@property
def as_dict(self):
return dict(last_time=self.last_time, timeout=self.timeout, cookies=self.cookies.get_dict(), debug=self.debug,
user_agent=self.headers['User-Agent'], request_log_name=self.request_log_name, proxies=self.proxies)
return dict(
last_time=self.last_time,
timeout=self.timeout,
cookies=self.cookies.get_dict(),
debug=self.debug,
user_agent=self.headers["User-Agent"],
request_log_name=self.request_log_name,
proxies=self.proxies,
)
def request(self, method, url, *args, **kwargs):
self._slow_down_requests()
@ -50,70 +57,72 @@ class SlowRequests(Session):
def _log_request(self, url, method, data=None, json=None, params=None, **kwargs):
if self.debug:
args = {}
kwargs.pop('allow_redirects', None)
kwargs.pop("allow_redirects", None)
if kwargs:
args.update({'kwargs': kwargs})
args.update({"kwargs": kwargs})
if data:
args.update({'data': data})
args.update({"data": data})
if json:
args.update({'json': json})
args.update({"json": json})
if params:
args.update({'params': params})
args.update({"params": params})
body = f"[{utils.now().strftime('%F %T')}]\tURL: '{url}'\tMETHOD: {method}\tARGS: {args}\n"
with open(self.request_log_name, 'ab') as file:
with open(self.request_log_name, "ab") as file:
file.write(body.encode("UTF-8"))
def _log_response(self, response: Response, *args, **kwargs):
redirect = kwargs.get('redirect')
redirect = kwargs.get("redirect")
url = response.request.url
if self.debug:
if response.history and not redirect:
for hist_resp in response.history:
self._log_request(hist_resp.request.url, 'REDIRECT')
self._log_request(hist_resp.request.url, "REDIRECT")
self._log_response(hist_resp, redirect=True)
fd_path = 'debug/requests'
fd_time = self.last_time.strftime('%Y/%m/%d/%H-%M-%S')
fd_path = "debug/requests"
fd_time = self.last_time.strftime("%Y/%m/%d/%H-%M-%S")
fd_name = utils.slugify(url[len(CitizenBaseAPI.url) :])
fd_extra = '_REDIRECT' if redirect else ""
fd_extra = "_REDIRECT" if redirect else ""
try:
utils.json.loads(response.text)
fd_ext = 'json'
fd_ext = "json"
except utils.json.JSONDecodeError:
fd_ext = 'html'
fd_ext = "html"
filename = f'{fd_path}/{fd_time}_{fd_name}{fd_extra}.{fd_ext}'
filename = f"{fd_path}/{fd_time}_{fd_name}{fd_extra}.{fd_ext}"
utils.write_file(filename, response.text)
if not redirect:
data = dump.dump_all(response)
utils.write_file(f'debug/dumps/{fd_time}_{fd_name}{fd_extra}.{fd_ext}.dump', data.decode('utf8'))
utils.write_file(f"debug/dumps/{fd_time}_{fd_name}{fd_extra}.{fd_ext}.dump", data.decode("utf8"))
@staticmethod
def get_random_user_agent() -> str:
windows_x64 = 'Windows NT 10.0; Win64; x64'
linux_x64 = 'X11; Linux x86_64'
android_11 = 'Android 11; Mobile'
android_10 = 'Android 10; Mobile'
android_9 = 'Android 9; Mobile'
windows_x64 = "Windows NT 10.0; Win64; x64"
linux_x64 = "X11; Linux x86_64"
android_11 = "Android 11; Mobile"
android_10 = "Android 10; Mobile"
android_9 = "Android 9; Mobile"
firefox_tmplt = 'Mozilla/5.0 ({osystem}; rv:{version}.0) Gecko/20100101 Firefox/{version}.0'
firefox_tmplt = "Mozilla/5.0 ({osystem}; rv:{version}.0) Gecko/20100101 Firefox/{version}.0"
ff_version = range(85, 92)
chrome_tmplt = 'Mozilla/5.0 ({osystem}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36'
chrome_version = ['85.0.4183.121',
'86.0.4240.183',
'87.0.4280.141',
'88.0.4324.182',
'89.0.4389.128',
'90.0.4430.18',
'91.0.4472.73',
'92.0.4515.14']
chrome_tmplt = "Mozilla/5.0 ({osystem}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36"
chrome_version = [
"85.0.4183.121",
"86.0.4240.183",
"87.0.4280.141",
"88.0.4324.182",
"89.0.4389.128",
"90.0.4430.18",
"91.0.4472.73",
"92.0.4515.14",
]
uas = []
for osystem in [windows_x64, linux_x64, android_9, android_10, android_11]:
@ -149,75 +158,90 @@ class CitizenBaseAPI:
return self.get(self.url)
def set_socks_proxy(self, host: str, port: int, username: str = None, password: str = None):
url = f'socks5://{username}:{password}@{host}:{port}' if username and password else f'socks5://{host}:{port}'
url = f"socks5://{username}:{password}@{host}:{port}" if username and password else f"socks5://{host}:{port}"
self._req.proxies = dict(http=url, https=url)
def set_http_proxy(self, host: str, port: int, username: str = None, password: str = None):
url = f'http://{username}:{password}@{host}:{port}' if username and password else f'http://{host}:{port}'
url = f"http://{username}:{password}@{host}:{port}" if username and password else f"http://{host}:{port}"
self._req.proxies = dict(http=url)
def _get_main_session_captcha(self) -> Response:
return self.get(f'{self.url}/main/sessionCaptcha')
return self.get(f"{self.url}/main/sessionCaptcha")
def _get_main_session_unlock_popup(self) -> Response:
return self.get(f'{self.url}/main/sessionUnlockPopup')
return self.get(f"{self.url}/main/sessionUnlockPopup")
def _post_main_session_get_challenge(self, captcha_id: int, image_id: str = "") -> Response:
c = [cookie.name for cookie in self._req.cookies if not cookie.has_nonstandard_attr('HttpOnly')]
env = dict(l=['tets'], s=[], c=c, m=0)
c = [cookie.name for cookie in self._req.cookies if not cookie.has_nonstandard_attr("HttpOnly")]
env = dict(l=["tets"], s=[], c=c, m=0)
data = dict(_token=self.token, captchaId=captcha_id, env=utils.b64json(env))
if image_id:
data.update(imageId=image_id, isRefresh=True)
return self.post(f'{self.url}/main/sessionGetChallenge', data=data)
return self.post(f"{self.url}/main/sessionGetChallenge", data=data)
def _post_main_session_unlock(
self, captcha_id: int, image_id: str, challenge_id: str, coords: List[Dict[str, int]], src: str
) -> Response:
c = [cookie.name for cookie in self._req.cookies if not cookie.has_nonstandard_attr('HttpOnly')]
env = dict(l=['tets'], s=[], c=c, m=0)
cookies = dict(sh=hashlib.sha256(','.join(env['l']+env['s']).encode('utf8')).hexdigest(),
ch=hashlib.sha256(','.join(env['c']).encode('utf8')).hexdigest())
cookie_kwargs = dict(expires=int(time.time())+120, path="/en/main/sessionUnlock", domain='.www.erepublik.com',
secure=True, rest={'HttpOnly': True})
self._req.cookies.set('sh', cookies['sh'], **cookie_kwargs)
self._req.cookies.set('ch', cookies['ch'], **cookie_kwargs)
c = [cookie.name for cookie in self._req.cookies if not cookie.has_nonstandard_attr("HttpOnly")]
env = dict(l=["tets"], s=[], c=c, m=0)
cookies = dict(
sh=hashlib.sha256(",".join(env["l"] + env["s"]).encode("utf8")).hexdigest(),
ch=hashlib.sha256(",".join(env["c"]).encode("utf8")).hexdigest(),
)
cookie_kwargs = dict(
expires=int(time.time()) + 120, path="/en/main/sessionUnlock", domain=".www.erepublik.com", secure=True, rest={"HttpOnly": True}
)
self._req.cookies.set("sh", cookies["sh"], **cookie_kwargs)
self._req.cookies.set("ch", cookies["ch"], **cookie_kwargs)
b64_env = utils.b64json(env)
data = dict(_token=self.token, captchaId=captcha_id, imageId=image_id, challengeId=challenge_id,
clickMatrix=utils.json_dumps(coords).replace(' ', ''), isMobile=0, env=b64_env, src=src)
return self.post(f'{self.url}/main/sessionUnlock', data=data, json=data,
headers={'X-Requested-With': 'XMLHttpRequest', 'Referer': 'https://www.erepublik.com/en'})
data = dict(
_token=self.token,
captchaId=captcha_id,
imageId=image_id,
challengeId=challenge_id,
clickMatrix=utils.json_dumps(coords).replace(" ", ""),
isMobile=0,
env=b64_env,
src=src,
)
return self.post(
f"{self.url}/main/sessionUnlock",
data=data,
json=data,
headers={"X-Requested-With": "XMLHttpRequest", "Referer": "https://www.erepublik.com/en"},
)
def _post_energy_refill_get_inventory(self):
return self.post(f'{self.url}/economy/energyRefill-getInventory', data={'_token': self.token})
return self.post(f"{self.url}/economy/energyRefill-getInventory", data={"_token": self.token})
class ErepublikAnniversaryAPI(CitizenBaseAPI):
def _post_main_collect_anniversary_reward(self) -> Response:
return self.post(f"{self.url}/main/collect-anniversary-reward", data={'_token': self.token})
return self.post(f"{self.url}/main/collect-anniversary-reward", data={"_token": self.token})
# 12th anniversary endpoints
def _get_anniversary_quest_data(self) -> Response:
return self.get(f"{self.url}/main/anniversaryQuestData")
def _post_map_rewards_unlock(self, node_id: int) -> Response:
data = {'nodeId': node_id, '_token': self.token}
data = {"nodeId": node_id, "_token": self.token}
return self.post(f"{self.url}/main/map-rewards-unlock", data=data)
def _post_map_rewards_speedup(self, node_id: int, currency_amount: int) -> Response:
data = {'nodeId': node_id, '_token': self.token, 'currencyCost': currency_amount}
data = {"nodeId": node_id, "_token": self.token, "currencyCost": currency_amount}
return self.post(f"{self.url}/main/map-rewards-speedup", data=data)
def _post_map_rewards_claim(self, node_id: int, extra: bool = False) -> Response:
data = {'nodeId': node_id, '_token': self.token}
data = {"nodeId": node_id, "_token": self.token}
if extra:
data['claimExtra'] = 1
data["claimExtra"] = 1
return self.post(f"{self.url}/main/map-rewards-claim", data=data)
def _post_main_wheel_of_fortune_spin(self, cost) -> Response:
return self.post(f"{self.url}/main/wheeloffortune-spin", data={'_token': self.token, '_currentCost': cost})
return self.post(f"{self.url}/main/wheeloffortune-spin", data={"_token": self.token, "_currentCost": cost})
def _post_main_wheel_of_fortune_build(self) -> Response:
return self.post(f"{self.url}/main/wheeloffortune-build", data={'_token': self.token})
return self.post(f"{self.url}/main/wheeloffortune-build", data={"_token": self.token})
class ErepublikArticleAPI(CitizenBaseAPI):
@ -230,13 +254,13 @@ class ErepublikArticleAPI(CitizenBaseAPI):
def _post_main_article_comments(self, article_id: int, page: int = 1) -> Response:
data = dict(_token=self.token, articleId=article_id, page=page)
if page:
data.update({'page': page})
data.update({"page": page})
return self.post(f"{self.url}/main/articleComments", data=data)
def _post_main_article_comments_create(self, message: str, article_id: int, parent: int = 0) -> Response:
data = dict(_token=self.token, message=message, articleId=article_id)
if parent:
data.update({'parentId': parent})
data.update({"parentId": parent})
return self.post(f"{self.url}/main/articleComments/create", data=data)
def _post_main_donate_article(self, article_id: int, amount: int) -> Response:
@ -244,8 +268,7 @@ class ErepublikArticleAPI(CitizenBaseAPI):
return self.post(f"{self.url}/main/donate-article", data=data)
def _post_main_write_article(self, title: str, content: str, country_id: int, kind_id: int) -> Response:
data = dict(_token=self.token, article_name=title, article_body=content, article_location=country_id,
article_category=kind_id)
data = dict(_token=self.token, article_name=title, article_body=content, article_location=country_id, article_category=kind_id)
return self.post(f"{self.url}/main/write-article", data=data)
def _post_main_vote_article(self, article_id: int) -> Response:
@ -255,13 +278,12 @@ class ErepublikArticleAPI(CitizenBaseAPI):
class ErepublikCompanyAPI(CitizenBaseAPI):
def _post_economy_assign_to_holding(self, factory_id: int, holding_id: int) -> Response:
data = dict(_token=self.token, factoryId=factory_id, action='assign', holdingCompanyId=holding_id)
data = dict(_token=self.token, factoryId=factory_id, action="assign", holdingCompanyId=holding_id)
return self.post(f"{self.url}/economy/assign-to-holding", data=data)
def _post_economy_create_company(self, industry_id: int, building_type: int = 1) -> Response:
data = {'_token': self.token, "company[industry_id]": industry_id, "company[building_type]": building_type}
return self.post(f"{self.url}/economy/create-company", data=data,
headers={'Referer': f"{self.url}/economy/create-company"})
data = {"_token": self.token, "company[industry_id]": industry_id, "company[building_type]": building_type}
return self.post(f"{self.url}/economy/create-company", data=data, headers={"Referer": f"{self.url}/economy/create-company"})
def _get_economy_inventory_items(self) -> Response:
return self.get(f"{self.url}/economy/inventory-items/")
@ -278,39 +300,43 @@ class ErepublikCompanyAPI(CitizenBaseAPI):
data["grounds[%i][id]" % idx] = tg_id
data["grounds[%i][train]" % idx] = 1
if data:
data['_token'] = self.token
data["_token"] = self.token
return self.post(f"{self.url}/economy/train", data=data)
def _post_economy_upgrade_company(self, factory: int, level: int, pin: str = None) -> Response:
data = dict(_token=self.token, type='upgrade', companyId=factory, level=level, pin="" if pin is None else pin)
data = dict(_token=self.token, type="upgrade", companyId=factory, level=level, pin="" if pin is None else pin)
return self.post(f"{self.url}/economy/upgrade-company", data=data)
def _post_economy_work(self, action_type: str, wam: List[int] = None, employ: Dict[int, int] = None) -> Response:
data: Dict[str, Union[int, str]] = dict(action_type=action_type, _token=self.token)
if action_type == 'production':
if action_type == "production":
if employ is None:
employ = {}
if wam is None:
wam = []
max_idx = 0
for company_id in sorted(wam or []):
data.update({
data.update(
{
f"companies[{max_idx}][id]": company_id,
f"companies[{max_idx}][employee_works]": employ.pop(company_id, 0),
f"companies[{max_idx}][own_work]": 1
})
f"companies[{max_idx}][own_work]": 1,
}
)
max_idx += 1
for company_id in sorted(employ or []):
data.update({
data.update(
{
f"companies[{max_idx}][id]": company_id,
f"companies[{max_idx}][employee_works]": employ.pop(company_id, 0),
f"companies[{max_idx}][own_work]": 0
})
f"companies[{max_idx}][own_work]": 0,
}
)
max_idx += 1
return self.post(f"{self.url}/economy/work", data=data)
def _post_economy_work_overtime(self) -> Response:
data = dict(action_type='workOvertime', _token=self.token)
data = dict(action_type="workOvertime", _token=self.token)
return self.post(f"{self.url}/economy/workOvertime", data=data)
def _post_economy_job_market_apply(self, citizen_id: int, salary: float) -> Response:
@ -318,29 +344,28 @@ class ErepublikCompanyAPI(CitizenBaseAPI):
return self.post(f"{self.url}/economy/job-market-apply", data=data)
def _post_economy_resign(self) -> Response:
return self.post(f"{self.url}/economy/resign",
return self.post(
f"{self.url}/economy/resign",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data={'_token': self.token, 'action_type': 'resign'})
data={"_token": self.token, "action_type": "resign"},
)
def _post_economy_sell_company(self, factory_id: int, pin: int = None, sell: bool = True) -> Response:
data = dict(_token=self.token, pin="" if pin is None else pin)
if sell:
data.update({'sell': 'sell'})
data.update({"sell": "sell"})
else:
data.update({'dissolve': factory_id})
return self.post(f"{self.url}/economy/sell-company/{factory_id}",
data=data, headers={'Referer': self.url})
data.update({"dissolve": factory_id})
return self.post(f"{self.url}/economy/sell-company/{factory_id}", data=data, headers={"Referer": self.url})
class ErepublikCountryAPI(CitizenBaseAPI):
def _get_country_military(self, country_name: str) -> Response:
return self.get(f"{self.url}/country/military/{country_name}")
def _post_main_country_donate(self, country_id: int, action: str, value: Union[int, float],
quality: int = None) -> Response:
def _post_main_country_donate(self, country_id: int, action: str, value: Union[int, float], quality: int = None) -> Response:
data = dict(countryId=country_id, action=action, _token=self.token, value=value, quality=quality)
return self.post(f"{self.url}/main/country-donate", data=data,
headers={'Referer': f"{self.url}/country/economy/Latvia"})
return self.post(f"{self.url}/main/country-donate", data=data, headers={"Referer": f"{self.url}/country/economy/Latvia"})
class ErepublikEconomyAPI(CitizenBaseAPI):
@ -362,20 +387,20 @@ class ErepublikEconomyAPI(CitizenBaseAPI):
return self.post(f"{self.url}/economy/activateBooster", data=data)
def _post_economy_activate_house(self, quality: int) -> Response:
data = dict(action='activate', quality=quality, type='house', _token=self.token)
data = dict(action="activate", quality=quality, type="house", _token=self.token)
return self.post(f"{self.url}/economy/activateHouse", data=data)
def _post_economy_donate_items_action(self, citizen_id: int, amount: int, industry: int,
quality: int) -> Response:
def _post_economy_donate_items_action(self, citizen_id: int, amount: int, industry: int, quality: int) -> Response:
data = dict(citizen_id=citizen_id, amount=amount, industry_id=industry, quality=quality, _token=self.token)
return self.post(f"{self.url}/economy/donate-items-action", data=data,
headers={'Referer': f"{self.url}/economy/donate-items/{citizen_id}"})
return self.post(
f"{self.url}/economy/donate-items-action", data=data, headers={"Referer": f"{self.url}/economy/donate-items/{citizen_id}"}
)
def _post_economy_donate_money_action(self, citizen_id: int, amount: float = 0.0,
currency: int = 62) -> Response:
def _post_economy_donate_money_action(self, citizen_id: int, amount: float = 0.0, currency: int = 62) -> Response:
data = dict(citizen_id=citizen_id, _token=self.token, currency_id=currency, amount=amount)
return self.post(f"{self.url}/economy/donate-money-action", data=data,
headers={'Referer': f"{self.url}/economy/donate-money/{citizen_id}"})
return self.post(
f"{self.url}/economy/donate-money-action", data=data, headers={"Referer": f"{self.url}/economy/donate-money/{citizen_id}"}
)
def _post_economy_exchange_purchase(self, amount: float, currency: int, offer: int) -> Response:
data = dict(_token=self.token, amount=amount, currencyId=currency, offerId=offer)
@ -386,46 +411,56 @@ class ErepublikEconomyAPI(CitizenBaseAPI):
return self.post(f"{self.url}/economy/exchange/retrieve/", data=data)
def _post_economy_game_tokens_market(self, action: str) -> Response:
assert action in ['retrieve', ]
assert action in [
"retrieve",
]
data = dict(_token=self.token, action=action)
return self.post(f"{self.url}/economy/gameTokensMarketAjax", data=data)
def _post_economy_marketplace(self, country: int, industry: int, quality: int,
order_asc: bool = True) -> Response:
data = dict(countryId=country, industryId=industry, quality=quality, ajaxMarket=1,
orderBy='price_asc' if order_asc else 'price_desc', _token=self.token)
def _post_economy_marketplace(self, country: int, industry: int, quality: int, order_asc: bool = True) -> Response:
data = dict(
countryId=country,
industryId=industry,
quality=quality,
ajaxMarket=1,
orderBy="price_asc" if order_asc else "price_desc",
_token=self.token,
)
return self.post(f"{self.url}/economy/marketplaceAjax", data=data)
def _post_economy_marketplace_actions(self, action: str, **kwargs) -> Response:
if action == 'buy':
data = dict(_token=self.token, offerId=kwargs['offer'], amount=kwargs['amount'],
orderBy='price_asc', currentPage=1, buyAction=1)
elif action == 'sell':
data = dict(_token=self.token, countryId=kwargs['country_id'], price=kwargs['price'],
industryId=kwargs['industry'], quality=kwargs['quality'], amount=kwargs['amount'],
sellAction='postOffer')
elif action == 'delete':
data = dict(_token=self.token, offerId=kwargs['offer_id'], sellAction='deleteOffer')
if action == "buy":
data = dict(
_token=self.token, offerId=kwargs["offer"], amount=kwargs["amount"], orderBy="price_asc", currentPage=1, buyAction=1
)
elif action == "sell":
data = dict(
_token=self.token,
countryId=kwargs["country_id"],
price=kwargs["price"],
industryId=kwargs["industry"],
quality=kwargs["quality"],
amount=kwargs["amount"],
sellAction="postOffer",
)
elif action == "delete":
data = dict(_token=self.token, offerId=kwargs["offer_id"], sellAction="deleteOffer")
else:
raise ValueError(f"Action '{action}' is not supported! Only 'buy/sell/delete' actions are available")
return self.post(f"{self.url}/economy/marketplaceActions", data=data)
class ErepublikLeaderBoardAPI(CitizenBaseAPI):
def _get_main_leaderboards_damage_aircraft_rankings(self, country_id: int, weeks: int = 0,
mu_id: int = 0) -> Response: # noqa
def _get_main_leaderboards_damage_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-damage-aircraft-rankings/{country_id}/{weeks}/{mu_id}/0")
def _get_main_leaderboards_damage_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0,
div: int = 0) -> Response: # noqa
def _get_main_leaderboards_damage_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-damage-rankings/{country_id}/{weeks}/{mu_id}/{div}")
def _get_main_leaderboards_kills_aircraft_rankings(self, country_id: int, weeks: int = 0,
mu_id: int = 0) -> Response: # noqa
def _get_main_leaderboards_kills_aircraft_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-kills-aircraft-rankings/{country_id}/{weeks}/{mu_id}/0")
def _get_main_leaderboards_kills_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0,
div: int = 0) -> Response: # noqa
def _get_main_leaderboards_kills_rankings(self, country_id: int, weeks: int = 0, mu_id: int = 0, div: int = 0) -> Response: # noqa
return self.get(f"{self.url}/main/leaderboards-kills-rankings/{country_id}/{weeks}/{mu_id}/{div}")
@ -433,7 +468,7 @@ class ErepublikLocationAPI(CitizenBaseAPI):
def _get_main_city_data_residents(self, city_id: int, page: int = 1, params: Mapping[str, Any] = None) -> Response:
if params is None:
params = {}
return self.get(f"{self.url}/main/city-data/{city_id}/residents", params={'currentPage': page, **params})
return self.get(f"{self.url}/main/city-data/{city_id}/residents", params={"currentPage": page, **params})
class ErepublikMilitaryAPI(CitizenBaseAPI):
@ -444,7 +479,7 @@ class ErepublikMilitaryAPI(CitizenBaseAPI):
return self.get(f"{self.url}/military/battlefield-choose-side/{battle_id}/{side_id}")
def _get_military_show_weapons(self, battle_id: int) -> Response:
return self.get(f"{self.url}/military/show-weapons", params={'_token': self.token, 'battleId': battle_id})
return self.get(f"{self.url}/military/show-weapons", params={"_token": self.token, "battleId": battle_id})
def _get_military_campaigns(self) -> Response:
return self.get(f"{self.url}/military/campaigns-new/")
@ -456,7 +491,7 @@ class ErepublikMilitaryAPI(CitizenBaseAPI):
return self.get(f"{self.url}/military/campaignsJson/citizen")
def _get_military_unit_data(self, unit_id: int, **kwargs) -> Response:
params = {'groupId': unit_id, 'panel': 'members', **kwargs}
params = {"groupId": unit_id, "panel": "members", **kwargs}
return self.get(f"{self.url}/military/military-unit-data/", params=params)
def _post_main_activate_battle_effect(self, battle_id: int, kind: str, citizen_id: int) -> Response:
@ -480,22 +515,32 @@ class ErepublikMilitaryAPI(CitizenBaseAPI):
data = dict(type=kind, quality=quality, duration=duration, battleId=battle_id, _token=self.token)
return self.post(f"{self.url}/military/fight-activateBooster", data=data)
def _post_military_change_weapon(self, battle_id: int, battle_zone: int, weapon_level: int, ) -> Response:
def _post_military_change_weapon(
self,
battle_id: int,
battle_zone: int,
weapon_level: int,
) -> Response:
data = dict(battleId=battle_id, _token=self.token, battleZoneId=battle_zone, customizationLevel=weapon_level)
return self.post(f"{self.url}/military/change-weapon", 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':
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(f"{self.url}/military/battle-console", data=data)
def _post_military_deploy_bomb(self, battle_id: int, division_id: int, side_id: int, bomb_id: int) -> Response:
data = dict(battleId=battle_id, battleZoneId=division_id, sideId=side_id, sideCountryId=side_id,
bombId=bomb_id, _token=self.token)
data = dict(battleId=battle_id, battleZoneId=division_id, sideId=side_id, sideCountryId=side_id, bombId=bomb_id, _token=self.token)
return self.post(f"{self.url}/military/deploy-bomb", data=data)
def _post_military_fight_air(self, battle_id: int, side_id: int, zone_id: int) -> Response:
@ -517,8 +562,15 @@ class ErepublikMilitaryAPI(CitizenBaseAPI):
def _post_fight_deploy_start_deploy(
self, battle_id: int, side_id: int, battle_zone_id: int, energy: int, weapon: int, **kwargs
) -> Response:
data = dict(_token=self.token, battleId=battle_id, battleZoneId=battle_zone_id, sideCountryId=side_id,
weaponQuality=weapon, totalEnergy=energy, **kwargs)
data = dict(
_token=self.token,
battleId=battle_id,
battleZoneId=battle_zone_id,
sideCountryId=side_id,
weaponQuality=weapon,
totalEnergy=energy,
**kwargs,
)
return self.post(f"{self.url}/military/fightDeploy-startDeploy", data=data)
def _post_military_fight_deploy_deploy_report_data(self, deployment_id: int) -> Response:
@ -544,8 +596,7 @@ class ErepublikPoliticsAPI(CitizenBaseAPI):
return self.get(f"{self.url}/main/presidential-elections/{country_id}/{timestamp}")
def _post_propose_president_candidate(self, party_slug: str, citizen_id: int) -> Response:
return self.post(f"{self.url}/propose-president-candidate/{party_slug}",
data=dict(_token=self.token, citizen=citizen_id))
return self.post(f"{self.url}/propose-president-candidate/{party_slug}", data=dict(_token=self.token, citizen=citizen_id))
def _get_auto_propose_president_candidate(self, party_slug: str) -> Response:
return self.get(f"{self.url}/auto-propose-president-candidate/{party_slug}")
@ -553,17 +604,15 @@ class ErepublikPoliticsAPI(CitizenBaseAPI):
class ErepublikPresidentAPI(CitizenBaseAPI):
def _post_wars_attack_region(self, war_id: int, region_id: int, region_name: str) -> Response:
data = {'_token': self.token, 'warId': war_id, 'regionName': region_name, 'regionNameConfirm': region_name}
return self.post(f'{self.url}/wars/attack-region/{war_id}/{region_id}', data=data)
data = {"_token": self.token, "warId": war_id, "regionName": region_name, "regionNameConfirm": region_name}
return self.post(f"{self.url}/wars/attack-region/{war_id}/{region_id}", data=data)
def _post_new_war(self, self_country_id: int, attack_country_id: int, debate: str = "") -> Response:
data = dict(requirments=1, _token=self.token, debate=debate,
countryNameConfirm=constants.COUNTRIES[attack_country_id].link)
data = dict(requirments=1, _token=self.token, debate=debate, countryNameConfirm=constants.COUNTRIES[attack_country_id].link)
return self.post(f"{self.url}/{constants.COUNTRIES[self_country_id].link}/new-war", data=data)
def _post_new_donation(self, country_id: int, amount: int, org_name: str, debate: str = "") -> Response:
data = dict(requirments=1, _token=self.token, debate=debate, currency=1, value=amount, commit='Propose',
type_name=org_name)
data = dict(requirments=1, _token=self.token, debate=debate, currency=1, value=amount, commit="Propose", type_name=org_name)
return self.post(f"{self.url}/{constants.COUNTRIES[country_id].link}/new-donation", data=data)
@ -584,10 +633,10 @@ class ErepublikProfileAPI(CitizenBaseAPI):
return self.get(f"{self.url}/main/messages-paginated/{page}")
def _get_main_money_donation_accept(self, donation_id: int) -> Response:
return self.get(f"{self.url}/main/money-donation/accept/{donation_id}", params={'_token': self.token})
return self.get(f"{self.url}/main/money-donation/accept/{donation_id}", params={"_token": self.token})
def _get_main_money_donation_reject(self, donation_id: int) -> Response:
return self.get(f"{self.url}/main/money-donation/reject/{donation_id}", params={'_token': self.token})
return self.get(f"{self.url}/main/money-donation/reject/{donation_id}", params={"_token": self.token})
def _get_main_notifications_ajax_community(self, page: int = 1) -> Response:
return self.get(f"{self.url}/main/notificationsAjax/community/{page}")
@ -607,16 +656,16 @@ class ErepublikProfileAPI(CitizenBaseAPI):
def _post_main_citizen_add_remove_friend(self, citizen: int, add: bool) -> Response:
data = dict(_token=self.token, citizenId=citizen, url="//www.erepublik.com/en/main/citizen-addRemoveFriend")
if add:
data.update({'action': 'addFriend'})
data.update({"action": "addFriend"})
else:
data.update({'action': 'removeFriend'})
data.update({"action": "removeFriend"})
return self.post(f"{self.url}/main/citizen-addRemoveFriend", data=data)
def _post_main_daily_task_reward(self) -> Response:
return self.post(f"{self.url}/main/daily-tasks-reward", data=dict(_token=self.token))
def _post_delete_message(self, msg_id: list) -> Response:
data = {'_token': self.token, "delete_message[]": msg_id}
data = {"_token": self.token, "delete_message[]": msg_id}
return self.post(f"{self.url}/main/messages-delete", data)
def _post_eat(self, color: str) -> Response:
@ -628,37 +677,36 @@ class ErepublikProfileAPI(CitizenBaseAPI):
return self.post(f"{self.url}/main/global-alerts/close", data=data)
def _post_forgot_password(self, email: str) -> Response:
data = dict(_token=self.token, email=email, commit='Reset password')
data = dict(_token=self.token, email=email, commit="Reset password")
return self.post(f"{self.url}/forgot-password", data=data)
def _post_login(self, email: str, password: str) -> Response:
data = dict(csrf_token=self.token, citizen_email=email, citizen_password=password, remember='on')
data = dict(csrf_token=self.token, citizen_email=email, citizen_password=password, remember="on")
return self.post(f"{self.url}/login", data=data)
def _post_main_messages_alert(self, notification_ids: List[int]) -> Response:
data = {'_token': self.token, "delete_alerts[]": notification_ids, 'deleteAllAlerts': '1', 'delete': 'Delete'}
data = {"_token": self.token, "delete_alerts[]": notification_ids, "deleteAllAlerts": "1", "delete": "Delete"}
return self.post(f"{self.url}/main/messages-alerts/1", data=data)
def _post_main_notifications_ajax_community(self, notification_ids: List[int], page: int = 1) -> Response:
data = {'_token': self.token, "delete_alerts[]": notification_ids}
data = {"_token": self.token, "delete_alerts[]": notification_ids}
return self.post(f"{self.url}/main/notificationsAjax/community/{page}", data=data)
def _post_main_notifications_ajax_system(self, notification_ids: List[int], page: int = 1) -> Response:
data = {'_token': self.token, "delete_alerts[]": notification_ids}
data = {"_token": self.token, "delete_alerts[]": notification_ids}
return self.post(f"{self.url}/main/notificationsAjax/system/{page}", data=data)
def _post_main_notifications_ajax_report(self, notification_ids: List[int], page: int = 1) -> Response:
data = {'_token': self.token, "delete_alerts[]": notification_ids}
data = {"_token": self.token, "delete_alerts[]": notification_ids}
return self.post(f"{self.url}/main/notificationsAjax/report/{page}", data=data)
def _post_main_messages_compose(self, subject: str, body: str, citizens: List[int]) -> Response:
url_pk = 0 if len(citizens) > 1 else str(citizens[0])
data = dict(citizen_name=",".join([str(x) for x in citizens]),
citizen_subject=subject, _token=self.token, citizen_message=body)
data = dict(citizen_name=",".join([str(x) for x in citizens]), citizen_subject=subject, _token=self.token, citizen_message=body)
return self.post(f"{self.url}/main/messages-compose/{url_pk}", data=data)
def _post_military_group_missions(self) -> Response:
data = dict(action='check', _token=self.token)
data = dict(action="check", _token=self.token)
return self.post(f"{self.url}/military/group-missions", data=data)
def _post_main_weekly_challenge_reward(self, reward_id: int) -> Response:
@ -670,7 +718,7 @@ class ErepublikProfileAPI(CitizenBaseAPI):
return self.post(f"{self.url}/main/weekly-challenge-collect-all", data=data)
def _post_main_profile_update(self, action: str, params: str):
data = {'action': action, 'params': params, '_token': self.token}
data = {"action": action, "params": params, "_token": self.token}
return self.post(f"{self.url}/main/profile-update", data=data)
@ -687,85 +735,95 @@ class ErepublikWallPostAPI(CitizenBaseAPI):
# ## Country
def _post_main_country_comment_retrieve(self, post_id: int) -> Response:
data = {'_token': self.token, 'postId': post_id}
data = {"_token": self.token, "postId": post_id}
return self.post(f"{self.url}/main/country-comment/retrieve/json", data=data)
def _post_main_country_comment_create(self, post_id: int, comment_message: str) -> Response:
data = {'_token': self.token, 'postId': post_id, 'comment_message': comment_message}
data = {"_token": self.token, "postId": post_id, "comment_message": comment_message}
return self.post(f"{self.url}/main/country-comment/create/json", data=data)
def _post_main_country_post_create(self, body: str, post_as: int) -> Response:
data = {'_token': self.token, 'post_message': body, 'post_as': post_as}
data = {"_token": self.token, "post_message": body, "post_as": post_as}
return self.post(f"{self.url}/main/country-post/create/json", data=data)
def _post_main_country_post_retrieve(self) -> Response:
data = {'_token': self.token, 'page': 1, 'switchedFrom': False}
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post(f"{self.url}/main/country-post/retrieve/json", data=data)
# ## Military Unit
def _post_main_military_unit_comment_retrieve(self, post_id: int) -> Response:
data = {'_token': self.token, 'postId': post_id}
data = {"_token": self.token, "postId": post_id}
return self.post(f"{self.url}/main/military-unit-comment/retrieve/json", data=data)
def _post_main_military_unit_comment_create(self, post_id: int, comment_message: str) -> Response:
data = {'_token': self.token, 'postId': post_id, 'comment_message': comment_message}
data = {"_token": self.token, "postId": post_id, "comment_message": comment_message}
return self.post(f"{self.url}/main/military-unit-comment/create/json", data=data)
def _post_main_military_unit_post_create(self, body: str, post_as: int) -> Response:
data = {'_token': self.token, 'post_message': body, 'post_as': post_as}
data = {"_token": self.token, "post_message": body, "post_as": post_as}
return self.post(f"{self.url}/main/military-unit-post/create/json", data=data)
def _post_main_military_unit_post_retrieve(self) -> Response:
data = {'_token': self.token, 'page': 1, 'switchedFrom': False}
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post(f"{self.url}/main/military-unit-post/retrieve/json", data=data)
# ## Party
def _post_main_party_comment_retrieve(self, post_id: int) -> Response:
data = {'_token': self.token, 'postId': post_id}
data = {"_token": self.token, "postId": post_id}
return self.post(f"{self.url}/main/party-comment/retrieve/json", data=data)
def _post_main_party_comment_create(self, post_id: int, comment_message: str) -> Response:
data = {'_token': self.token, 'postId': post_id, 'comment_message': comment_message}
data = {"_token": self.token, "postId": post_id, "comment_message": comment_message}
return self.post(f"{self.url}/main/party-comment/create/json", data=data)
def _post_main_party_post_create(self, body: str) -> Response:
data = {'_token': self.token, 'post_message': body}
data = {"_token": self.token, "post_message": body}
return self.post(f"{self.url}/main/party-post/create/json", data=data)
def _post_main_party_post_retrieve(self) -> Response:
data = {'_token': self.token, 'page': 1, 'switchedFrom': False}
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post(f"{self.url}/main/party-post/retrieve/json", data=data)
# ## Friend's Wall
def _post_main_wall_comment_retrieve(self, post_id: int) -> Response:
data = {'_token': self.token, 'postId': post_id}
data = {"_token": self.token, "postId": post_id}
return self.post(f"{self.url}/main/wall-comment/retrieve/json", data=data)
def _post_main_wall_comment_create(self, post_id: int, comment_message: str) -> Response:
data = {'_token': self.token, 'postId': post_id, 'comment_message': comment_message}
data = {"_token": self.token, "postId": post_id, "comment_message": comment_message}
return self.post(f"{self.url}/main/wall-comment/create/json", data=data)
def _post_main_wall_post_create(self, body: str) -> Response:
data = {'_token': self.token, 'post_message': body}
data = {"_token": self.token, "post_message": body}
return self.post(f"{self.url}/main/wall-post/create/json", data=data)
def _post_main_wall_post_retrieve(self) -> Response:
data = {'_token': self.token, 'page': 1, 'switchedFrom': False}
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post(f"{self.url}/main/wall-post/retrieve/json", data=data)
# ## Medal posting
def _post_main_wall_post_automatic(self, message: str, achievement_id: int) -> Response:
return self.post(f"{self.url}/main/wall-post/automatic", data=dict(_token=self.token, message=message,
achievementId=achievement_id))
return self.post(
f"{self.url}/main/wall-post/automatic", data=dict(_token=self.token, message=message, achievementId=achievement_id)
)
class CitizenAPI(
ErepublikArticleAPI, ErepublikCountryAPI, ErepublikCompanyAPI, ErepublikEconomyAPI,
ErepublikLeaderBoardAPI, ErepublikLocationAPI, ErepublikMilitaryAPI, ErepublikProfileAPI,
ErepublikPresidentAPI, ErepublikPoliticsAPI, ErepublikAnniversaryAPI, ErepublikWallPostAPI,
ErepublikTravelAPI
ErepublikArticleAPI,
ErepublikCountryAPI,
ErepublikCompanyAPI,
ErepublikEconomyAPI,
ErepublikLeaderBoardAPI,
ErepublikLocationAPI,
ErepublikMilitaryAPI,
ErepublikProfileAPI,
ErepublikPresidentAPI,
ErepublikPoliticsAPI,
ErepublikAnniversaryAPI,
ErepublikWallPostAPI,
ErepublikTravelAPI,
):
pass

1595
erepublik/citizen.py
View File

File diff suppressed because it is too large Load Diff

545
erepublik/classes.py
View File

@ -12,9 +12,25 @@ from requests import HTTPError, Response, Session, post
from erepublik import _types as types
from erepublik import constants, utils
__all__ = ['Battle', 'BattleDivision', 'BattleSide', 'Company', 'Config', 'Details', 'Energy', 'ErepublikException',
'ErepublikNetworkException', 'EnergyToFight', 'Holding', 'Inventory', 'MyCompanies', 'OfferItem', 'Politics',
'Reporter', 'TelegramReporter', ]
__all__ = [
"Battle",
"BattleDivision",
"BattleSide",
"Company",
"Config",
"Details",
"Energy",
"ErepublikException",
"ErepublikNetworkException",
"EnergyToFight",
"Holding",
"Inventory",
"MyCompanies",
"OfferItem",
"Politics",
"Reporter",
"TelegramReporter",
]
class ErepublikException(Exception):
@ -39,7 +55,7 @@ class CaptchaSessionError(ErepublikNetworkException):
class Holding:
id: int
region: int
companies: List['Company']
companies: List["Company"]
name: str
_citizen = weakref.ReferenceType
@ -47,16 +63,16 @@ class Holding:
self._citizen = weakref.ref(citizen)
self.id: int = _id
self.region: int = region
self.companies: List['Company'] = list()
self.companies: List["Company"] = list()
if name:
self.name = name
else:
comp_sum = len(self.companies)
name = f"Holding (#{self.id}) with {comp_sum} "
if comp_sum == 1:
name += 'company'
name += "company"
else:
name += 'companies'
name += "companies"
self.name = name
@property
@ -64,20 +80,20 @@ class Holding:
return len([1 for company in self.companies if company.wam_enabled and not company.already_worked])
@property
def wam_companies(self) -> Iterable['Company']:
def wam_companies(self) -> Iterable["Company"]:
return [company for company in self.companies if company.wam_enabled]
@property
def employable_companies(self) -> Iterable['Company']:
def employable_companies(self) -> Iterable["Company"]:
return [company for company in self.companies if company.preset_works]
def add_company(self, company: 'Company') -> NoReturn:
def add_company(self, company: "Company") -> NoReturn:
self.companies.append(company)
self.companies.sort()
def get_wam_raw_usage(self) -> Dict[str, Decimal]:
frm = Decimal('0.00')
wrm = Decimal('0.00')
frm = Decimal("0.00")
wrm = Decimal("0.00")
for company in self.wam_companies:
if company.industry in [1, 7, 8, 9, 10, 11]:
frm += company.raw_usage
@ -85,11 +101,11 @@ class Holding:
wrm += company.raw_usage
return dict(frm=frm, wrm=wrm)
def get_wam_companies(self, raw_factory: bool = None) -> List['Company']:
def get_wam_companies(self, raw_factory: bool = None) -> List["Company"]:
raw = []
factory = []
for company in self.wam_companies:
if not company.already_worked and not company.cannot_wam_reason == 'war':
if not company.already_worked and not company.cannot_wam_reason == "war":
if company.is_raw:
raw.append(company)
else:
@ -103,9 +119,9 @@ class Holding:
comp = len(self.companies)
name = f"Holding (#{self.id}) with {comp} "
if comp == 1:
name += 'company'
name += "company"
else:
name += 'companies'
name += "companies"
return name
def __repr__(self):
@ -113,8 +129,7 @@ class Holding:
@property
def as_dict(self) -> Dict[str, Union[str, int, List[Dict[str, Union[str, int, bool, float, Decimal]]]]]:
return dict(name=self.name, id=self.id, region=self.region,
companies=[c.as_dict for c in self.companies], wam_count=self.wam_count)
return dict(name=self.name, id=self.id, region=self.region, companies=[c.as_dict for c in self.companies], wam_count=self.wam_count)
@property
def citizen(self):
@ -137,9 +152,20 @@ class Company:
preset_works: int
def __init__(
self, holding: Holding, _id: int, quality: int, is_raw: bool, effective_bonus: Decimal, raw_usage: Decimal,
base_production: Decimal, wam_enabled: bool, can_wam: bool, cannot_wam_reason: str, industry: int,
already_worked: bool, preset_works: int
self,
holding: Holding,
_id: int,
quality: int,
is_raw: bool,
effective_bonus: Decimal,
raw_usage: Decimal,
base_production: Decimal,
wam_enabled: bool,
can_wam: bool,
cannot_wam_reason: str,
industry: int,
already_worked: bool,
preset_works: int,
):
self._holding = weakref.ref(holding)
self.id: int = _id
@ -196,22 +222,22 @@ class Company:
def __hash__(self):
return hash(self._sort_keys)
def __lt__(self, other: 'Company'):
def __lt__(self, other: "Company"):
return self._sort_keys < other._sort_keys
def __le__(self, other: 'Company'):
def __le__(self, other: "Company"):
return self._sort_keys <= other._sort_keys
def __gt__(self, other: 'Company'):
def __gt__(self, other: "Company"):
return self._sort_keys > other._sort_keys
def __ge__(self, other: 'Company'):
def __ge__(self, other: "Company"):
return self._sort_keys >= other._sort_keys
def __eq__(self, other: 'Company'):
def __eq__(self, other: "Company"):
return self._sort_keys == other._sort_keys
def __ne__(self, other: 'Company'):
def __ne__(self, other: "Company"):
return self._sort_keys != other._sort_keys
def __str__(self):
@ -225,10 +251,21 @@ class Company:
@property
def as_dict(self) -> Dict[str, Union[str, int, bool, float, Decimal]]:
return dict(name=str(self), holding=self.holding.id, id=self.id, quality=self.quality, is_raw=self.is_raw,
raw_usage=self.raw_usage, products_made=self.products_made, wam_enabled=self.wam_enabled,
can_wam=self.can_wam, cannot_wam_reason=self.cannot_wam_reason, industry=self.industry,
already_worked=self.already_worked, preset_works=self.preset_works)
return dict(
name=str(self),
holding=self.holding.id,
id=self.id,
quality=self.quality,
is_raw=self.is_raw,
raw_usage=self.raw_usage,
products_made=self.products_made,
wam_enabled=self.wam_enabled,
can_wam=self.can_wam,
cannot_wam_reason=self.cannot_wam_reason,
industry=self.industry,
already_worked=self.already_worked,
preset_works=self.preset_works,
)
def dissolve(self) -> Response:
self.holding.citizen.write_log(f"{self} dissolved!")
@ -265,12 +302,10 @@ class MyCompanies:
:param holdings: Parsed JSON to dict from en/economy/myCompanies
"""
for holding in holdings.values():
if holding.get('id') not in self.holdings:
self.holdings.update({
int(holding.get('id')): Holding(holding['id'], holding['region_id'], self.citizen, holding['name'])
})
if holding.get("id") not in self.holdings:
self.holdings.update({int(holding.get("id")): Holding(holding["id"], holding["region_id"], self.citizen, holding["name"])})
if not self.holdings.get(0):
self.holdings.update({0: Holding(0, 0, self.citizen, 'Unassigned')}) # unassigned
self.holdings.update({0: Holding(0, 0, self.citizen, "Unassigned")}) # unassigned
def prepare_companies(self, companies: Dict[str, Dict[str, Any]]):
"""
@ -278,19 +313,27 @@ class MyCompanies:
"""
self.__clear_data()
for company_dict in companies.values():
holding = self.holdings.get(int(company_dict['holding_company_id']))
quality = company_dict.get('quality')
is_raw = company_dict.get('is_raw')
holding = self.holdings.get(int(company_dict["holding_company_id"]))
quality = company_dict.get("quality")
is_raw = company_dict.get("is_raw")
if is_raw:
raw_usage = Decimal('0.0')
raw_usage = Decimal("0.0")
else:
raw_usage = Decimal(str(company_dict.get('upgrades').get(str(quality)).get('raw_usage')))
raw_usage = Decimal(str(company_dict.get("upgrades").get(str(quality)).get("raw_usage")))
company = Company(
holding, company_dict.get('id'), quality, is_raw,
Decimal(str(company_dict.get('effective_bonus'))) / 100,
raw_usage, Decimal(str(company_dict.get('base_production'))), company_dict.get('wam_enabled'),
company_dict.get('can_work_as_manager'), company_dict.get('cannot_work_as_manager_reason'),
company_dict.get('industry_id'), company_dict.get('already_worked'), company_dict.get('preset_works')
holding,
company_dict.get("id"),
quality,
is_raw,
Decimal(str(company_dict.get("effective_bonus"))) / 100,
raw_usage,
Decimal(str(company_dict.get("base_production"))),
company_dict.get("wam_enabled"),
company_dict.get("can_work_as_manager"),
company_dict.get("cannot_work_as_manager_reason"),
company_dict.get("industry_id"),
company_dict.get("already_worked"),
company_dict.get("preset_works"),
)
self._companies.add(company)
holding.add_company(company)
@ -326,13 +369,20 @@ class MyCompanies:
self._companies.clear()
@property
def as_dict(self) -> Dict[str, Union[str, int, datetime.datetime, Dict[str, Dict[str, Union[
str, int, List[Dict[str, Union[str, int, bool, float, Decimal]]]]
]]]]:
return dict(name=str(self), work_units=self.work_units, next_ot_time=self.next_ot_time,
def as_dict(
self,
) -> Dict[
str,
Union[str, int, datetime.datetime, Dict[str, Dict[str, Union[str, int, List[Dict[str, Union[str, int, bool, float, Decimal]]]]]]],
]:
return dict(
name=str(self),
work_units=self.work_units,
next_ot_time=self.next_ot_time,
ff_lockdown=self.ff_lockdown,
holdings={str(hi): h.as_dict for hi, h in self.holdings.items()},
company_count=sum(1 for _ in self.companies))
company_count=sum(1 for _ in self.companies),
)
@property
def citizen(self):
@ -408,16 +458,37 @@ class Config:
@property
def as_dict(self) -> Dict[str, Union[bool, int, str, List[str]]]:
return dict(email=self.email, work=self.work, train=self.train, wam=self.wam, ot=self.ot,
auto_sell=self.auto_sell, auto_sell_all=self.auto_sell_all, employees=self.employees,
fight=self.fight, air=self.air, ground=self.ground, all_in=self.all_in,
next_energy=self.next_energy, travel_to_fight=self.travel_to_fight,
always_travel=self.always_travel, epic_hunt=self.epic_hunt, epic_hunt_ebs=self.epic_hunt_ebs,
rw_def_side=self.rw_def_side, interactive=self.interactive, maverick=self.maverick,
continuous_fighting=self.continuous_fighting, auto_buy_raw=self.auto_buy_raw,
force_wam=self.force_wam, sort_battles_time=self.sort_battles_time, force_travel=self.force_travel,
telegram=self.telegram, telegram_chat_id=self.telegram_chat_id, telegram_token=self.telegram_token,
spin_wheel_of_fortune=self.spin_wheel_of_fortune)
return dict(
email=self.email,
work=self.work,
train=self.train,
wam=self.wam,
ot=self.ot,
auto_sell=self.auto_sell,
auto_sell_all=self.auto_sell_all,
employees=self.employees,
fight=self.fight,
air=self.air,
ground=self.ground,
all_in=self.all_in,
next_energy=self.next_energy,
travel_to_fight=self.travel_to_fight,
always_travel=self.always_travel,
epic_hunt=self.epic_hunt,
epic_hunt_ebs=self.epic_hunt_ebs,
rw_def_side=self.rw_def_side,
interactive=self.interactive,
maverick=self.maverick,
continuous_fighting=self.continuous_fighting,
auto_buy_raw=self.auto_buy_raw,
force_wam=self.force_wam,
sort_battles_time=self.sort_battles_time,
force_travel=self.force_travel,
telegram=self.telegram,
telegram_chat_id=self.telegram_chat_id,
telegram_token=self.telegram_token,
spin_wheel_of_fortune=self.spin_wheel_of_fortune,
)
class Energy:
@ -434,12 +505,12 @@ class Energy:
@property
def recovered(self):
warnings.warn('Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy', DeprecationWarning)
warnings.warn("Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy", DeprecationWarning)
return self.energy
@property
def recoverable(self):
warnings.warn('Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy', DeprecationWarning)
warnings.warn("Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy", DeprecationWarning)
return 0
def set_reference_time(self, recovery_time: datetime.datetime):
@ -459,12 +530,12 @@ class Energy:
@property
def is_recoverable_full(self):
warnings.warn('Deprecated since auto auto-eat! Will be removed soon. Use Energy.is_energy_full', DeprecationWarning)
warnings.warn("Deprecated since auto auto-eat! Will be removed soon. Use Energy.is_energy_full", DeprecationWarning)
return self.is_energy_full
@property
def is_recovered_full(self):
warnings.warn('Deprecated since auto auto-eat! Will be removed soon. Use Energy.is_energy_full', DeprecationWarning)
warnings.warn("Deprecated since auto auto-eat! Will be removed soon. Use Energy.is_energy_full", DeprecationWarning)
return self.is_energy_full
@property
@ -473,14 +544,19 @@ class Energy:
@property
def available(self):
warnings.warn('Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy', DeprecationWarning)
warnings.warn("Deprecated since auto auto-eat! Will be removed soon. Use Energy.energy", DeprecationWarning)
return self.energy
@property
def as_dict(self) -> Dict[str, Union[int, datetime.datetime, bool]]:
return dict(limit=self.limit, interval=self.interval, energy=self.energy,
reference_time=self.reference_time, food_fights=self.food_fights,
is_energy_full=self.is_energy_full)
return dict(
limit=self.limit,
interval=self.interval,
energy=self.energy,
reference_time=self.reference_time,
food_fights=self.food_fights,
is_energy_full=self.is_energy_full,
)
class Details:
@ -504,7 +580,7 @@ class Details:
def __init__(self):
self.next_pp = []
self.mayhem_skills = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0, 13: 0, 14: 0}
_default_country = constants.Country(0, 'Unknown', 'Unknown', 'XX')
_default_country = constants.Country(0, "Unknown", "Unknown", "XX")
self.citizenship = self.current_country = self.residence_country = _default_country
@property
@ -535,12 +611,25 @@ class Details:
@property
def as_dict(self) -> Dict[str, Union[int, float, str, constants.Country, bool]]:
return dict(xp=self.xp, cc=self.cc, pp=self.pp, pin=self.pin, gold=self.gold, next_pp=self.next_pp,
level=self.level, citizen_id=self.citizen_id, citizenship=self.citizenship,
current_region=self.current_region, current_country=self.current_country,
residence_region=self.residence_region, residence_country=self.residence_country,
daily_task_done=self.daily_task_done, daily_task_reward=self.daily_task_reward,
mayhem_skills=self.mayhem_skills, xp_till_level_up=self.xp_till_level_up)
return dict(
xp=self.xp,
cc=self.cc,
pp=self.pp,
pin=self.pin,
gold=self.gold,
next_pp=self.next_pp,
level=self.level,
citizen_id=self.citizen_id,
citizenship=self.citizenship,
current_region=self.current_region,
current_country=self.current_country,
residence_region=self.residence_region,
residence_country=self.residence_country,
daily_task_done=self.daily_task_done,
daily_task_reward=self.daily_task_reward,
mayhem_skills=self.mayhem_skills,
xp_till_level_up=self.xp_till_level_up,
)
@property
def is_elite(self):
@ -558,9 +647,14 @@ class Politics:
@property
def as_dict(self) -> Dict[str, Union[bool, int, str]]:
return dict(is_party_member=self.is_party_member, party_id=self.party_id, party_slug=self.party_slug,
is_party_president=self.is_party_president, is_congressman=self.is_congressman,
is_country_president=self.is_country_president)
return dict(
is_party_member=self.is_party_member,
party_id=self.party_id,
party_slug=self.party_slug,
is_party_president=self.is_party_president,
is_congressman=self.is_congressman,
is_country_president=self.is_country_president,
)
class House:
@ -596,17 +690,22 @@ class Reporter:
@property
def as_dict(self) -> Dict[str, Union[bool, int, str, List[Dict[Any, Any]]]]:
return dict(name=self.name, email=self.email, citizen_id=self.citizen_id, key=self.key, allowed=self.allowed,
queue=self.__to_update)
return dict(
name=self.name, email=self.email, citizen_id=self.citizen_id, key=self.key, allowed=self.allowed, queue=self.__to_update
)
def __init__(self, citizen):
self._citizen = weakref.ref(citizen)
self._req = Session()
self.url = "https://api.erep.lv"
self._req.headers.update({"user-agent": 'eRepublik Script Reporter v3',
'erep-version': utils.__version__,
'erep-user-id': str(self.citizen_id),
'erep-user-name': self.citizen.name})
self._req.headers.update(
{
"user-agent": "eRepublik Script Reporter v3",
"erep-version": utils.__version__,
"erep-user-id": str(self.citizen_id),
"erep-user-name": self.citizen.name,
}
)
self.__to_update = []
self.__registered: bool = False
@ -651,67 +750,102 @@ class Reporter:
if not self.__registered:
r = self.__bot_update(dict(key=self.key, check=True, player_id=self.citizen_id))
if r:
if not r.json().get('status'):
self._req.post(f"{self.url}/bot/register", json=dict(name=self.name, email=self.email,
player_id=self.citizen_id))
if not r.json().get("status"):
self._req.post(f"{self.url}/bot/register", json=dict(name=self.name, email=self.email, player_id=self.citizen_id))
self.__registered = True
self.allowed = True
self.report_action('STARTED', value=utils.now().strftime("%F %T"))
self.report_action("STARTED", value=utils.now().strftime("%F %T"))
def send_state_update(self, xp: int, cc: float, gold: float, inv_total: int, inv: int,
hp_limit: int, hp_interval: int, hp_available: int, food: int, pp: int):
def send_state_update(
self,
xp: int,
cc: float,
gold: float,
inv_total: int,
inv: int,
hp_limit: int,
hp_interval: int,
hp_available: int,
food: int,
pp: int,
):
data = dict(key=self.key, player_id=self.citizen_id, state=dict(
xp=xp, cc=cc, gold=gold, inv_total=inv_total, inv_free=inv_total - inv, inv=inv, food=food,
pp=pp, hp_limit=hp_limit, hp_interval=hp_interval, hp_available=hp_available,
))
data = dict(
key=self.key,
player_id=self.citizen_id,
state=dict(
xp=xp,
cc=cc,
gold=gold,
inv_total=inv_total,
inv_free=inv_total - inv,
inv=inv,
food=food,
pp=pp,
hp_limit=hp_limit,
hp_interval=hp_interval,
hp_available=hp_available,
),
)
self._bot_update(data)
def report_action(self, action: str, json_val: Dict[Any, Any] = None, value: str = None):
json_data = dict(
player_id=getattr(self, 'citizen_id', None), log={'action': action}, key=getattr(self, 'key', None)
)
json_data = dict(player_id=getattr(self, "citizen_id", None), log={"action": action}, key=getattr(self, "key", None))
if json_val:
json_data['log'].update(dict(json=json_val))
json_data["log"].update(dict(json=json_val))
if value:
json_data['log'].update(dict(value=value))
json_data["log"].update(dict(value=value))
if not any([self.key, self.email, self.name, self.citizen_id]):
return
self._bot_update(json_data)
def report_fighting(self, battle: 'Battle', invader: bool, division: 'BattleDivision', damage: float, hits: int):
def report_fighting(self, battle: "Battle", invader: bool, division: "BattleDivision", damage: float, hits: int):
side = battle.invader if invader else battle.defender
self.report_action('FIGHT', dict(battle_id=battle.id, side=side, dmg=damage,
air=battle.has_air, hits=hits,
round=battle.zone_id, extra=dict(battle=battle, side=side, division=division)))
self.report_action(
"FIGHT",
dict(
battle_id=battle.id,
side=side,
dmg=damage,
air=battle.has_air,
hits=hits,
round=battle.zone_id,
extra=dict(battle=battle, side=side, division=division),
),
)
def report_money_donation(self, citizen_id: int, amount: float, is_currency: bool = True):
cur = 'cc' if is_currency else 'gold'
self.report_action('DONATE_MONEY', dict(citizen_id=citizen_id, amount=amount, currency=cur),
f"Successfully donated {amount}{cur} to citizen with id {citizen_id}!")
cur = "cc" if is_currency else "gold"
self.report_action(
"DONATE_MONEY",
dict(citizen_id=citizen_id, amount=amount, currency=cur),
f"Successfully donated {amount}{cur} to citizen with id {citizen_id}!",
)
def report_item_donation(self, citizen_id: int, amount: float, quality: int, industry: str):
self.report_action('DONATE_ITEMS',
self.report_action(
"DONATE_ITEMS",
dict(citizen_id=citizen_id, amount=amount, quality=quality, industry=industry),
f"Successfully donated {amount} x {industry} q{quality} to citizen with id {citizen_id}!")
f"Successfully donated {amount} x {industry} q{quality} to citizen with id {citizen_id}!",
)
def report_promo(self, kind: str, time_until: datetime.datetime):
self._req.post(f"{self.url}/promos/add/", data=dict(kind=kind, time_untill=time_until))
def fetch_battle_priorities(self, country: constants.Country) -> List['Battle']:
def fetch_battle_priorities(self, country: constants.Country) -> List["Battle"]:
try:
battle_response = self._req.get(f'{self.url}/api/v1/battles/{country.id}')
return [self.citizen.all_battles[bid] for bid in battle_response.json().get('battle_ids', []) if
bid in self.citizen.all_battles]
battle_response = self._req.get(f"{self.url}/api/v1/battles/{country.id}")
return [
self.citizen.all_battles[bid] for bid in battle_response.json().get("battle_ids", []) if bid in self.citizen.all_battles
]
except: # noqa
return []
def fetch_tasks(self) -> List[Dict[str, Any]]:
try:
task_response = self._req.post(
f'{self.url}/api/v1/command', data=dict(citizen=self.citizen_id, key=self.key)).json()
if task_response.get('status'):
return task_response.get('data')
task_response = self._req.post(f"{self.url}/api/v1/command", data=dict(citizen=self.citizen_id, key=self.key)).json()
if task_response.get("status"):
return task_response.get("data")
else:
return []
except: # noqa
@ -722,13 +856,20 @@ class BattleSide:
points: int
deployed: List[constants.Country]
allies: List[constants.Country]
battle: 'Battle'
battle: "Battle"
_battle: weakref.ReferenceType
country: constants.Country
is_defender: bool
def __init__(self, battle: 'Battle', country: constants.Country, points: int, allies: List[constants.Country],
deployed: List[constants.Country], defender: bool):
def __init__(
self,
battle: "Battle",
country: constants.Country,
points: int,
allies: List[constants.Country],
deployed: List[constants.Country],
defender: bool,
):
self._battle = weakref.ref(battle)
self.country = country
self.points = points
@ -741,11 +882,11 @@ class BattleSide:
return self.country.id
def __repr__(self):
side_text = 'Defender' if self.is_defender else 'Invader '
side_text = "Defender" if self.is_defender else "Invader "
return f"<BattleSide: {side_text} {self.country.name}|{self.points:>2d}p>"
def __str__(self):
side_text = 'Defender' if self.is_defender else 'Invader '
side_text = "Defender" if self.is_defender else "Invader "
return f"{side_text} {self.country.name} - {self.points:>2d} points"
def __format__(self, format_spec):
@ -753,8 +894,7 @@ class BattleSide:
@property
def as_dict(self) -> Dict[str, Union[int, constants.Country, bool, List[constants.Country]]]:
return dict(points=self.points, country=self.country, is_defender=self.is_defender, allies=self.allies,
deployed=self.deployed)
return dict(points=self.points, country=self.country, is_defender=self.is_defender, allies=self.allies, deployed=self.deployed)
@property
def battle(self):
@ -771,13 +911,14 @@ class BattleDivision:
inv_medal: Dict[str, int]
terrain: int
div: int
battle: 'Battle'
battle: "Battle"
_battle: weakref.ReferenceType
@property
def as_dict(self):
return dict(id=self.id, division=self.div, terrain=(self.terrain, self.terrain_display), wall=self.wall,
epic=self.epic, end=self.div_end)
return dict(
id=self.id, division=self.div, terrain=(self.terrain, self.terrain_display), wall=self.wall, epic=self.epic, end=self.div_end
)
@property
def is_air(self):
@ -787,8 +928,17 @@ class BattleDivision:
def div_end(self) -> bool:
return utils.now() >= self.end
def __init__(self, battle: 'Battle', div_id: int, end: datetime.datetime, epic: bool, div: int, wall_for: int,
wall_dom: float, terrain_id: int = 0):
def __init__(
self,
battle: "Battle",
div_id: int,
end: datetime.datetime,
epic: bool,
div: int,
wall_for: int,
wall_dom: float,
terrain_id: int = 0,
):
"""Battle division helper class
:type div_id: int
@ -803,7 +953,7 @@ class BattleDivision:
self.id = div_id
self.end = end
self.epic = epic
self.wall = {'for': wall_for, 'dom': wall_dom}
self.wall = {"for": wall_for, "dom": wall_dom}
self.terrain = terrain_id
self.div = div
@ -816,7 +966,7 @@ class BattleDivision:
if self.terrain:
base_name += f" ({self.terrain_display})"
if self.div_end:
base_name += ' Ended'
base_name += " Ended"
return base_name
def __repr__(self):
@ -842,9 +992,18 @@ class Battle:
@property
def as_dict(self):
return dict(id=self.id, war_id=self.war_id, divisions=self.div, zone=self.zone_id, rw=self.is_rw,
dict_lib=self.is_dict_lib, start=self.start, sides={'inv': self.invader, 'def': self.defender},
region=[self.region_id, self.region_name], link=self.link)
return dict(
id=self.id,
war_id=self.war_id,
divisions=self.div,
zone=self.zone_id,
rw=self.is_rw,
dict_lib=self.is_dict_lib,
start=self.start,
sides={"inv": self.invader, "def": self.defender},
region=[self.region_id, self.region_name],
link=self.link,
)
@property
def has_air(self) -> bool:
@ -874,43 +1033,52 @@ class Battle:
:param battle: Dict object for single battle from '/military/campaignsJson/list' response's 'battles' object
"""
self.id = int(battle.get('id'))
self.war_id = int(battle.get('war_id'))
self.zone_id = int(battle.get('zone_id'))
self.is_rw = bool(battle.get('is_rw'))
self.is_as = bool(battle.get('is_as'))
self.is_dict_lib = bool(battle.get('is_dict')) or bool(battle.get('is_lib'))
self.region_id = battle.get('region', {}).get('id')
self.region_name = battle.get('region', {}).get('name')
self.start = datetime.datetime.fromtimestamp(int(battle.get('start', 0)), tz=constants.erep_tz)
self.id = int(battle.get("id"))
self.war_id = int(battle.get("war_id"))
self.zone_id = int(battle.get("zone_id"))
self.is_rw = bool(battle.get("is_rw"))
self.is_as = bool(battle.get("is_as"))
self.is_dict_lib = bool(battle.get("is_dict")) or bool(battle.get("is_lib"))
self.region_id = battle.get("region", {}).get("id")
self.region_name = battle.get("region", {}).get("name")
self.start = datetime.datetime.fromtimestamp(int(battle.get("start", 0)), tz=constants.erep_tz)
self.invader = BattleSide(
self, constants.COUNTRIES[battle.get('inv', {}).get('id')], battle.get('inv', {}).get('points'),
[constants.COUNTRIES[row.get('id')] for row in battle.get('inv', {}).get('ally_list')],
[constants.COUNTRIES[row.get('id')] for row in battle.get('inv', {}).get('ally_list') if row['deployed']],
False
self,
constants.COUNTRIES[battle.get("inv", {}).get("id")],
battle.get("inv", {}).get("points"),
[constants.COUNTRIES[row.get("id")] for row in battle.get("inv", {}).get("ally_list")],
[constants.COUNTRIES[row.get("id")] for row in battle.get("inv", {}).get("ally_list") if row["deployed"]],
False,
)
self.defender = BattleSide(
self, constants.COUNTRIES[battle.get('def', {}).get('id')], battle.get('def', {}).get('points'),
[constants.COUNTRIES[row.get('id')] for row in battle.get('def', {}).get('ally_list')],
[constants.COUNTRIES[row.get('id')] for row in battle.get('def', {}).get('ally_list') if row['deployed']],
True
self,
constants.COUNTRIES[battle.get("def", {}).get("id")],
battle.get("def", {}).get("points"),
[constants.COUNTRIES[row.get("id")] for row in battle.get("def", {}).get("ally_list")],
[constants.COUNTRIES[row.get("id")] for row in battle.get("def", {}).get("ally_list") if row["deployed"]],
True,
)
self.div = {}
for div, data in battle.get('div', {}).items():
for div, data in battle.get("div", {}).items():
div = int(div)
if data.get('end'):
end = datetime.datetime.fromtimestamp(data.get('end'), tz=constants.erep_tz)
if data.get("end"):
end = datetime.datetime.fromtimestamp(data.get("end"), tz=constants.erep_tz)
else:
end = constants.max_datetime
battle_div = BattleDivision(self, div_id=data.get('id'), div=data.get('div'), end=end,
epic=data.get('epic_type') in [1, 5],
wall_for=data.get('wall').get('for'),
wall_dom=data.get('wall').get('dom'),
terrain_id=data.get('terrain', 0))
battle_div = BattleDivision(
self,
div_id=data.get("id"),
div=data.get("div"),
end=end,
epic=data.get("epic_type") in [1, 5],
wall_for=data.get("wall").get("for"),
wall_dom=data.get("wall").get("dom"),
terrain_id=data.get("terrain", 0),
)
self.div.update({div: battle_div})
@ -922,8 +1090,10 @@ class Battle:
else:
time_part = f"-{self.start - time_now}"
return (f"Battle {self.id} for {self.region_name[:16]:16} | "
f"{self.invader} : {self.defender} | Round time {time_part} | {'R'+str(self.zone_id):>3}")
return (
f"Battle {self.id} for {self.region_name[:16]:16} | "
f"{self.invader} : {self.defender} | Round time {time_part} | {'R'+str(self.zone_id):>3}"
)
def __repr__(self):
return f"<Battle #{self.id} {self.invader}:{self.defender} R{self.zone_id}>"
@ -981,9 +1151,16 @@ class TelegramReporter:
@property
def as_dict(self):
return {'chat_id': self.chat_id, 'api_url': self.api_url, 'player': self.player_name,
'last_time': self._last_time, 'next_time': self._next_time, 'queue': self.__queue,
'initialized': self.__initialized, 'has_threads': not self._threads}
return {
"chat_id": self.chat_id,
"api_url": self.api_url,
"player": self.player_name,
"last_time": self._last_time,
"next_time": self._next_time,
"queue": self.__queue,
"initialized": self.__initialized,
"has_threads": not self._threads,
}
def do_init(self, chat_id: int, token: str = None, player_name: str = None):
if token is None:
@ -995,7 +1172,7 @@ class TelegramReporter:
self._last_time = utils.good_timedelta(utils.now(), datetime.timedelta(minutes=-5))
self._last_full_energy_report = utils.good_timedelta(utils.now(), datetime.timedelta(minutes=-30))
if self.__queue:
self.send_message('Telegram initialized')
self.send_message("Telegram initialized")
def send_message(self, message: str) -> bool:
self.__queue.append(message)
@ -1020,22 +1197,27 @@ class TelegramReporter:
self.send_message(message)
def report_medal(self, msg, multiple: bool = True):
new_line = '\n' if multiple else ''
new_line = "\n" if multiple else ""
self.send_message(f"New award: {new_line}*{msg}*")
def report_fight(self, battle: 'Battle', invader: bool, division: 'BattleDivision', damage: float, hits: int):
def report_fight(self, battle: "Battle", invader: bool, division: "BattleDivision", damage: float, hits: int):
side_txt = (battle.invader if invader else battle.defender).country.iso
self.send_message(f"*Fight report*:\n{int(damage):,d} dmg ({hits} hits) in"
self.send_message(
f"*Fight report*:\n{int(damage):,d} dmg ({hits} hits) in"
f" [battle {battle.id} for {battle.region_name[:16]}]({battle.link}) in d{division.div} on "
f"{side_txt} side")
f"{side_txt} side"
)
def report_item_donation(self, citizen_id: int, amount: float, product: str):
self.send_message(f"*Donation*: {amount} x {product} to citizen "
f"[{citizen_id}](https://www.erepublik.com/en/citizen/profile/{citizen_id})")
self.send_message(
f"*Donation*: {amount} x {product} to citizen " f"[{citizen_id}](https://www.erepublik.com/en/citizen/profile/{citizen_id})"
)
def report_money_donation(self, citizen_id: int, amount: float, is_currency: bool = True):
self.send_message(f"*Donation*: {amount}{'cc' if is_currency else 'gold'} to citizen "
f"[{citizen_id}](https://www.erepublik.com/en/citizen/profile/{citizen_id})")
self.send_message(
f"*Donation*: {amount}{'cc' if is_currency else 'gold'} to citizen "
f"[{citizen_id}](https://www.erepublik.com/en/citizen/profile/{citizen_id})"
)
def __send_messages(self):
while self._next_time > utils.now():
@ -1046,9 +1228,9 @@ class TelegramReporter:
message = "\n\n".join(self.__queue)
if self.player_name:
message = f"Player *{self.player_name}*\n\n" + message
response = post(f"{self.api_url}/sendMessage", json=dict(chat_id=self.chat_id, text=message, parse_mode='Markdown'))
response = post(f"{self.api_url}/sendMessage", json=dict(chat_id=self.chat_id, text=message, parse_mode="Markdown"))
self._last_time = utils.now()
if response.json().get('ok'):
if response.json().get("ok"):
self.__queue.clear()
return True
return False
@ -1056,14 +1238,16 @@ class TelegramReporter:
def send_photos(self, photos: List[Tuple[str, BytesIO]]):
for photo_title, photo in photos:
photo.seek(0)
post(f"https://{self.api_url}/sendPhoto",
post(
f"https://{self.api_url}/sendPhoto",
data=dict(chat_id=self.chat_id, caption=photo_title),
files=[('photo', ("f{utils.slugify(photo_title)}.png", photo))])
files=[("photo", ("f{utils.slugify(photo_title)}.png", photo))],
)
return
class OfferItem(NamedTuple):
price: float = 999_999_999.
price: float = 999_999_999.0
country: constants.Country = constants.Country(0, "", "", "")
amount: int = 0
offer_id: int = 0
@ -1090,5 +1274,6 @@ class Inventory:
@property
def as_dict(self) -> Dict[str, Union[types.InvFinal, types.InvRaw, int]]:
return dict(active=self.active, final=self.final, boosters=self.boosters, raw=self.raw, offers=self.offers,
total=self.total, used=self.used)
return dict(
active=self.active, final=self.final, boosters=self.boosters, raw=self.raw, offers=self.offers, total=self.total, used=self.used
)

605
erepublik/constants.py
View File

@ -3,10 +3,21 @@ from typing import Dict, Optional, Union
import pytz
__all__ = ['erep_tz', 'min_datetime', 'max_datetime', 'Country', 'AIR_RANKS', 'COUNTRIES', 'FOOD_ENERGY',
'GROUND_RANKS', 'GROUND_RANK_POINTS', 'INDUSTRIES', 'TERRAINS']
__all__ = [
"erep_tz",
"min_datetime",
"max_datetime",
"Country",
"AIR_RANKS",
"COUNTRIES",
"FOOD_ENERGY",
"GROUND_RANKS",
"GROUND_RANK_POINTS",
"INDUSTRIES",
"TERRAINS",
]
erep_tz = pytz.timezone('US/Pacific')
erep_tz = pytz.timezone("US/Pacific")
min_datetime = erep_tz.localize(datetime.datetime(2007, 11, 20))
max_datetime = erep_tz.localize(datetime.datetime(2281, 9, 4))
@ -53,18 +64,70 @@ class Country:
class Industries:
__by_name = {'food': 1, 'weapon': 2, 'ticket': 3, 'house': 4, 'aircraft': 23,
'foodraw': 7, 'weaponraw': 12, 'houseraw': 18, 'aircraftraw': 24, 'airplaneraw': 24,
'frm': 7, 'wrm': 12, 'hrm': 18, 'arm': 24,
'frm q1': 7, 'frm q2': 8, 'frm q3': 9, 'frm q4': 10, 'frm q5': 11,
'wrm q1': 12, 'wrm q2': 13, 'wrm q3': 14, 'wrm q4': 15, 'wrm q5': 16,
'hrm q1': 18, 'hrm q2': 19, 'hrm q3': 20, 'hrm q4': 21, 'hrm q5': 22,
'arm q1': 24, 'arm q2': 25, 'arm q3': 26, 'arm q4': 27, 'arm q5': 28}
__by_id = {1: 'Food', 2: 'Weapon', 3: 'Ticket', 4: 'House', 23: 'Aircraft',
7: 'foodRaw', 8: 'FRM q2', 9: 'FRM q3', 10: 'FRM q4', 11: 'FRM q5',
12: 'weaponRaw', 13: 'WRM q2', 14: 'WRM q3', 15: 'WRM q4', 16: 'WRM q5',
17: 'houseRaw', 18: 'houseRaw', 19: 'HRM q2', 20: 'HRM q3', 21: 'HRM q4', 22: 'HRM q5',
24: 'aircraftRaw', 25: 'ARM q2', 26: 'ARM q3', 27: 'ARM q4', 28: 'ARM q5'}
__by_name = {
"food": 1,
"weapon": 2,
"ticket": 3,
"house": 4,
"aircraft": 23,
"foodraw": 7,
"weaponraw": 12,
"houseraw": 18,
"aircraftraw": 24,
"airplaneraw": 24,
"frm": 7,
"wrm": 12,
"hrm": 18,
"arm": 24,
"frm q1": 7,
"frm q2": 8,
"frm q3": 9,
"frm q4": 10,
"frm q5": 11,
"wrm q1": 12,
"wrm q2": 13,
"wrm q3": 14,
"wrm q4": 15,
"wrm q5": 16,
"hrm q1": 18,
"hrm q2": 19,
"hrm q3": 20,
"hrm q4": 21,
"hrm q5": 22,
"arm q1": 24,
"arm q2": 25,
"arm q3": 26,
"arm q4": 27,
"arm q5": 28,
}
__by_id = {
1: "Food",
2: "Weapon",
3: "Ticket",
4: "House",
23: "Aircraft",
7: "foodRaw",
8: "FRM q2",
9: "FRM q3",
10: "FRM q4",
11: "FRM q5",
12: "weaponRaw",
13: "WRM q2",
14: "WRM q3",
15: "WRM q4",
16: "WRM q5",
17: "houseRaw",
18: "houseRaw",
19: "HRM q2",
20: "HRM q3",
21: "HRM q4",
22: "HRM q5",
24: "aircraftRaw",
25: "ARM q2",
26: "ARM q3",
27: "ARM q4",
28: "ARM q5",
}
def __getitem__(self, item) -> Optional[Union[int, str]]:
if isinstance(item, int):
@ -144,110 +207,452 @@ class Rank:
AIR_RANK_NAMES: Dict[int, str] = {
1: 'Airman', 2: 'Airman 1st Class', 3: 'Airman 1st Class*', 4: 'Airman 1st Class**', 5: 'Airman 1st Class***', 6: 'Airman 1st Class****', 7: 'Airman 1st Class*****',
8: 'Senior Airman', 9: 'Senior Airman*', 10: 'Senior Airman**', 11: 'Senior Airman***', 12: 'Senior Airman****', 13: 'Senior Airman*****',
14: 'Staff Sergeant', 15: 'Staff Sergeant*', 16: 'Staff Sergeant**', 17: 'Staff Sergeant***', 18: 'Staff Sergeant****', 19: 'Staff Sergeant*****',
20: 'Aviator', 21: 'Aviator*', 22: 'Aviator**', 23: 'Aviator***', 24: 'Aviator****', 25: 'Aviator*****',
26: 'Flight Lieutenant', 27: 'Flight Lieutenant*', 28: 'Flight Lieutenant**', 29: 'Flight Lieutenant***', 30: 'Flight Lieutenant****', 31: 'Flight Lieutenant*****',
32: 'Squadron Leader', 33: 'Squadron Leader*', 34: 'Squadron Leader**', 35: 'Squadron Leader***', 36: 'Squadron Leader****', 37: 'Squadron Leader*****',
38: 'Chief Master Sergeant', 39: 'Chief Master Sergeant*', 40: 'Chief Master Sergeant**', 41: 'Chief Master Sergeant***', 42: 'Chief Master Sergeant****', 43: 'Chief Master Sergeant*****',
44: 'Wing Commander', 45: 'Wing Commander*', 46: 'Wing Commander**', 47: 'Wing Commander***', 48: 'Wing Commander****', 49: 'Wing Commander*****',
50: 'Group Captain', 51: 'Group Captain*', 52: 'Group Captain**', 53: 'Group Captain***', 54: 'Group Captain****', 55: 'Group Captain*****',
56: 'Air Commodore', 57: 'Air Commodore*', 58: 'Air Commodore**', 59: 'Air Commodore***', 60: 'Air Commodore****', 61: 'Air Commodore*****',
62: 'Air Vice Marshal', 63: 'Air Vice Marshal*', 64: 'Air Vice Marshal**', 65: 'Air Vice Marshal***', 66: 'Air Vice Marshal****', 67: 'Air Vice Marshal*****',
68: 'Air Marshal', 69: 'Air Marshal*', 70: 'Air Marshal**', 71: 'Air Marshal***', 72: 'Air Marshal****', 73: 'Air Marshal*****',
74: 'Air Chief Marshal', 75: 'Air Chief Marshal*', 76: 'Air Chief Marshal**', 77: 'Air Chief Marshal***', 78: 'Air Chief Marshal****', 79: 'Air Chief Marshal*****',
1: "Airman",
2: "Airman 1st Class",
3: "Airman 1st Class*",
4: "Airman 1st Class**",
5: "Airman 1st Class***",
6: "Airman 1st Class****",
7: "Airman 1st Class*****",
8: "Senior Airman",
9: "Senior Airman*",
10: "Senior Airman**",
11: "Senior Airman***",
12: "Senior Airman****",
13: "Senior Airman*****",
14: "Staff Sergeant",
15: "Staff Sergeant*",
16: "Staff Sergeant**",
17: "Staff Sergeant***",
18: "Staff Sergeant****",
19: "Staff Sergeant*****",
20: "Aviator",
21: "Aviator*",
22: "Aviator**",
23: "Aviator***",
24: "Aviator****",
25: "Aviator*****",
26: "Flight Lieutenant",
27: "Flight Lieutenant*",
28: "Flight Lieutenant**",
29: "Flight Lieutenant***",
30: "Flight Lieutenant****",
31: "Flight Lieutenant*****",
32: "Squadron Leader",
33: "Squadron Leader*",
34: "Squadron Leader**",
35: "Squadron Leader***",
36: "Squadron Leader****",
37: "Squadron Leader*****",
38: "Chief Master Sergeant",
39: "Chief Master Sergeant*",
40: "Chief Master Sergeant**",
41: "Chief Master Sergeant***",
42: "Chief Master Sergeant****",
43: "Chief Master Sergeant*****",
44: "Wing Commander",
45: "Wing Commander*",
46: "Wing Commander**",
47: "Wing Commander***",
48: "Wing Commander****",
49: "Wing Commander*****",
50: "Group Captain",
51: "Group Captain*",
52: "Group Captain**",
53: "Group Captain***",
54: "Group Captain****",
55: "Group Captain*****",
56: "Air Commodore",
57: "Air Commodore*",
58: "Air Commodore**",
59: "Air Commodore***",
60: "Air Commodore****",
61: "Air Commodore*****",
62: "Air Vice Marshal",
63: "Air Vice Marshal*",
64: "Air Vice Marshal**",
65: "Air Vice Marshal***",
66: "Air Vice Marshal****",
67: "Air Vice Marshal*****",
68: "Air Marshal",
69: "Air Marshal*",
70: "Air Marshal**",
71: "Air Marshal***",
72: "Air Marshal****",
73: "Air Marshal*****",
74: "Air Chief Marshal",
75: "Air Chief Marshal*",
76: "Air Chief Marshal**",
77: "Air Chief Marshal***",
78: "Air Chief Marshal****",
79: "Air Chief Marshal*****",
}
AIR_RANK_POINTS: Dict[int, Optional[int]] = {
1: 0, 2: 10, 3: 25, 4: 45, 5: 70, 6: 100, 7: 140, 8: 190, 9: 270, 10: 380, 11: 530, 12: 850, 13: 1300, 14: 2340, 15: 3300, 16: 4200, 17: 5150, 18: 6100, 19: 7020, 20: 9100, 21: 12750, 22: 16400, 23: 20000, 24: 23650, 25: 27300,
26: 35500, 27: 48000, 28: 60000, 29: 72400, 30: 84500, 31: 97000, 32: 110000, 33: 140000, 34: 170000, 35: 210000, 36: 290000, 37: 350000, 38: 429000, 39: 601000, 40: 772000, 41: 944000, 42: 1115000, 43: 1287000,
44: 1673000, 45: 2238000, 46: 2804000, 47: 3369000, 48: 3935000, 49: 4500000, 50: 5020000, 51: 7028000, 52: 9036000, 53: 11044000, 54: 13052000, 55: 15060000,
56: 19580000, 57: 27412000, 58: 35244000, 59: 43076000, 60: 50908000, 61: 58740000, 62: 76360000, 63: 113166443, 64: 137448000, 65: None, 66: None, 67: None,
68: None, 69: None, 70: None, 71: None, 72: None, 73: None, 74: None, 75: None, 76: None, 77: None, 78: None, 79: None,
1: 0,
2: 10,
3: 25,
4: 45,
5: 70,
6: 100,
7: 140,
8: 190,
9: 270,
10: 380,
11: 530,
12: 850,
13: 1300,
14: 2340,
15: 3300,
16: 4200,
17: 5150,
18: 6100,
19: 7020,
20: 9100,
21: 12750,
22: 16400,
23: 20000,
24: 23650,
25: 27300,
26: 35500,
27: 48000,
28: 60000,
29: 72400,
30: 84500,
31: 97000,
32: 110000,
33: 140000,
34: 170000,
35: 210000,
36: 290000,
37: 350000,
38: 429000,
39: 601000,
40: 772000,
41: 944000,
42: 1115000,
43: 1287000,
44: 1673000,
45: 2238000,
46: 2804000,
47: 3369000,
48: 3935000,
49: 4500000,
50: 5020000,
51: 7028000,
52: 9036000,
53: 11044000,
54: 13052000,
55: 15060000,
56: 19580000,
57: 27412000,
58: 35244000,
59: 43076000,
60: 50908000,
61: 58740000,
62: 76360000,
63: 113166443,
64: 137448000,
65: None,
66: None,
67: None,
68: None,
69: None,
70: None,
71: None,
72: None,
73: None,
74: None,
75: None,
76: None,
77: None,
78: None,
79: None,
}
AIR_RANKS: Dict[int, Rank] = {i: Rank(i, AIR_RANK_NAMES[i], AIR_RANK_POINTS[i], True) for i in range(1, 80)}
COUNTRIES: Dict[int, Country] = {
1: Country(1, 'Romania', 'Romania', 'ROU'), 9: Country(9, 'Brazil', 'Brazil', 'BRA'),
10: Country(10, 'Italy', 'Italy', 'ITA'), 11: Country(11, 'France', 'France', 'FRA'),
12: Country(12, 'Germany', 'Germany', 'DEU'), 13: Country(13, 'Hungary', 'Hungary', 'HUN'),
14: Country(14, 'China', 'China', 'CHN'), 15: Country(15, 'Spain', 'Spain', 'ESP'),
23: Country(23, 'Canada', 'Canada', 'CAN'), 24: Country(24, 'USA', 'USA', 'USA'),
26: Country(26, 'Mexico', 'Mexico', 'MEX'), 27: Country(27, 'Argentina', 'Argentina', 'ARG'),
28: Country(28, 'Venezuela', 'Venezuela', 'VEN'), 29: Country(29, 'United Kingdom', 'United-Kingdom', 'GBR'),
30: Country(30, 'Switzerland', 'Switzerland', 'CHE'), 31: Country(31, 'Netherlands', 'Netherlands', 'NLD'),
32: Country(32, 'Belgium', 'Belgium', 'BEL'), 33: Country(33, 'Austria', 'Austria', 'AUT'),
34: Country(34, 'Czech Republic', 'Czech-Republic', 'CZE'), 35: Country(35, 'Poland', 'Poland', 'POL'),
36: Country(36, 'Slovakia', 'Slovakia', 'SVK'), 37: Country(37, 'Norway', 'Norway', 'NOR'),
38: Country(38, 'Sweden', 'Sweden', 'SWE'), 39: Country(39, 'Finland', 'Finland', 'FIN'),
40: Country(40, 'Ukraine', 'Ukraine', 'UKR'), 41: Country(41, 'Russia', 'Russia', 'RUS'),
42: Country(42, 'Bulgaria', 'Bulgaria', 'BGR'), 43: Country(43, 'Turkey', 'Turkey', 'TUR'),
44: Country(44, 'Greece', 'Greece', 'GRC'), 45: Country(45, 'Japan', 'Japan', 'JPN'),
47: Country(47, 'South Korea', 'South-Korea', 'KOR'), 48: Country(48, 'India', 'India', 'IND'),
49: Country(49, 'Indonesia', 'Indonesia', 'IDN'), 50: Country(50, 'Australia', 'Australia', 'AUS'),
51: Country(51, 'South Africa', 'South-Africa', 'ZAF'),
52: Country(52, 'Republic of Moldova', 'Republic-of-Moldova', 'MDA'),
53: Country(53, 'Portugal', 'Portugal', 'PRT'), 54: Country(54, 'Ireland', 'Ireland', 'IRL'),
55: Country(55, 'Denmark', 'Denmark', 'DNK'), 56: Country(56, 'Iran', 'Iran', 'IRN'),
57: Country(57, 'Pakistan', 'Pakistan', 'PAK'), 58: Country(58, 'Israel', 'Israel', 'ISR'),
59: Country(59, 'Thailand', 'Thailand', 'THA'), 61: Country(61, 'Slovenia', 'Slovenia', 'SVN'),
63: Country(63, 'Croatia', 'Croatia', 'HRV'), 64: Country(64, 'Chile', 'Chile', 'CHL'),
65: Country(65, 'Serbia', 'Serbia', 'SRB'), 66: Country(66, 'Malaysia', 'Malaysia', 'MYS'),
67: Country(67, 'Philippines', 'Philippines', 'PHL'), 68: Country(68, 'Singapore', 'Singapore', 'SGP'),
69: Country(69, 'Bosnia and Herzegovina', 'Bosnia-Herzegovina', 'BiH'),
70: Country(70, 'Estonia', 'Estonia', 'EST'), 80: Country(80, 'Montenegro', 'Montenegro', 'MNE'),
71: Country(71, 'Latvia', 'Latvia', 'LVA'), 72: Country(72, 'Lithuania', 'Lithuania', 'LTU'),
73: Country(73, 'North Korea', 'North-Korea', 'PRK'), 74: Country(74, 'Uruguay', 'Uruguay', 'URY'),
75: Country(75, 'Paraguay', 'Paraguay', 'PRY'), 76: Country(76, 'Bolivia', 'Bolivia', 'BOL'),
77: Country(77, 'Peru', 'Peru', 'PER'), 78: Country(78, 'Colombia', 'Colombia', 'COL'),
79: Country(79, 'Republic of Macedonia (FYROM)', 'Republic-of-Macedonia-FYROM', 'MKD'),
81: Country(81, 'Republic of China (Taiwan)', 'Republic-of-China-Taiwan', 'TWN'),
82: Country(82, 'Cyprus', 'Cyprus', 'CYP'), 167: Country(167, 'Albania', 'Albania', 'ALB'),
83: Country(83, 'Belarus', 'Belarus', 'BLR'), 84: Country(84, 'New Zealand', 'New-Zealand', 'NZL'),
164: Country(164, 'Saudi Arabia', 'Saudi-Arabia', 'SAU'), 165: Country(165, 'Egypt', 'Egypt', 'EGY'),
166: Country(166, 'United Arab Emirates', 'United-Arab-Emirates', 'UAE'),
168: Country(168, 'Georgia', 'Georgia', 'GEO'), 169: Country(169, 'Armenia', 'Armenia', 'ARM'),
170: Country(170, 'Nigeria', 'Nigeria', 'NGA'), 171: Country(171, 'Cuba', 'Cuba', 'CUB')
1: Country(1, "Romania", "Romania", "ROU"),
9: Country(9, "Brazil", "Brazil", "BRA"),
10: Country(10, "Italy", "Italy", "ITA"),
11: Country(11, "France", "France", "FRA"),
12: Country(12, "Germany", "Germany", "DEU"),
13: Country(13, "Hungary", "Hungary", "HUN"),
14: Country(14, "China", "China", "CHN"),
15: Country(15, "Spain", "Spain", "ESP"),
23: Country(23, "Canada", "Canada", "CAN"),
24: Country(24, "USA", "USA", "USA"),
26: Country(26, "Mexico", "Mexico", "MEX"),
27: Country(27, "Argentina", "Argentina", "ARG"),
28: Country(28, "Venezuela", "Venezuela", "VEN"),
29: Country(29, "United Kingdom", "United-Kingdom", "GBR"),
30: Country(30, "Switzerland", "Switzerland", "CHE"),
31: Country(31, "Netherlands", "Netherlands", "NLD"),
32: Country(32, "Belgium", "Belgium", "BEL"),
33: Country(33, "Austria", "Austria", "AUT"),
34: Country(34, "Czech Republic", "Czech-Republic", "CZE"),
35: Country(35, "Poland", "Poland", "POL"),
36: Country(36, "Slovakia", "Slovakia", "SVK"),
37: Country(37, "Norway", "Norway", "NOR"),
38: Country(38, "Sweden", "Sweden", "SWE"),
39: Country(39, "Finland", "Finland", "FIN"),
40: Country(40, "Ukraine", "Ukraine", "UKR"),
41: Country(41, "Russia", "Russia", "RUS"),
42: Country(42, "Bulgaria", "Bulgaria", "BGR"),
43: Country(43, "Turkey", "Turkey", "TUR"),
44: Country(44, "Greece", "Greece", "GRC"),
45: Country(45, "Japan", "Japan", "JPN"),
47: Country(47, "South Korea", "South-Korea", "KOR"),
48: Country(48, "India", "India", "IND"),
49: Country(49, "Indonesia", "Indonesia", "IDN"),
50: Country(50, "Australia", "Australia", "AUS"),
51: Country(51, "South Africa", "South-Africa", "ZAF"),
52: Country(52, "Republic of Moldova", "Republic-of-Moldova", "MDA"),
53: Country(53, "Portugal", "Portugal", "PRT"),
54: Country(54, "Ireland", "Ireland", "IRL"),
55: Country(55, "Denmark", "Denmark", "DNK"),
56: Country(56, "Iran", "Iran", "IRN"),
57: Country(57, "Pakistan", "Pakistan", "PAK"),
58: Country(58, "Israel", "Israel", "ISR"),
59: Country(59, "Thailand", "Thailand", "THA"),
61: Country(61, "Slovenia", "Slovenia", "SVN"),
63: Country(63, "Croatia", "Croatia", "HRV"),
64: Country(64, "Chile", "Chile", "CHL"),
65: Country(65, "Serbia", "Serbia", "SRB"),
66: Country(66, "Malaysia", "Malaysia", "MYS"),
67: Country(67, "Philippines", "Philippines", "PHL"),
68: Country(68, "Singapore", "Singapore", "SGP"),
69: Country(69, "Bosnia and Herzegovina", "Bosnia-Herzegovina", "BiH"),
70: Country(70, "Estonia", "Estonia", "EST"),
80: Country(80, "Montenegro", "Montenegro", "MNE"),
71: Country(71, "Latvia", "Latvia", "LVA"),
72: Country(72, "Lithuania", "Lithuania", "LTU"),
73: Country(73, "North Korea", "North-Korea", "PRK"),
74: Country(74, "Uruguay", "Uruguay", "URY"),
75: Country(75, "Paraguay", "Paraguay", "PRY"),
76: Country(76, "Bolivia", "Bolivia", "BOL"),
77: Country(77, "Peru", "Peru", "PER"),
78: Country(78, "Colombia", "Colombia", "COL"),
79: Country(79, "Republic of Macedonia (FYROM)", "Republic-of-Macedonia-FYROM", "MKD"),
81: Country(81, "Republic of China (Taiwan)", "Republic-of-China-Taiwan", "TWN"),
82: Country(82, "Cyprus", "Cyprus", "CYP"),
167: Country(167, "Albania", "Albania", "ALB"),
83: Country(83, "Belarus", "Belarus", "BLR"),
84: Country(84, "New Zealand", "New-Zealand", "NZL"),
164: Country(164, "Saudi Arabia", "Saudi-Arabia", "SAU"),
165: Country(165, "Egypt", "Egypt", "EGY"),
166: Country(166, "United Arab Emirates", "United-Arab-Emirates", "UAE"),
168: Country(168, "Georgia", "Georgia", "GEO"),
169: Country(169, "Armenia", "Armenia", "ARM"),
170: Country(170, "Nigeria", "Nigeria", "NGA"),
171: Country(171, "Cuba", "Cuba", "CUB"),
}
FOOD_ENERGY: Dict[str, int] = dict(q1=2, q2=4, q3=6, q4=8, q5=10, q6=12, q7=20)
GROUND_RANK_NAMES: Dict[int, str] = {
1: 'Recruit', 2: 'Private', 3: 'Private*', 4: 'Private**', 5: 'Private***', 6: 'Corporal', 7: 'Corporal*', 8: 'Corporal**', 9: 'Corporal***',
10: 'Sergeant', 11: 'Sergeant*', 12: 'Sergeant**', 13: 'Sergeant***', 14: 'Lieutenant', 15: 'Lieutenant*', 16: 'Lieutenant**', 17: 'Lieutenant***',
18: 'Captain', 19: 'Captain*', 20: 'Captain**', 21: 'Captain***', 22: 'Major', 23: 'Major*', 24: 'Major**', 25: 'Major***',
26: 'Commander', 27: 'Commander*', 28: 'Commander**', 29: 'Commander***', 30: 'Lt Colonel', 31: 'Lt Colonel*', 32: 'Lt Colonel**', 33: 'Lt Colonel***',
34: 'Colonel', 35: 'Colonel*', 36: 'Colonel**', 37: 'Colonel***', 38: 'General', 39: 'General*', 40: 'General**', 41: 'General***',
42: 'Field Marshal', 43: 'Field Marshal*', 44: 'Field Marshal**', 45: 'Field Marshal***', 46: 'Supreme Marshal', 47: 'Supreme Marshal*', 48: 'Supreme Marshal**', 49: 'Supreme Marshal***',
50: 'National Force', 51: 'National Force*', 52: 'National Force**', 53: 'National Force***', 54: 'World Class Force', 55: 'World Class Force*', 56: 'World Class Force**', 57: 'World Class Force***',
58: 'Legendary Force', 59: 'Legendary Force*', 60: 'Legendary Force**', 61: 'Legendary Force***', 62: 'God of War', 63: 'God of War*', 64: 'God of War**', 65: 'God of War***',
66: 'Titan', 67: 'Titan*', 68: 'Titan**', 69: 'Titan***',
70: 'Legends I', 71: 'Legends II', 72: 'Legends III', 73: 'Legends IV', 74: 'Legends V', 75: 'Legends VI', 76: 'Legends VII', 77: 'Legends VIII', 78: 'Legends IX', 79: 'Legends X',
80: 'Legends XI', 81: 'Legends XII', 82: 'Legends XIII', 83: 'Legends XIV', 84: 'Legends XV', 85: 'Legends XVI', 86: 'Legends XVII', 87: 'Legends XVIII', 88: 'Legends XIX', 89: 'Legends XX'
1: "Recruit",
2: "Private",
3: "Private*",
4: "Private**",
5: "Private***",
6: "Corporal",
7: "Corporal*",
8: "Corporal**",
9: "Corporal***",
10: "Sergeant",
11: "Sergeant*",
12: "Sergeant**",
13: "Sergeant***",
14: "Lieutenant",
15: "Lieutenant*",
16: "Lieutenant**",
17: "Lieutenant***",
18: "Captain",
19: "Captain*",
20: "Captain**",
21: "Captain***",
22: "Major",
23: "Major*",
24: "Major**",
25: "Major***",
26: "Commander",
27: "Commander*",
28: "Commander**",
29: "Commander***",
30: "Lt Colonel",
31: "Lt Colonel*",
32: "Lt Colonel**",
33: "Lt Colonel***",
34: "Colonel",
35: "Colonel*",
36: "Colonel**",
37: "Colonel***",
38: "General",
39: "General*",
40: "General**",
41: "General***",
42: "Field Marshal",
43: "Field Marshal*",
44: "Field Marshal**",
45: "Field Marshal***",
46: "Supreme Marshal",
47: "Supreme Marshal*",
48: "Supreme Marshal**",
49: "Supreme Marshal***",
50: "National Force",
51: "National Force*",
52: "National Force**",
53: "National Force***",
54: "World Class Force",
55: "World Class Force*",
56: "World Class Force**",
57: "World Class Force***",
58: "Legendary Force",
59: "Legendary Force*",
60: "Legendary Force**",
61: "Legendary Force***",
62: "God of War",
63: "God of War*",
64: "God of War**",
65: "God of War***",
66: "Titan",
67: "Titan*",
68: "Titan**",
69: "Titan***",
70: "Legends I",
71: "Legends II",
72: "Legends III",
73: "Legends IV",
74: "Legends V",
75: "Legends VI",
76: "Legends VII",
77: "Legends VIII",
78: "Legends IX",
79: "Legends X",
80: "Legends XI",
81: "Legends XII",
82: "Legends XIII",
83: "Legends XIV",
84: "Legends XV",
85: "Legends XVI",
86: "Legends XVII",
87: "Legends XVIII",
88: "Legends XIX",
89: "Legends XX",
}
GROUND_RANK_POINTS: Dict[int, int] = {
1: 0, 2: 15, 3: 45, 4: 80, 5: 120, 6: 170, 7: 250, 8: 350, 9: 450, 10: 600, 11: 800, 12: 1000,
13: 1400, 14: 1850, 15: 2350, 16: 3000, 17: 3750, 18: 5000, 19: 6500, 20: 9000, 21: 12000,
22: 15500, 23: 20000, 24: 25000, 25: 31000, 26: 40000, 27: 52000, 28: 67000, 29: 85000,
30: 110000, 31: 140000, 32: 180000, 33: 225000, 34: 285000, 35: 355000, 36: 435000, 37: 540000,
38: 660000, 39: 800000, 40: 950000, 41: 1140000, 42: 1350000, 43: 1600000, 44: 1875000,
45: 2185000, 46: 2550000, 47: 3000000, 48: 3500000, 49: 4150000, 50: 4900000, 51: 5800000,
52: 7000000, 53: 9000000, 54: 11500000, 55: 14500000, 56: 18000000, 57: 22000000, 58: 26500000,
59: 31500000, 60: 37000000, 61: 43000000, 62: 50000000, 63: 100000000, 64: 200000000,
65: 500000000, 66: 1000000000, 67: 2000000000, 68: 4000000000, 69: 10000000000, 70: 20000000000,
71: 30000000000, 72: 40000000000, 73: 50000000000, 74: 60000000000, 75: 70000000000,
76: 80000000000, 77: 90000000000, 78: 100000000000, 79: 110000000000, 80: 120000000000,
81: 130000000000, 82: 140000000000, 83: 150000000000, 84: 160000000000, 85: 170000000000,
86: 180000000000, 87: 190000000000, 88: 200000000000, 89: 210000000000
1: 0,
2: 15,
3: 45,
4: 80,
5: 120,
6: 170,
7: 250,
8: 350,
9: 450,
10: 600,
11: 800,
12: 1000,
13: 1400,
14: 1850,
15: 2350,
16: 3000,
17: 3750,
18: 5000,
19: 6500,
20: 9000,
21: 12000,
22: 15500,
23: 20000,
24: 25000,
25: 31000,
26: 40000,
27: 52000,
28: 67000,
29: 85000,
30: 110000,
31: 140000,
32: 180000,
33: 225000,
34: 285000,
35: 355000,
36: 435000,
37: 540000,
38: 660000,
39: 800000,
40: 950000,
41: 1140000,
42: 1350000,
43: 1600000,
44: 1875000,
45: 2185000,
46: 2550000,
47: 3000000,
48: 3500000,
49: 4150000,
50: 4900000,
51: 5800000,
52: 7000000,
53: 9000000,
54: 11500000,
55: 14500000,
56: 18000000,
57: 22000000,
58: 26500000,
59: 31500000,
60: 37000000,
61: 43000000,
62: 50000000,
63: 100000000,
64: 200000000,
65: 500000000,
66: 1000000000,
67: 2000000000,
68: 4000000000,
69: 10000000000,
70: 20000000000,
71: 30000000000,
72: 40000000000,
73: 50000000000,
74: 60000000000,
75: 70000000000,
76: 80000000000,
77: 90000000000,
78: 100000000000,
79: 110000000000,
80: 120000000000,
81: 130000000000,
82: 140000000000,
83: 150000000000,
84: 160000000000,
85: 170000000000,
86: 180000000000,
87: 190000000000,
88: 200000000000,
89: 210000000000,
}
GROUND_RANKS: Dict[int, Rank] = {i: Rank(i, GROUND_RANK_NAMES[i], GROUND_RANK_POINTS[i], False) for i in range(1, 90)}
INDUSTRIES = Industries()
TERRAINS: Dict[int, str] = {0: 'Standard', 1: 'Industrial', 2: 'Urban', 3: 'Suburbs', 4: 'Airport', 5: 'Plains',
6: 'Wasteland', 7: 'Mountains', 8: 'Beach', 9: 'Swamp', 10: 'Mud', 11: 'Hills',
12: 'Jungle', 13: 'Forest', 14: 'Desert'}
TERRAINS: Dict[int, str] = {
0: "Standard",
1: "Industrial",
2: "Urban",
3: "Suburbs",
4: "Airport",
5: "Plains",
6: "Wasteland",
7: "Mountains",
8: "Beach",
9: "Swamp",
10: "Mud",
11: "Hills",
12: "Jungle",
13: "Forest",
14: "Desert",
}

142
erepublik/utils.py
View File

@ -23,10 +23,33 @@ except ImportError:
import json
__all__ = [
'VERSION', 'calculate_hit', 'date_from_eday', 'eday_from_date', 'deprecation', 'get_final_hit_dmg', 'write_file',
'get_air_hit_dmg_value', 'get_file', 'get_ground_hit_dmg_value', 'get_sleep_seconds', 'good_timedelta', 'slugify',
'interactive_sleep', 'json', 'localize_dt', 'localize_timestamp', 'normalize_html_json', 'now', 'silent_sleep',
'json_decode_object_hook', 'json_load', 'json_loads', 'json_dump', 'json_dumps', 'b64json', 'ErepublikJSONEncoder',
"VERSION",
"calculate_hit",
"date_from_eday",
"eday_from_date",
"deprecation",
"get_final_hit_dmg",
"write_file",
"get_air_hit_dmg_value",
"get_file",
"get_ground_hit_dmg_value",
"get_sleep_seconds",
"good_timedelta",
"slugify",
"interactive_sleep",
"json",
"localize_dt",
"localize_timestamp",
"normalize_html_json",
"now",
"silent_sleep",
"json_decode_object_hook",
"json_load",
"json_loads",
"json_dump",
"json_dumps",
"b64json",
"ErepublikJSONEncoder",
]
VERSION: str = __version__
@ -107,7 +130,7 @@ def get_file(filepath: str) -> str:
file = Path(filepath)
if file.exists():
if file.is_dir():
return str(file / 'new_file.txt')
return str(file / "new_file.txt")
else:
version = 1
try:
@ -129,16 +152,16 @@ def get_file(filepath: str) -> str:
def write_file(filename: str, content: str) -> int:
filename = get_file(filename)
with open(filename, 'ab') as f:
with open(filename, "ab") as f:
ret = f.write(content.encode("utf-8"))
return ret
def normalize_html_json(js: str) -> str:
js = re.sub(r' \'(.*?)\'', lambda a: f'"{a.group(1)}"', js)
js = re.sub(r'(\d\d):(\d\d):(\d\d)', r'\1\2\3', js)
js = re.sub(r" \'(.*?)\'", lambda a: f'"{a.group(1)}"', js)
js = re.sub(r"(\d\d):(\d\d):(\d\d)", r"\1\2\3", js)
js = re.sub(r'([{\s,])(\w+)(:)(?!"})', r'\1"\2"\3', js)
js = re.sub(r',\s*}', '}', js)
js = re.sub(r",\s*}", "}", js)
return js
@ -151,47 +174,51 @@ def slugify(value, allow_unicode=False) -> str:
"""
value = str(value)
if allow_unicode:
value = unicodedata.normalize('NFKC', value)
value = unicodedata.normalize("NFKC", value)
else:
value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode('ascii')
value = re.sub(r'[^\w\s-]', '_', value).strip().lower()
return re.sub(r'[-\s]+', '-', value)
value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
value = re.sub(r"[^\w\s-]", "_", value).strip().lower()
return re.sub(r"[-\s]+", "-", value)
def calculate_hit(strength: float, rang: int, tp: bool, elite: bool, ne: bool, booster: int = 0,
weapon: int = 200, is_deploy: bool = False) -> Decimal:
def calculate_hit(
strength: float, rang: int, tp: bool, elite: bool, ne: bool, booster: int = 0, weapon: int = 200, is_deploy: bool = False
) -> Decimal:
dec = 3 if is_deploy else 0
base_str = (1 + Decimal(str(round(strength, 3))) / 400)
base_rnk = (1 + Decimal(str(rang / 5)))
base_wpn = (1 + Decimal(str(weapon / 100)))
base_str = 1 + Decimal(str(round(strength, 3))) / 400
base_rnk = 1 + Decimal(str(rang / 5))
base_wpn = 1 + Decimal(str(weapon / 100))
dmg = 10 * base_str * base_rnk * base_wpn
dmg = get_final_hit_dmg(dmg, rang, tp=tp, elite=elite, ne=ne, booster=booster)
return Decimal(round(dmg, dec))
def get_ground_hit_dmg_value(citizen_id: int, natural_enemy: bool = False, true_patriot: bool = False,
booster: int = 0, weapon_power: int = 200) -> Decimal:
r = requests.get(f'https://www.erepublik.com/en/main/citizen-profile-json/{citizen_id}').json()
rang = r['military']['militaryData']['ground']['rankNumber']
strength = r['military']['militaryData']['ground']['strength']
elite = r['citizenAttributes']['level'] > 100
def get_ground_hit_dmg_value(
citizen_id: int, natural_enemy: bool = False, true_patriot: bool = False, booster: int = 0, weapon_power: int = 200
) -> Decimal:
r = requests.get(f"https://www.erepublik.com/en/main/citizen-profile-json/{citizen_id}").json()
rang = r["military"]["militaryData"]["ground"]["rankNumber"]
strength = r["military"]["militaryData"]["ground"]["strength"]
elite = r["citizenAttributes"]["level"] > 100
if natural_enemy:
true_patriot = True
return calculate_hit(strength, rang, true_patriot, elite, natural_enemy, booster, weapon_power)
def get_air_hit_dmg_value(citizen_id: int, natural_enemy: bool = False, true_patriot: bool = False, booster: int = 0,
weapon_power: int = 0) -> Decimal:
r = requests.get(f'https://www.erepublik.com/en/main/citizen-profile-json/{citizen_id}').json()
rang = r['military']['militaryData']['aircraft']['rankNumber']
elite = r['citizenAttributes']['level'] > 100
def get_air_hit_dmg_value(
citizen_id: int, natural_enemy: bool = False, true_patriot: bool = False, booster: int = 0, weapon_power: int = 0
) -> Decimal:
r = requests.get(f"https://www.erepublik.com/en/main/citizen-profile-json/{citizen_id}").json()
rang = r["military"]["militaryData"]["aircraft"]["rankNumber"]
elite = r["citizenAttributes"]["level"] > 100
return calculate_hit(0, rang, true_patriot, elite, natural_enemy, booster, weapon_power)
def get_final_hit_dmg(base_dmg: Union[Decimal, float, str], rang: int,
tp: bool = False, elite: bool = False, ne: bool = False, booster: int = 0) -> Decimal:
def get_final_hit_dmg(
base_dmg: Union[Decimal, float, str], rang: int, tp: bool = False, elite: bool = False, ne: bool = False, booster: int = 0
) -> Decimal:
dmg = Decimal(str(base_dmg))
if elite:
@ -216,18 +243,18 @@ def json_decode_object_hook(
:param o:
:return: Union[Dict[str, Any], List[Any], int, float, str, datetime.date, datetime.datetime, datetime.timedelta]
"""
if o.get('__type__'):
_type = o.get('__type__')
if _type == 'datetime':
dt = datetime.datetime.strptime(f"{o['date']} {o['time']}", '%Y-%m-%d %H:%M:%S')
if o.get('tzinfo'):
dt = pytz.timezone(o['tzinfo']).localize(dt)
if o.get("__type__"):
_type = o.get("__type__")
if _type == "datetime":
dt = datetime.datetime.strptime(f"{o['date']} {o['time']}", "%Y-%m-%d %H:%M:%S")
if o.get("tzinfo"):
dt = pytz.timezone(o["tzinfo"]).localize(dt)
return dt
elif _type == 'date':
dt = datetime.datetime.strptime(f"{o['date']}", '%Y-%m-%d')
elif _type == "date":
dt = datetime.datetime.strptime(f"{o['date']}", "%Y-%m-%d")
return dt.date()
elif _type == 'timedelta':
return datetime.timedelta(seconds=o['total_seconds'])
elif _type == "timedelta":
return datetime.timedelta(seconds=o["total_seconds"])
return o
@ -242,20 +269,20 @@ def json_loads(s: str, **kwargs):
def json_dump(obj, fp, *args, **kwargs):
if not kwargs.get('cls'):
if not kwargs.get("cls"):
kwargs.update(cls=ErepublikJSONEncoder)
return json.dump(obj, fp, *args, **kwargs)
def json_dumps(obj, *args, **kwargs):
if not kwargs.get('cls'):
if not kwargs.get("cls"):
kwargs.update(cls=ErepublikJSONEncoder)
return json.dumps(obj, *args, **kwargs)
def b64json(obj: Union[Dict[str, Union[int, List[str]]], List[str]]):
if isinstance(obj, list):
return b64encode(json.dumps(obj, separators=(',', ':')).encode('utf-8')).decode('utf-8')
return b64encode(json.dumps(obj, separators=(",", ":")).encode("utf-8")).decode("utf-8")
elif isinstance(obj, (int, str)):
return obj
elif isinstance(obj, dict):
@ -263,27 +290,34 @@ def b64json(obj: Union[Dict[str, Union[int, List[str]]], List[str]]):
obj[k] = b64json(v)
else:
from .classes import ErepublikException
raise ErepublikException(f'Unhandled object type! obj is {type(obj)}')
return b64encode(json.dumps(obj, separators=(',', ':')).encode('utf-8')).decode('utf-8')
raise ErepublikException(f"Unhandled object type! obj is {type(obj)}")
return b64encode(json.dumps(obj, separators=(",", ":")).encode("utf-8")).decode("utf-8")
class ErepublikJSONEncoder(json.JSONEncoder):
def default(self, o):
try:
from erepublik.citizen import Citizen
if isinstance(o, Decimal):
return float(f"{o:.02f}")
elif isinstance(o, datetime.datetime):
return dict(__type__='datetime', date=o.strftime("%Y-%m-%d"), time=o.strftime("%H:%M:%S"),
tzinfo=str(o.tzinfo) if o.tzinfo else None)
return dict(
__type__="datetime",
date=o.strftime("%Y-%m-%d"),
time=o.strftime("%H:%M:%S"),
tzinfo=str(o.tzinfo) if o.tzinfo else None,
)
elif isinstance(o, datetime.date):
return dict(__type__='date', date=o.strftime("%Y-%m-%d"))
return dict(__type__="date", date=o.strftime("%Y-%m-%d"))
elif isinstance(o, datetime.timedelta):
return dict(__type__='timedelta', days=o.days, seconds=o.seconds,
microseconds=o.microseconds, total_seconds=o.total_seconds())
return dict(
__type__="timedelta", days=o.days, seconds=o.seconds, microseconds=o.microseconds, total_seconds=o.total_seconds()
)
elif isinstance(o, Response):
return dict(headers=dict(o.__dict__['headers']), url=o.url, text=o.text, status_code=o.status_code)
elif hasattr(o, 'as_dict'):
return dict(headers=dict(o.__dict__["headers"]), url=o.url, text=o.text, status_code=o.status_code)
elif hasattr(o, "as_dict"):
return o.as_dict
elif isinstance(o, set):
return list(o)
@ -291,7 +325,7 @@ class ErepublikJSONEncoder(json.JSONEncoder):
return o.to_json()
elif isinstance(o, Logger):
return str(o)
elif hasattr(o, '__dict__'):
elif hasattr(o, "__dict__"):
return o.__dict__
else:
return super().default(o)

4
pyproject.toml Normal file
View File

@ -0,0 +1,4 @@
[tool.black]
line-length = 140
target-version = ['py38', 'py39']

10
setup.cfg
View File

@ -19,11 +19,13 @@ universal = 1
[flake8]
exclude = docs,.git,log,debug,venv
max-line-length = 240
ignore = D100,D101,D102,D103
line_length = 140
max-line-length = 140
ignore = D100,D101,D102,D103,E203
[pycodestyle]
max-line-length = 240
line_length = 140
max-line-length = 140
exclude = .git,log,debug,venv, build
[mypy]
@ -36,4 +38,4 @@ warn_unused_configs = True
[isort]
multi_line_output = 2
line_length = 240
line_length = 140

81
tests/test_erepublik_script.py
View File

@ -17,14 +17,13 @@ class TestErepublik(unittest.TestCase):
self.citizen.config.interactive = False
def test_should_do_levelup(self):
self.citizen.energy.recovered = 1900
self.citizen.energy.recoverable = 2940
self.citizen.energy.energy = 5950
self.citizen.energy.interval = 30
self.citizen.energy.limit = 3000
self.citizen.energy.limit = 6000
self.citizen.details.xp = 14850
self.assertTrue(self.citizen.should_do_levelup)
self.citizen.energy.recoverable = 1000
self.citizen.energy.energy = 1000
self.assertFalse(self.citizen.should_do_levelup)
def test_should_travel_to_fight(self):
@ -33,40 +32,38 @@ class TestErepublik(unittest.TestCase):
self.citizen.config.always_travel = False
self.assertFalse(self.citizen.should_travel_to_fight())
self.citizen.energy.recovered = 1900
self.citizen.energy.recoverable = 2940
self.citizen.energy.energy = 5960
self.citizen.energy.interval = 30
self.citizen.energy.limit = 3000
self.citizen.energy.limit = 6000
self.citizen.details.xp = 14850
self.assertTrue(self.citizen.should_travel_to_fight())
self.citizen.details.xp = 15000
self.citizen.energy.energy = 5000
self.assertFalse(self.citizen.should_travel_to_fight())
self.citizen.energy.recovered = 3000
self.citizen.energy.recoverable = 2910
self.citizen.energy.energy = 5910
self.assertTrue(self.citizen.should_travel_to_fight())
self.citizen.energy.recoverable = 2900
self.citizen.energy.energy = 5900
self.assertFalse(self.citizen.should_travel_to_fight())
# self.citizen.next_reachable_energy and self.citizen.config.next_energy
self.citizen.config.next_energy = True
self.citizen.energy.limit = 5000
self.citizen.energy.limit = 10000
self.citizen.details.next_pp = [5000, 5250, 5750, 6250, 6750]
self.citizen.details.pp = 4900
self.citizen.energy.recovered = 4000
self.citizen.energy.recoverable = 4510
self.citizen.energy.energy = 8510
self.assertEqual(self.citizen.next_reachable_energy, 850)
self.citizen.energy.recoverable = 4490
self.citizen.energy.energy = 8490
self.assertTrue(self.citizen.should_travel_to_fight())
self.assertEqual(self.citizen.next_reachable_energy, 350)
self.citizen.energy.recovered = 100
self.citizen.energy.recoverable = 150
self.citizen.energy.energy = 250
self.assertFalse(self.citizen.should_travel_to_fight())
self.assertEqual(self.citizen.next_reachable_energy, 0)
def test_should_fight(self):
def is_wc_close():
return self.citizen.max_time_till_full_ff > self.citizen.time_till_week_change
self.citizen.config.fight = False
self.assertEqual(self.citizen.should_fight(), (0, "Fighting not allowed!", False))
@ -76,62 +73,64 @@ class TestErepublik(unittest.TestCase):
self.citizen.energy.limit = 3000
self.citizen.details.xp = 24705
if not is_wc_close:
self.assertEqual(self.citizen.should_fight(), (0, 'Level up', False))
self.assertEqual(self.citizen.should_fight(), (0, "Level up", False))
self.citizen.energy.recovered = 3000
self.citizen.energy.recoverable = 2950
self.citizen.energy.energy = 5950
self.citizen.energy.interval = 30
self.assertEqual(self.citizen.should_fight(), (900, 'Level up', True))
self.assertEqual(self.citizen.should_fight(), (900, "Level up", True))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (900, 'Level up', True))
self.assertEqual(self.citizen.should_fight(), (900, "Level up", True))
self.citizen.my_companies.ff_lockdown = 0
# Level up reachable
self.citizen.details.xp = 24400
self.assertEqual(self.citizen.should_fight(), (305, 'Fighting for close Levelup. Doing 305 hits', True))
self.assertEqual(self.citizen.should_fight(), (305, "Fighting for close Levelup. Doing 305 hits", True))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (305, 'Fighting for close Levelup. Doing 305 hits', True))
self.assertEqual(self.citizen.should_fight(), (305, "Fighting for close Levelup. Doing 305 hits", True))
self.citizen.my_companies.ff_lockdown = 0
self.citizen.details.xp = 21000
self.assertEqual(self.citizen.should_fight(), (75, 'Obligatory fighting for at least 75pp', True))
self.assertEqual(self.citizen.should_fight(), (75, "Obligatory fighting for at least 75pp", True))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (75, 'Obligatory fighting for at least 75pp', True))
self.assertEqual(self.citizen.should_fight(), (75, "Obligatory fighting for at least 75pp", True))
self.citizen.my_companies.ff_lockdown = 0
self.citizen.details.pp = 80
# All-in (type = all-in and full ff)
self.citizen.config.all_in = True
self.assertEqual(self.citizen.should_fight(), (595, 'Fighting all-in. Doing 595 hits', False))
self.assertEqual(self.citizen.should_fight(), (595, "Fighting all-in. Doing 595 hits", False))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (
435, 'Fight count modified (old count: 595 | FF: 595 | WAM ff_lockdown: 160 | New count: 435)', False
))
self.assertEqual(
self.citizen.should_fight(),
(435, "Fight count modified (old count: 595 | FF: 595 | WAM ff_lockdown: 160 | New count: 435)", False),
)
self.citizen.my_companies.ff_lockdown = 0
self.citizen.config.air = True
self.citizen.energy.recoverable = 1000
self.assertEqual(self.citizen.should_fight(), (400, 'Fighting all-in in AIR. Doing 400 hits', False))
self.citizen.energy.energy = 4000
self.assertEqual(self.citizen.should_fight(), (400, "Fighting all-in in AIR. Doing 400 hits", False))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (
240, 'Fight count modified (old count: 400 | FF: 400 | WAM ff_lockdown: 160 | New count: 240)', False
))
self.assertEqual(
self.citizen.should_fight(),
(240, "Fight count modified (old count: 400 | FF: 400 | WAM ff_lockdown: 160 | New count: 240)", False),
)
self.citizen.my_companies.ff_lockdown = 0
self.citizen.config.all_in = False
self.citizen.config.next_energy = True
self.citizen.energy.limit = 5000
self.citizen.energy.limit = 10000
self.citizen.details.next_pp = [100, 150, 250, 400, 500]
self.assertEqual(self.citizen.should_fight(), (320, 'Fighting for +1 energy. Doing 320 hits', False))
self.assertEqual(self.citizen.should_fight(), (320, "Fighting for +1 energy. Doing 320 hits", False))
self.citizen.my_companies.ff_lockdown = 160
self.assertEqual(self.citizen.should_fight(), (
160, 'Fight count modified (old count: 320 | FF: 400 | WAM ff_lockdown: 160 | New count: 160)', False
))
self.assertEqual(
self.citizen.should_fight(),
(160, "Fight count modified (old count: 320 | FF: 400 | WAM ff_lockdown: 160 | New count: 160)", False),
)
self.citizen.my_companies.ff_lockdown = 0
self.citizen.energy.limit = 3000
self.citizen.details.next_pp = [19250, 20000]
self.citizen.config.next_energy = False
# 1h worth of energy
self.citizen.energy.recoverable = 2910
self.assertEqual(self.citizen.should_fight(), (30, 'Fighting for 1h energy. Doing 30 hits', True))
self.citizen.energy.energy = 5910
self.assertEqual(self.citizen.should_fight(), (30, "Fighting for 1h energy. Doing 30 hits", True))