Move eRepublik API helper class to seperate file and split up class into logical pieces

This commit is contained in:
Eriks Karls
2020-02-24 17:19:06 +02:00
parent a5dfc07018
commit ce02a39158
2 changed files with 642 additions and 599 deletions

640
erepublik/access_points.py Normal file
View File

@ -0,0 +1,640 @@
import datetime
import random
import time
from typing import Mapping, Any, Union, Dict, List
from requests import Response, Session
from erepublik import utils
__all__ = ['SlowRequests', 'CitizenAPI']
class SlowRequests(Session):
last_time: datetime.datetime
timeout = datetime.timedelta(milliseconds=500)
uas = [
# Chrome
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.106 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.106 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
# FireFox
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:72.0) Gecko/20100101 Firefox/72.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:71.0) Gecko/20100101 Firefox/71.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:70.0) Gecko/20100101 Firefox/70.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:72.0) Gecko/20100101 Firefox/72.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:71.0) Gecko/20100101 Firefox/71.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:70.0) Gecko/20100101 Firefox/70.0',
]
debug = False
def __init__(self):
super().__init__()
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': random.choice(self.uas)
})
@property
def __dict__(self):
return dict(last_time=self.last_time, timeout=self.timeout, user_agent=self.headers['User-Agent'],
request_log_name=self.request_log_name, debug=self.debug)
def request(self, method, url, *args, **kwargs):
self._slow_down_requests()
self._log_request(url, method, **kwargs)
resp = super().request(method, url, *args, **kwargs)
self._log_response(url, resp)
return resp
def _slow_down_requests(self):
ltt = utils.good_timedelta(self.last_time, self.timeout)
if ltt > utils.now():
seconds = (ltt - utils.now()).total_seconds()
time.sleep(seconds if seconds > 0 else 0)
self.last_time = utils.now()
def _log_request(self, url, method, data=None, json=None, params=None, **kwargs):
if self.debug:
args = {}
kwargs.pop('allow_redirects', None)
if kwargs:
args.update({'kwargs': kwargs})
if data:
args.update({"data": data})
if json:
args.update({"json": json})
if params:
args.update({"params": params})
body = "[{dt}]\tURL: '{url}'\tMETHOD: {met}\tARGS: {args}\n".format(dt=utils.now().strftime("%F %T"),
url=url, met=method, args=args)
utils.get_file(self.request_log_name)
with open(self.request_log_name, 'ab') as file:
file.write(body.encode("UTF-8"))
def _log_response(self, url, resp, redirect: bool = False):
from erepublik import Citizen
if self.debug:
if resp.history and not redirect:
for hist_resp in resp.history:
self._log_request(hist_resp.request.url, "REDIRECT")
self._log_response(hist_resp.request.url, hist_resp, redirect=True)
file_data = {
"path": 'debug/requests',
"time": self.last_time.strftime('%Y/%m/%d/%H-%M-%S'),
"name": utils.slugify(url[len(Citizen.url):]),
"extra": "_REDIRECT" if redirect else ""
}
try:
utils.json.loads(resp.text)
file_data.update({"ext": "json"})
except utils.json.JSONDecodeError:
file_data.update({"ext": "html"})
filename = 'debug/requests/{time}_{name}{extra}.{ext}'.format(**file_data)
with open(utils.get_file(filename), 'wb') as f:
f.write(resp.text.encode('utf-8'))
class CitizenBaseAPI:
url: str = "https://www.erepublik.com/en"
_req: SlowRequests = None
token: str = ""
def __init__(self):
"""
Class for unifying eRepublik known endpoints and their required/optional parameters
"""
self._req = SlowRequests()
def post(self, url: str, data=None, json=None, **kwargs) -> Response:
return self._req.post(url, data, json, **kwargs)
def get(self, url: str, **kwargs) -> Response:
return self._req.get(url, **kwargs)
def _get_main(self) -> Response:
return self.get(self.url)
class ErepublikAnniversaryAPI(CitizenBaseAPI):
def _post_main_collect_anniversary_reward(self) -> Response:
return self.post("{}/main/collect-anniversary-reward".format(self.url), data={"_token": self.token})
# 12th anniversary endpoints
def _get_anniversary_quest_data(self) -> Response:
return self.get("{}/main/anniversaryQuestData".format(self.url))
def _post_map_rewards_unlock(self, node_id: int) -> Response:
data = {'nodeId': node_id, '_token': self.token}
return self.post("{}/main/map-rewards-unlock".format(self.url), 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}
return self.post("{}/main/map-rewards-speedup".format(self.url), data=data)
def _post_map_rewards_claim(self, node_id: int) -> Response:
data = {'nodeId': node_id, '_token': self.token}
return self.post("{}/main/map-rewards-claim".format(self.url), data=data)
class ErepublikArticleAPI(CitizenBaseAPI):
def _get_main_article_json(self, article_id: int) -> Response:
return self.get("{}/main/articleJson/{}".format(self.url, article_id))
def _post_main_article_comments(self, article: int, page: int = 1) -> Response:
data = dict(_token=self.token, articleId=article, page=page)
if page:
data.update({'page': page})
return self.post("{}/main/articleComments".format(self.url), data=data)
def _post_main_article_comments_create(self, message: str, article: int, parent: int = 0) -> Response:
data = dict(_token=self.token, message=message, articleId=article)
if parent:
data.update({"parentId": parent})
return self.post("{}/main/articleComments/create".format(self.url), data=data)
def _post_main_donate_article(self, article_id: int, amount: int) -> Response:
data = dict(_token=self.token, articleId=article_id, amount=amount)
return self.post("{}/main/donate-article".format(self.url), data=data)
def _post_main_write_article(self, title: str, content: str, location: int, kind: int) -> Response:
data = dict(_token=self.token, article_name=title, article_body=content, article_location=location,
article_category=kind)
return self.post("{}/main/write-article".format(self.url), data=data)
def _post_main_vote_article(self, article_id: int) -> Response:
data = dict(_token=self.token, articleId=article_id)
return self.post("{}/main/vote-article".format(self.url), data=data)
class ErepublikCompanyAPI(CitizenBaseAPI):
def _post_economy_assign_to_holding(self, factory: int, holding: int) -> Response:
data = dict(_token=self.token, factoryId=factory, action="assign", holdingCompanyId=holding)
return self.post("{}/economy/assign-to-holding".format(self.url), data=data)
def _post_economy_create_company(self, industry: int, building_type: int = 1) -> Response:
data = {"_token": self.token, "company[industry_id]": industry, "company[building_type]": building_type}
return self.post("{}/economy/create-company".format(self.url), data=data,
headers={"Referer": "{}/economy/create-company".format(self.url)})
def _get_economy_inventory_items(self) -> Response:
return self.get("{}/economy/inventory-items/".format(self.url))
def _get_economy_job_market_json(self, country: int) -> Response:
return self.get("{}/economy/job-market-json/{}/1/desc".format(self.url, country))
def _get_economy_my_companies(self) -> Response:
return self.get("{}/economy/myCompanies".format(self.url))
def _post_economy_train(self, tg_ids: List[int]) -> Response:
data: Dict[str, Union[int, str]] = {}
for idx, tg_id in enumerate(tg_ids):
data["grounds[%i][id]" % idx] = tg_id
data["grounds[%i][train]" % idx] = 1
if data:
data['_token'] = self.token
return self.post("{}/economy/train".format(self.url), 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)
return self.post("{}/economy/upgrade-company".format(self.url), data=data)
def _post_economy_work(self, action_type: str, wam: List[int] = None, employ: Dict[int, int] = None) -> Response:
"""
:return: requests.Response or None
"""
if employ is None:
employ = {}
if wam is None:
wam = []
data: Dict[str, Union[int, str]] = dict(action_type=action_type, _token=self.token)
if action_type == "production":
max_idx = 0
for company_id in sorted(wam or []):
data.update({
"companies[%i][id]" % max_idx: company_id,
"companies[%i][employee_works]" % max_idx: employ.pop(company_id, 0),
"companies[%i][own_work]" % max_idx: 1
})
max_idx += 1
for company_id in sorted(employ or []):
data.update({
"companies[%i][id]" % max_idx: company_id,
"companies[%i][employee_works]" % max_idx: employ.pop(company_id),
"companies[%i][own_work]" % max_idx: 0
})
max_idx += 1
return self.post("{}/economy/work".format(self.url), data=data)
def _post_economy_work_overtime(self) -> Response:
data = dict(action_type="workOvertime", _token=self.token)
return self.post("{}/economy/workOvertime".format(self.url), data=data)
def _post_economy_job_market_apply(self, citizen: int, salary: int) -> Response:
data = dict(_token=self.token, citizenId=citizen, salary=salary)
return self.post("{}/economy/job-market-apply".format(self.url), data=data)
def _post_economy_resign(self) -> Response:
return self.post("{}/economy/resign".format(self.url),
headers={"Content-Type": "application/x-www-form-urlencoded"},
data={"_token": self.token, "action_type": "resign"})
def _post_economy_sell_company(self, factory: 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"})
else:
data.update({"dissolve": factory})
return self.post("{}/economy/sell-company/{}".format(self.url, factory),
data=data, headers={"Referer": self.url})
class ErepublikCountryAPI(CitizenBaseAPI):
def _get_country_military(self, country: str) -> Response:
return self.get("{}/country/military/{}".format(self.url, country))
def _post_main_country_donate(self, country: int, action: str, value: Union[int, float],
quality: int = None) -> Response:
json = dict(countryId=country, action=action, _token=self.token, value=value, quality=quality)
return self.post("{}/main/country-donate".format(self.url), data=json,
headers={"Referer": "{}/country/economy/Latvia".format(self.url)})
class ErepublikEconomyAPI(CitizenBaseAPI):
def _get_economy_citizen_accounts(self, organisation_id: int) -> Response:
return self.get("{}/economy/citizen-accounts/{}".format(self.url, organisation_id))
def _get_economy_my_market_offers(self) -> Response:
return self.get("{}/economy/myMarketOffers".format(self.url))
def _get_main_job_data(self) -> Response:
return self.get("{}/main/job-data".format(self.url))
def _post_main_buy_gold_items(self, currency: str, item: str, amount: int) -> Response:
data = dict(itemId=item, currency=currency, amount=amount, _token=self.token)
return self.post("{}/main/buyGoldItems".format(self.url), data=data)
def _post_economy_activate_booster(self, quality: int, duration: int, kind: str) -> Response:
data = dict(type=kind, quality=quality, duration=duration, fromInventory=True)
return self.post("{}/economy/activateBooster".format(self.url), data=data)
def _post_economy_activate_house(self, quality: int) -> Response:
data = {"action": "activate", "quality": quality, "type": "house", "_token": self.token}
return self.post("{}/economy/activateHouse".format(self.url), data=data)
def _post_economy_donate_items_action(self, citizen: int, amount: int, industry: int,
quality: int) -> Response:
data = dict(citizen_id=citizen, amount=amount, industry_id=industry, quality=quality, _token=self.token)
return self.post("{}/economy/donate-items-action".format(self.url), data=data,
headers={"Referer": "{}/economy/donate-items/{}".format(self.url, citizen)})
def _post_economy_donate_money_action(self, citizen: int, amount: float = 0.0,
currency: int = 62) -> Response:
data = dict(citizen_id=citizen, _token=self.token, currency_id=currency, amount=amount)
return self.post("{}/economy/donate-money-action".format(self.url), data=data,
headers={"Referer": "{}/economy/donate-money/{}".format(self.url, citizen)})
def _post_economy_exchange_purchase(self, amount: float, currency: int, offer: int) -> Response:
data = dict(_token=self.token, amount=amount, currencyId=currency, offerId=offer)
return self.post("{}/economy/exchange/purchase/".format(self.url), data=data)
def _post_economy_exchange_retrieve(self, personal: bool, page: int, currency: int) -> Response:
data = dict(_token=self.token, personalOffers=int(personal), page=page, currencyId=currency)
return self.post("{}/economy/exchange/retrieve/".format(self.url), data=data)
def _post_economy_game_tokens_market(self, action: str) -> Response:
assert action in ['retrieve', ]
data = dict(_token=self.token, action=action)
return self.post("{}/economy/gameTokensMarketAjax".format(self.url), 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)
return self.post("{}/economy/marketplaceAjax".format(self.url), data=data)
def _post_economy_marketplace_actions(self, amount: int, buy: bool = False, **kwargs) -> Response:
if buy:
data = dict(_token=self.token, offerId=kwargs['offer'], amount=amount, orderBy="price_asc", currentPage=1,
buyAction=1)
else:
data = dict(_token=self.token, countryId=kwargs["country"], price=kwargs["price"],
industryId=kwargs["industry"], quality=kwargs["quality"], amount=amount, sellAction='postOffer')
return self.post("{}/economy/marketplaceActions".format(self.url), data=data)
class ErepublikLeaderboardAPI(CitizenBaseAPI):
def _get_main_leaderboards_damage_aircraft_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Response:
data = (country, weeks, mu)
return self.get("{}/main/leaderboards-damage-aircraft-rankings/{}/{}/{}/0".format(self.url, *data))
def _get_main_leaderboards_damage_rankings(self, country: int, weeks: int = 0, mu: int = 0,
div: int = 0) -> Response:
data = (country, weeks, mu, div)
return self.get("{}/main/leaderboards-damage-rankings/{}/{}/{}/{}".format(self.url, *data))
def _get_main_leaderboards_kills_aircraft_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Response:
data = (country, weeks, mu)
return self.get("{}/main/leaderboards-kills-aircraft-rankings/{}/{}/{}/0".format(self.url, *data))
def _get_main_leaderboards_kills_rankings(self, country: int, weeks: int = 0, mu: int = 0,
div: int = 0) -> Response:
data = (country, weeks, mu, div)
return self.get("{}/main/leaderboards-kills-rankings/{}/{}/{}/{}".format(self.url, *data))
class ErepublikLocationAPI(CitizenBaseAPI):
def _get_main_city_data_residents(self, city: int, page: int = 1, params: Mapping[str, Any] = None) -> Response:
if params is None:
params = {}
return self.get("{}/main/city-data/{}/residents".format(self.url, city), params={"currentPage": page, **params})
class ErepublikMilitaryAPI(CitizenBaseAPI):
def _get_military_battlefield_choose_side(self, battle: int, side: int) -> Response:
return self.get("{}/military/battlefield-choose-side/{}/{}".format(self.url, battle, side))
def _get_military_show_weapons(self, battle: int) -> Response:
return self.get("{}/military/show-weapons".format(self.url), params={'_token': self.token, 'battleId': battle})
def _get_military_campaigns(self) -> Response:
return self.get("{}/military/campaigns-new/".format(self.url))
def _get_military_campaigns_json_list(self) -> Response:
return self.get("{}/military/campaignsJson/list".format(self.url))
def _get_military_unit_data(self, unit_id: int, **kwargs) -> Response:
params = {"groupId": unit_id, "panel": "members", **kwargs}
return self.get("{}/military/military-unit-data/".format(self.url), params=params)
def _post_main_activate_battle_effect(self, battle: int, kind: str, citizen_id: int) -> Response:
data = dict(battleId=battle, citizenId=citizen_id, type=kind, _token=self.token)
return self.post("{}/main/fight-activateBattleEffect".format(self.url), data=data)
def _post_main_battlefield_travel(self, side_id: int, battle_id: int) -> Response:
data = dict(_token=self.token, sideCountryId=side_id, battleId=battle_id)
return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
def _post_main_battlefield_change_division(self, battle_id: int, division_id: int) -> Response:
data = dict(_token=self.token, battleZoneId=division_id, battleId=battle_id)
return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
def _get_wars_show(self, war_id: int) -> Response:
return self.get("{}/wars/show/{}".format(self.url, war_id))
def _post_military_fight_activate_booster(self, battle: int, quality: int, duration: int, kind: str) -> Response:
data = dict(type=kind, quality=quality, duration=duration, battleId=battle, _token=self.token)
return self.post("{}/military/fight-activateBooster".format(self.url), data=data)
def _post_military_change_weapon(self, battle: int, battle_zone: int, weapon_level: int, ) -> Response:
data = dict(battleId=battle, _token=self.token, battleZoneId=battle_zone, customizationLevel=weapon_level)
return self.post("{}/military/change-weapon".format(self.url), data=data)
def _post_military_battle_console(self, battle_id: int, action: str, page: int = 1, **kwargs) -> Response:
data = dict(battleId=battle_id, action=action, _token=self.token)
if action == "battleStatistics":
data.update(round=kwargs["round_id"], zoneId=kwargs["round_id"], leftPage=page, rightPage=page,
division=kwargs["division"], type=kwargs.get("type", 'damage'), )
elif action == "warList":
data.update(page=page)
return self.post("{}/military/battle-console".format(self.url), data=data)
def _post_military_deploy_bomb(self, battle_id: int, bomb_id: int) -> Response:
data = dict(battleId=battle_id, bombId=bomb_id, _token=self.token)
return self.post("{}/military/deploy-bomb".format(self.url), data=data)
def _post_military_fight_air(self, battle_id: int, side_id: int, zone_id: int) -> Response:
data = dict(sideId=side_id, battleId=battle_id, _token=self.token, battleZoneId=zone_id)
return self.post("{}/military/fight-shoooot/{}".format(self.url, battle_id), data=data)
def _post_military_fight_ground(self, battle_id: int, side_id: int, zone_id: int) -> Response:
data = dict(sideId=side_id, battleId=battle_id, _token=self.token, battleZoneId=zone_id)
return self.post("{}/military/fight-shooot/{}".format(self.url, battle_id), data=data)
class ErepublikPoliticsAPI(CitizenBaseAPI):
def _get_candidate_party(self, party_slug: str) -> Response:
return self.post("{}/candidate/{}".format(self.url, party_slug))
def _get_main_party_members(self, party: int) -> Response:
return self.get("{}/main/party-members/{}".format(self.url, party))
def _get_main_rankings_parties(self, country: int) -> Response:
return self.get("{}/main/rankings-parties/1/{}".format(self.url, country))
def _post_candidate_for_congress(self, presentation: str = "") -> Response:
data = dict(_token=self.token, presentation=presentation)
return self.post("{}/candidate-for-congress".format(self.url), data=data)
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('{}/wars/attack-region/{}/{}'.format(self.url, 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=utils.COUNTRY_LINK[attack_country_id])
return self.post("{}/{}/new-war".format(self.url, utils.COUNTRY_LINK[self_country_id]), 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)
return self.post("{}/{}/new-donation".format(self.url, utils.COUNTRY_LINK[country_id]), data=data)
class ErepublikProfileAPI(CitizenBaseAPI):
def _get_main_citizen_hovercard(self, citizen: int) -> Response:
return self.get("{}/main/citizen-hovercard/{}".format(self.url, citizen))
def _get_main_citizen_profile_json(self, player_id: int) -> Response:
return self.get("{}/main/citizen-profile-json/{}".format(self.url, player_id))
def _get_main_citizen_notifications(self) -> Response:
return self.get("{}/main/citizenDailyAssistant".format(self.url))
def _get_main_citizen_daily_assistant(self) -> Response:
return self.get("{}/main/citizenNotifications".format(self.url))
def _get_main_messages_paginated(self, page: int = 1) -> Response:
return self.get("{}/main/messages-paginated/{}".format(self.url, page))
def _get_main_money_donation_accept(self, donation_id: int) -> Response:
return self.get("{}/main/money-donation/accept/{}".format(self.url, donation_id), params={"_token": self.token})
def _get_main_money_donation_reject(self, donation_id: int) -> Response:
return self.get("{}/main/money-donation/reject/{}".format(self.url, donation_id), params={"_token": self.token})
def _get_main_notifications_ajax_community(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/community/{}".format(self.url, page))
def _get_main_notifications_ajax_system(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/system/{}".format(self.url, page))
def _get_main_notifications_ajax_report(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/report/{}".format(self.url, page))
def _get_main_training_grounds_json(self) -> Response:
return self.get("{}/main/training-grounds-json".format(self.url))
def _get_main_weekly_challenge_data(self) -> Response:
return self.get("{}/main/weekly-challenge-data".format(self.url))
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"})
else:
data.update({"action": "removeFriend"})
return self.post("{}/main/citizen-addRemoveFriend".format(self.url), data=data)
def _post_main_daily_task_reward(self) -> Response:
return self.post("{}/main/daily-tasks-reward".format(self.url), data=dict(_token=self.token))
def _post_delete_message(self, msg_id: list) -> Response:
data = {"_token": self.token, "delete_message[]": msg_id}
return self.post("{}/main/messages-delete".format(self.url), data)
def _post_eat(self, color: str) -> Response:
data = dict(_token=self.token, buttonColor=color)
return self.post("{}/main/eat".format(self.url), params=data)
def _post_main_global_alerts_close(self, alert_id: int) -> Response:
data = dict(_token=self.token, alert_id=alert_id)
return self.post("{}/main/global-alerts/close".format(self.url), data=data)
def _post_forgot_password(self, email: str) -> Response:
data = dict(_token=self.token, email=email, commit="Reset password")
return self.post("{}/forgot-password".format(self.url), 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')
return self.post("{}/login".format(self.url), 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"}
return self.post("{}/main/messages-alerts/1".format(self.url), 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)
return self.post("{}/main/messages-compose/{}".format(self.url, url_pk), data=data)
def _post_military_group_missions(self) -> Response:
data = dict(action="check", _token=self.token)
return self.post("{}/military/group-missions".format(self.url), data=data)
def _post_main_weekly_challenge_reward(self, reward_id: int) -> Response:
data = dict(_token=self.token, rewardId=reward_id)
return self.post("{}/main/weekly-challenge-collect-reward".format(self.url), data=data)
class ErepublikTravelAPI(CitizenBaseAPI):
def _post_main_travel(self, check: str, **kwargs) -> Response:
data = dict(_token=self.token, check=check, **kwargs)
return self.post("{}/main/travel".format(self.url), data=data)
def _post_main_travel_data(self, **kwargs) -> Response:
return self.post("{}/main/travelData".format(self.url), data=dict(_token=self.token, **kwargs))
class ErepublikWallPostAPI(CitizenBaseAPI):
# ## Country
def _post_main_country_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/country-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/country-comment/create/json".format(self.url), 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}
return self.post("{}/main/country-post/create/json".format(self.url), data=data)
def _post_main_country_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/country-post/retrieve/json".format(self.url), data=data)
# ## Military Unit
def _post_main_military_unit_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/military-unit-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/military-unit-comment/create/json".format(self.url), 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}
return self.post("{}/main/military-unit-post/create/json".format(self.url), data=data)
def _post_main_military_unit_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/military-unit-post/retrieve/json".format(self.url), data=data)
# ## Party
def _post_main_party_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/party-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/party-comment/create/json".format(self.url), data=data)
def _post_main_party_post_create(self, body: str) -> Response:
data = {"_token": self.token, "post_message": body}
return self.post("{}/main/party-post/create/json".format(self.url), data=data)
def _post_main_party_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/party-post/retrieve/json".format(self.url), data=data)
# ## Friend's Wall
def _post_main_wall_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/wall-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/wall-comment/create/json".format(self.url), data=data)
def _post_main_wall_post_create(self, body: str) -> Response:
data = {"_token": self.token, "post_message": body}
return self.post("{}/main/wall-post/create/json".format(self.url), data=data)
def _post_main_wall_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/wall-post/retrieve/json".format(self.url), data=data)
# ## Medal posting
def _post_main_wall_post_automatic(self, message: str, achievement_id: int) -> Response:
return self.post("{}/main/wall-post/automatic".format(self.url), 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
):
pass

View File

@ -1,15 +1,14 @@
import datetime
import decimal
import hashlib
import random
import threading
import time
from collections import defaultdict, deque
from typing import Any, Dict, Iterable, List, Mapping, Tuple, Union
from typing import Any, Dict, Iterable, List, Tuple, Union
from requests import Response, Session, post
from erepublik import utils
from erepublik.access_points import CitizenAPI
try:
import simplejson as json
@ -199,102 +198,6 @@ class MyCompanies:
# return ret
class SlowRequests(Session):
last_time: datetime.datetime
timeout = datetime.timedelta(milliseconds=500)
uas = [
# Chrome
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.132 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.132 Safari/537.36',
# FireFox
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:70.0) Gecko/20100101 Firefox/70.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:69.0) Gecko/20100101 Firefox/69.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:68.0) Gecko/20100101 Firefox/68.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:70.0) Gecko/20100101 Firefox/70.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:69.0) Gecko/20100101 Firefox/69.0',
'Mozilla/5.0 (X11; Linux x86_64; rv:68.0) Gecko/20100101 Firefox/68.0',
]
debug = False
def __init__(self):
super().__init__()
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': random.choice(self.uas)
})
@property
def __dict__(self):
return dict(last_time=self.last_time, timeout=self.timeout, user_agent=self.headers['User-Agent'],
request_log_name=self.request_log_name, debug=self.debug)
def request(self, method, url, *args, **kwargs):
self._slow_down_requests()
self._log_request(url, method, **kwargs)
resp = super().request(method, url, *args, **kwargs)
self._log_response(url, resp)
return resp
def _slow_down_requests(self):
ltt = utils.good_timedelta(self.last_time, self.timeout)
if ltt > utils.now():
seconds = (ltt - utils.now()).total_seconds()
time.sleep(seconds if seconds > 0 else 0)
self.last_time = utils.now()
def _log_request(self, url, method, data=None, json=None, params=None, **kwargs):
if self.debug:
args = {}
kwargs.pop('allow_redirects', None)
if kwargs:
args.update({'kwargs': kwargs})
if data:
args.update({"data": data})
if json:
args.update({"json": json})
if params:
args.update({"params": params})
body = "[{dt}]\tURL: '{url}'\tMETHOD: {met}\tARGS: {args}\n".format(dt=utils.now().strftime("%F %T"),
url=url, met=method, args=args)
utils.get_file(self.request_log_name)
with open(self.request_log_name, 'ab') as file:
file.write(body.encode("UTF-8"))
def _log_response(self, url, resp, redirect: bool = False):
from erepublik import Citizen
if self.debug:
if resp.history and not redirect:
for hist_resp in resp.history:
self._log_request(hist_resp.request.url, "REDIRECT")
self._log_response(hist_resp.request.url, hist_resp, redirect=True)
file_data = {
"path": 'debug/requests',
"time": self.last_time.strftime('%Y/%m/%d/%H-%M-%S'),
"name": utils.slugify(url[len(Citizen.url):]),
"extra": "_REDIRECT" if redirect else ""
}
try:
json.loads(resp.text)
file_data.update({"ext": "json"})
except json.JSONDecodeError:
file_data.update({"ext": "html"})
filename = 'debug/requests/{time}_{name}{extra}.{ext}'.format(**file_data)
with open(utils.get_file(filename), 'wb') as f:
f.write(resp.text.encode('utf-8'))
class Config:
email = ""
password = ""
@ -489,506 +392,6 @@ class House:
return self.active_untill
class CitizenAPI:
url: str = "https://www.erepublik.com/en"
_req: SlowRequests = None
token: str = ""
def __init__(self):
"""
Class for unifying eRepublik known endpoints and their required/optional parameters
"""
self._req = SlowRequests()
def post(self, url: str, data=None, json=None, **kwargs) -> Response:
return self._req.post(url, data, json, **kwargs)
def get(self, url: str, **kwargs) -> Response:
return self._req.get(url, **kwargs)
def _get_main_article_json(self, article_id: int) -> Response:
return self.get("{}/main/articleJson/{}".format(self.url, article_id))
def _get_military_battlefield_choose_side(self, battle: int, side: int) -> Response:
return self.get("{}/military/battlefield-choose-side/{}/{}".format(self.url, battle, side))
def _get_military_show_weapons(self, battle: int) -> Response:
return self.get("{}/military/show-weapons".format(self.url), params={'_token': self.token, 'battleId': battle})
def _get_candidate_party(self, party_slug: str) -> Response:
return self.post("{}/candidate/{}".format(self.url, party_slug))
def _get_main_citizen_hovercard(self, citizen: int) -> Response:
return self.get("{}/main/citizen-hovercard/{}".format(self.url, citizen))
def _get_main_citizen_profile_json(self, player_id: int) -> Response:
return self.get("{}/main/citizen-profile-json/{}".format(self.url, player_id))
def _get_main_citizen_notifications(self) -> Response:
return self.get("{}/main/citizenDailyAssistant".format(self.url))
def _get_main_citizen_daily_assistant(self) -> Response:
return self.get("{}/main/citizenNotifications".format(self.url))
def _get_main_city_data_residents(self, city: int, page: int = 1, params: Mapping[str, Any] = None) -> Response:
if params is None:
params = {}
return self.get("{}/main/city-data/{}/residents".format(self.url, city), params={"currentPage": page, **params})
def _get_country_military(self, country: str) -> Response:
return self.get("{}/country/military/{}".format(self.url, country))
def _get_economy_citizen_accounts(self, organisation_id: int) -> Response:
return self.get("{}/economy/citizen-accounts/{}".format(self.url, organisation_id))
def _get_economy_inventory_items(self) -> Response:
return self.get("{}/economy/inventory-items/".format(self.url))
def _get_economy_job_market_json(self, country: int) -> Response:
return self.get("{}/economy/job-market-json/{}/1/desc".format(self.url, country))
def _get_economy_my_companies(self) -> Response:
return self.get("{}/economy/myCompanies".format(self.url))
def _get_economy_my_market_offers(self) -> Response:
return self.get("{}/economy/myMarketOffers".format(self.url))
def _get_main_job_data(self) -> Response:
return self.get("{}/main/job-data".format(self.url))
def _get_main_leaderboards_damage_aircraft_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Response:
data = (country, weeks, mu)
return self.get("{}/main/leaderboards-damage-aircraft-rankings/{}/{}/{}/0".format(self.url, *data))
def _get_main_leaderboards_damage_rankings(self, country: int, weeks: int = 0, mu: int = 0,
div: int = 0) -> Response:
data = (country, weeks, mu, div)
return self.get("{}/main/leaderboards-damage-rankings/{}/{}/{}/{}".format(self.url, *data))
def _get_main_leaderboards_kills_aircraft_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Response:
data = (country, weeks, mu)
return self.get("{}/main/leaderboards-kills-aircraft-rankings/{}/{}/{}/0".format(self.url, *data))
def _get_main_leaderboards_kills_rankings(self, country: int, weeks: int = 0, mu: int = 0,
div: int = 0) -> Response:
data = (country, weeks, mu, div)
return self.get("{}/main/leaderboards-kills-rankings/{}/{}/{}/{}".format(self.url, *data))
def _get_main(self) -> Response:
return self.get(self.url)
def _get_main_messages_paginated(self, page: int = 1) -> Response:
return self.get("{}/main/messages-paginated/{}".format(self.url, page))
def _get_military_campaigns(self) -> Response:
return self.get("{}/military/campaigns-new/".format(self.url))
def _get_military_campaigns_json_list(self) -> Response:
return self.get("{}/military/campaignsJson/list".format(self.url))
def _get_military_unit_data(self, unit_id: int, **kwargs) -> Response:
params = {"groupId": unit_id, "panel": "members", **kwargs}
return self.get("{}/military/military-unit-data/".format(self.url), params=params)
def _get_main_money_donation_accept(self, donation_id: int) -> Response:
return self.get("{}/main/money-donation/accept/{}".format(self.url, donation_id), params={"_token": self.token})
def _get_main_money_donation_reject(self, donation_id: int) -> Response:
return self.get("{}/main/money-donation/reject/{}".format(self.url, donation_id), params={"_token": self.token})
def _get_main_notifications_ajax_community(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/community/{}".format(self.url, page))
def _get_main_notifications_ajax_system(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/system/{}".format(self.url, page))
def _get_main_notifications_ajax_report(self, page: int = 1) -> Response:
return self.get("{}/main/notificationsAjax/report/{}".format(self.url, page))
def _get_main_party_members(self, party: int) -> Response:
return self.get("{}/main/party-members/{}".format(self.url, party))
def _get_main_rankings_parties(self, country: int) -> Response:
return self.get("{}/main/rankings-parties/1/{}".format(self.url, country))
def _get_main_training_grounds_json(self) -> Response:
return self.get("{}/main/training-grounds-json".format(self.url))
def _get_main_weekly_challenge_data(self) -> Response:
return self.get("{}/main/weekly-challenge-data".format(self.url))
def _get_wars_show(self, war_id: int) -> Response:
return self.get("{}/wars/show/{}".format(self.url, war_id))
def _post_main_activate_battle_effect(self, battle: int, kind: str, citizen_id: int) -> Response:
data = dict(battleId=battle, citizenId=citizen_id, type=kind, _token=self.token)
return self.post("{}/main/fight-activateBattleEffect".format(self.url), data=data)
def _post_main_article_comments(self, article: int, page: int = 1) -> Response:
data = dict(_token=self.token, articleId=article, page=page)
if page:
data.update({'page': page})
return self.post("{}/main/articleComments".format(self.url), data=data)
def _post_main_article_comments_create(self, message: str, article: int, parent: int = 0) -> Response:
data = dict(_token=self.token, message=message, articleId=article)
if parent:
data.update({"parentId": parent})
return self.post("{}/main/articleComments/create".format(self.url), data=data)
def _post_main_battlefield_travel(self, side_id: int, battle_id: int) -> Response:
data = dict(_token=self.token, sideCountryId=side_id, battleId=battle_id)
return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
def _post_main_battlefield_change_division(self, battle_id: int, division_id: int) -> Response:
data = dict(_token=self.token, battleZoneId=division_id, battleId=battle_id)
return self.post("{}/main/battlefieldTravel".format(self.url), data=data)
def _post_main_buy_gold_items(self, currency: str, item: str, amount: int) -> Response:
data = dict(itemId=item, currency=currency, amount=amount, _token=self.token)
return self.post("{}/main/buyGoldItems".format(self.url), data=data)
def _post_candidate_for_congress(self, presentation: str = "") -> Response:
data = dict(_token=self.token, presentation=presentation)
return self.post("{}/candidate-for-congress".format(self.url), data=data)
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"})
else:
data.update({"action": "removeFriend"})
return self.post("{}/main/citizen-addRemoveFriend".format(self.url), data=data)
def _post_main_collect_anniversary_reward(self) -> Response:
return self.post("{}/main/collect-anniversary-reward".format(self.url), data={"_token": self.token})
def _post_main_country_donate(self, country: int, action: str, value: Union[int, float],
quality: int = None) -> Response:
json = dict(countryId=country, action=action, _token=self.token, value=value, quality=quality)
return self.post("{}/main/country-donate".format(self.url), data=json,
headers={"Referer": "{}/country/economy/Latvia".format(self.url)})
def _post_main_daily_task_reward(self) -> Response:
return self.post("{}/main/daily-tasks-reward".format(self.url), data=dict(_token=self.token))
def _post_main_donate_article(self, article_id: int, amount: int) -> Response:
data = dict(_token=self.token, articleId=article_id, amount=amount)
return self.post("{}/main/donate-article".format(self.url), data=data)
def _post_main_global_alerts_close(self, alert_id: int) -> Response:
data = dict(_token=self.token, alert_id=alert_id)
return self.post("{}/main/global-alerts/close".format(self.url), data=data)
def _post_delete_message(self, msg_id: list) -> Response:
data = {"_token": self.token, "delete_message[]": msg_id}
return self.post("{}/main/messages-delete".format(self.url), data)
def _post_eat(self, color: str) -> Response:
data = dict(_token=self.token, buttonColor=color)
return self.post("{}/main/eat".format(self.url), params=data)
def _post_economy_activate_booster(self, quality: int, duration: int, kind: str) -> Response:
data = dict(type=kind, quality=quality, duration=duration, fromInventory=True)
return self.post("{}/economy/activateBooster".format(self.url), data=data)
def _post_economy_activate_house(self, quality: int) -> Response:
data = {"action": "activate", "quality": quality, "type": "house", "_token": self.token}
return self.post("{}/economy/activateHouse".format(self.url), data=data)
def _post_economy_assign_to_holding(self, factory: int, holding: int) -> Response:
data = dict(_token=self.token, factoryId=factory, action="assign", holdingCompanyId=holding)
return self.post("{}/economy/assign-to-holding".format(self.url), data=data)
def _post_economy_create_company(self, industry: int, building_type: int = 1) -> Response:
data = {"_token": self.token, "company[industry_id]": industry, "company[building_type]": building_type}
return self.post("{}/economy/create-company".format(self.url), data=data,
headers={"Referer": "{}/economy/create-company".format(self.url)})
def _post_economy_donate_items_action(self, citizen: int, amount: int, industry: int,
quality: int) -> Response:
data = dict(citizen_id=citizen, amount=amount, industry_id=industry, quality=quality, _token=self.token)
return self.post("{}/economy/donate-items-action".format(self.url), data=data,
headers={"Referer": "{}/economy/donate-items/{}".format(self.url, citizen)})
def _post_economy_donate_money_action(self, citizen: int, amount: float = 0.0,
currency: int = 62) -> Response:
data = dict(citizen_id=citizen, _token=self.token, currency_id=currency, amount=amount)
return self.post("{}/economy/donate-money-action".format(self.url), data=data,
headers={"Referer": "{}/economy/donate-money/{}".format(self.url, citizen)})
def _post_economy_exchange_purchase(self, amount: float, currency: int, offer: int) -> Response:
data = dict(_token=self.token, amount=amount, currencyId=currency, offerId=offer)
return self.post("{}/economy/exchange/purchase/".format(self.url), data=data)
def _post_economy_exchange_retrieve(self, personal: bool, page: int, currency: int) -> Response:
data = dict(_token=self.token, personalOffers=int(personal), page=page, currencyId=currency)
return self.post("{}/economy/exchange/retrieve/".format(self.url), data=data)
def _post_economy_game_tokens_market(self, action: str) -> Response:
assert action in ['retrieve', ]
data = dict(_token=self.token, action=action)
return self.post("{}/economy/gameTokensMarketAjax".format(self.url), data=data)
def _post_economy_job_market_apply(self, citizen: int, salary: int) -> Response:
data = dict(_token=self.token, citizenId=citizen, salary=salary)
return self.post("{}/economy/job-market-apply".format(self.url), 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)
return self.post("{}/economy/marketplaceAjax".format(self.url), data=data)
def _post_economy_marketplace_actions(self, amount: int, buy: bool = False, **kwargs) -> Response:
if buy:
data = dict(_token=self.token, offerId=kwargs['offer'], amount=amount, orderBy="price_asc", currentPage=1,
buyAction=1)
else:
data = dict(_token=self.token, countryId=kwargs["country"], price=kwargs["price"],
industryId=kwargs["industry"], quality=kwargs["quality"], amount=amount, sellAction='postOffer')
return self.post("{}/economy/marketplaceActions".format(self.url), data=data)
def _post_economy_resign(self) -> Response:
return self.post("{}/economy/resign".format(self.url),
headers={"Content-Type": "application/x-www-form-urlencoded"},
data={"_token": self.token, "action_type": "resign"})
def _post_economy_sell_company(self, factory: 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"})
else:
data.update({"dissolve": factory})
return self.post("{}/economy/sell-company/{}".format(self.url, factory),
data=data, headers={"Referer": self.url})
def _post_economy_train(self, tg_ids: List[int]) -> Response:
data: Dict[str, Union[int, str]] = {}
if not tg_ids:
return self._get_main_training_grounds_json()
else:
for idx, tg_id in enumerate(tg_ids):
data["grounds[%i][id]" % idx] = tg_id
data["grounds[%i][train]" % idx] = 1
if data:
data['_token'] = self.token
return self.post("{}/economy/train".format(self.url), 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)
return self.post("{}/economy/upgrade-company".format(self.url), data=data)
def _post_economy_work(self, action_type: str, wam: List[int] = None, employ: Dict[int, int] = None) -> Response:
"""
:return: requests.Response or None
"""
if employ is None:
employ = dict()
if wam is None:
wam = []
data: Dict[str, Union[int, str]] = dict(action_type=action_type, _token=self.token)
if action_type == "production":
max_idx = 0
for company_id in sorted(wam or []):
data.update({
"companies[%i][id]" % max_idx: company_id,
"companies[%i][employee_works]" % max_idx: employ.pop(company_id, 0),
"companies[%i][own_work]" % max_idx: 1
})
max_idx += 1
for company_id in sorted(employ or []):
data.update({
"companies[%i][id]" % max_idx: company_id,
"companies[%i][employee_works]" % max_idx: employ.pop(company_id),
"companies[%i][own_work]" % max_idx: 0
})
max_idx += 1
return self.post("{}/economy/work".format(self.url), data=data)
def _post_economy_work_overtime(self) -> Response:
data = dict(action_type="workOvertime", _token=self.token)
return self.post("{}/economy/workOvertime".format(self.url), data=data)
def _post_forgot_password(self, email: str) -> Response:
data = dict(_token=self.token, email=email, commit="Reset password")
return self.post("{}/forgot-password".format(self.url), data=data)
def _post_military_fight_activate_booster(self, battle: int, quality: int, duration: int, kind: str) -> Response:
data = dict(type=kind, quality=quality, duration=duration, battleId=battle, _token=self.token)
return self.post("{}/military/fight-activateBooster".format(self.url), data=data)
def _post_military_change_weapon(self, battle: int, battle_zone: int, weapon_level: int,) -> Response:
data = dict(battleId=battle, _token=self.token, battleZoneId=battle_zone, customizationLevel=weapon_level)
return self.post("{}/military/change-weapon".format(self.url), 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')
return self.post("{}/login".format(self.url), 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"}
return self.post("{}/main/messages-alerts/1".format(self.url), 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)
return self.post("{}/main/messages-compose/{}".format(self.url, url_pk), data=data)
def _post_military_battle_console(self, battle_id: int, action: str, page: int = 1, **kwargs) -> Response:
data = dict(battleId=battle_id, action=action, _token=self.token)
if action == "battleStatistics":
data.update(round=kwargs["round_id"], zoneId=kwargs["round_id"], leftPage=page, rightPage=page,
division=kwargs["division"], type=kwargs.get("type", 'damage'), )
elif action == "warList":
data.update(page=page)
return self.post("{}/military/battle-console".format(self.url), data=data)
def _post_military_deploy_bomb(self, battle_id: int, bomb_id: int) -> Response:
data = dict(battleId=battle_id, bombId=bomb_id, _token=self.token)
return self.post("{}/military/deploy-bomb".format(self.url), data=data)
def _post_military_fight_air(self, battle_id: int, side_id: int, zone_id: int) -> Response:
data = dict(sideId=side_id, battleId=battle_id, _token=self.token, battleZoneId=zone_id)
return self.post("{}/military/fight-shoooot/{}".format(self.url, battle_id), data=data)
def _post_military_fight_ground(self, battle_id: int, side_id: int, zone_id: int) -> Response:
data = dict(sideId=side_id, battleId=battle_id, _token=self.token, battleZoneId=zone_id)
return self.post("{}/military/fight-shooot/{}".format(self.url, battle_id), data=data)
def _post_military_group_missions(self) -> Response:
data = dict(action="check", _token=self.token)
return self.post("{}/military/group-missions".format(self.url), data=data)
def _post_main_travel(self, check: str, **kwargs) -> Response:
data = dict(_token=self.token, check=check, **kwargs)
return self.post("{}/main/travel".format(self.url), data=data)
def _post_main_vote_article(self, article_id: int) -> Response:
data = dict(_token=self.token, articleId=article_id)
return self.post("{}/main/vote-article".format(self.url), data=data)
def _post_main_travel_data(self, **kwargs) -> Response:
return self.post("{}/main/travelData".format(self.url), data=dict(_token=self.token, **kwargs))
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('{}/wars/attack-region/{}/{}'.format(self.url, war_id, region_id), data=data)
def _post_main_weekly_challenge_reward(self, reward_id: int) -> Response:
data = dict(_token=self.token, rewardId=reward_id)
return self.post("{}/main/weekly-challenge-collect-reward".format(self.url), data=data)
def _post_main_write_article(self, title: str, content: str, location: int, kind: int) -> Response:
data = dict(_token=self.token, article_name=title, article_body=content, article_location=location,
article_category=kind)
return self.post("{}/main/write-article".format(self.url), data=data)
# Wall Posts
# ## Country
def _post_main_country_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/country-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/country-comment/create/json".format(self.url), 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}
return self.post("{}/main/country-post/create/json".format(self.url), data=data)
def _post_main_country_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/country-post/retrieve/json".format(self.url), data=data)
# ## Military Unit
def _post_main_military_unit_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/military-unit-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/military-unit-comment/create/json".format(self.url), 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}
return self.post("{}/main/military-unit-post/create/json".format(self.url), data=data)
def _post_main_military_unit_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/military-unit-post/retrieve/json".format(self.url), data=data)
# ## Party
def _post_main_party_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/party-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/party-comment/create/json".format(self.url), data=data)
def _post_main_party_post_create(self, body: str) -> Response:
data = {"_token": self.token, "post_message": body}
return self.post("{}/main/party-post/create/json".format(self.url), data=data)
def _post_main_party_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/party-post/retrieve/json".format(self.url), data=data)
# ## Friend's Wall
def _post_main_wall_comment_retrieve(self, post_id: int) -> Response:
data = {"_token": self.token, "postId": post_id}
return self.post("{}/main/wall-comment/retrieve/json".format(self.url), 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}
return self.post("{}/main/wall-comment/create/json".format(self.url), data=data)
def _post_main_wall_post_create(self, body: str) -> Response:
data = {"_token": self.token, "post_message": body}
return self.post("{}/main/wall-post/create/json".format(self.url), data=data)
def _post_main_wall_post_automatic(self, **kwargs) -> Response:
kwargs.update(_token=self.token)
return self.post("{}/main/wall-post/create/json".format(self.url), data=kwargs)
def _post_main_wall_post_retrieve(self) -> Response:
data = {"_token": self.token, "page": 1, "switchedFrom": False}
return self.post("{}/main/wall-post/retrieve/json".format(self.url), data=data)
# 12th anniversary endpoints
def _get_anniversary_quest_data(self) -> Response:
return self.get("{}/main/anniversaryQuestData".format(self.url))
def _post_map_rewards_unlock(self, node_id: int) -> Response:
data = {'nodeId': node_id, '_token': self.token}
return self.post("{}/main/map-rewards-unlock".format(self.url), 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}
return self.post("{}/main/map-rewards-speedup".format(self.url), data=data)
def _post_map_rewards_claim(self, node_id: int) -> Response:
data = {'nodeId': node_id, '_token': self.token}
return self.post("{}/main/map-rewards-claim".format(self.url), 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=utils.COUNTRY_LINK[attack_country_id])
return self.post("{}/{}/new-war".format(self.url, utils.COUNTRY_LINK[self_country_id]), 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)
return self.post("{}/{}/new-donation".format(self.url, utils.COUNTRY_LINK[country_id]), data=data)
class Reporter:
__to_update: List[Dict[Any, Any]] = None
name: str = ""