@ -2,12 +2,13 @@ import datetime
import decimal
import hashlib
import random
import threading
import time
from collections import deque
from json import JSONDecodeError , loads , JSONEncoder
from typing import Any , Dict , List , Union , Mapping , Iterable
from typing import Any , Dict , List , Union , Mapping , Iterable , Tuple
from requests import Response , Session
from requests import Response , Session , post
from erepublik import utils
@ -17,7 +18,7 @@ class ErepublikException(Exception):
super ( ) . __init__ ( message )
class ErepublikNetworkException ( Exception) :
class ErepublikNetworkException ( Erepublik Exception) :
def __init__ ( self , message , request ) :
super ( ) . __init__ ( message )
self . request = request
@ -61,7 +62,7 @@ class MyCompanies:
production = 0 , base_production = 0 , wam_enabled = False , can_work_as_manager = False ,
preset_own_work = 0 , already_worked = False , can_assign_employees = False , preset_works = 0 ,
todays_works = 0 , holding_company_id = None , is_assigned_to_holding = False ,
cannot_work_as_manager_reason = False )
cannot_work_as_manager_reason = False , industry_id = 0 )
for c_id , company in companies . items ( ) :
tmp = { }
@ -119,7 +120,10 @@ class MyCompanies:
raw = [ ]
factory = [ ]
if holding_id in self . holdings :
for company_id in self . holdings . get ( holding_id , { } ) . get ( ' companies ' , [ ] ) :
for company_id in sorted ( self . holdings . get ( holding_id , { } ) . get ( ' companies ' , [ ] ) ,
key = lambda cid : ( - self . companies [ cid ] . get ( ' is_raw ' ) , # True, False
self . companies [ cid ] . get ( ' industry_id ' ) , # F W H A
- self . companies [ cid ] . get ( ' quality ' ) , ) ) : # 7, 6, .. 2, 1
company = self . companies . get ( company_id , { } )
wam_enabled = bool ( company . get ( ' wam_enabled ' , { } ) )
already_worked = not company . get ( ' already_worked ' , { } )
@ -156,28 +160,33 @@ class MyCompanies:
raise ErepublikException ( " Wrong function call " )
# @property
# def __dict__(self):
# ret = {}
# for key in dir(self):
# if not key.startswith('_'):
# ret[key] = getattr(self, key)
# 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/73 .0.3683.103 Safari/537.36 ' ,
' Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
' Chrome/72 .0.3626.13 Safari/537.36 ' ,
' Mozilla/5.0 (Windows NT 10.0 ; W in64; x 64) AppleWebKit/537.36 (KHTML, like Gecko) '
' Chrome/71 .0.3578.98 Safari/537.36 ' ,
' Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36 ' ,
' Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.11 Safari/537.36 ' ,
' Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 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 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.80 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 ; L inux x86_ 64) 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/75 .0.3770.80 Safari/537.36' ,
# FireFox
' Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:66 .0) Gecko/20100101 Firefox/66 .0 ' ,
' Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:65 .0) Gecko/20100101 Firefox/65 .0 ' ,
' Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:64 .0) Gecko/20100101 Firefox/64 .0 ' ,
' Mozilla/5.0 (X11; Linux x86_64; rv:66 .0) Gecko/20100101 Firefox/66 .0 ' ,
' Mozilla/5.0 (X11; Linux x86_64; rv:65 .0) Gecko/20100101 Firefox/65 .0 ' ,
' Mozilla/5.0 (X11; Linux x86_64; rv:64 .0) Gecko/20100101 Firefox/64 .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 (Windows NT 10.0; Win64; x64; rv:67 .0) Gecko/20100101 Firefox/67 .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 ' ,
' Mozilla/5.0 (X11; Linux x86_64; rv:67 .0) Gecko/20100101 Firefox/67 .0 ' ,
]
debug = False
@ -192,7 +201,7 @@ class SlowRequests(Session):
@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 )
request_log_name = self . request_log_name , debug = self . debug )
def request ( self , method , url , * args , * * kwargs ) :
self . _slow_down_requests ( )
@ -211,7 +220,10 @@ class SlowRequests(Session):
def _log_request ( self , url , method , data = None , json = None , params = None , * * kwargs ) :
if self . debug :
args = { }
args . update ( { ' kwarg s' : kwargs } )
kw args. pop ( ' allow_redirect s' , None )
if kwargs :
args . update ( { ' kwargs ' : kwargs } )
if data :
args . update ( { " data " : data } )
@ -235,10 +247,9 @@ class SlowRequests(Session):
self . _log_request ( hist_resp . request . url , " REDIRECT " )
self . _log_response ( hist_resp . request . url , hist_resp , redirect = True )
# TODO: Must thoroughly check response writing on windows systems
file_data = {
" path " : ' debug/requests ' ,
" time " : self . last_time . strftime ( ' % Y- % m- %d _ % H- % M- % S ' ) ,
" time " : self . last_time . strftime ( ' % Y/ % m/ %d / % H- % M- % S ' ) ,
" name " : utils . slugify ( url [ len ( Citizen . url ) : ] ) ,
" extra " : " _REDIRECT " if redirect else " "
}
@ -260,7 +271,7 @@ class Config:
work = True
train = True
wam = False
auto_sell : List [ str ] = list ( )
auto_sell : List [ str ] = None
auto_sell_all = False
employees = False
fight = False
@ -280,38 +291,44 @@ class Config:
force_wam = False
sort_battles_time = True
force_travel = False
telegram = True
telegram_chat_id = 0
telegram_token = " "
def __init__ ( self ) :
self . auto_sell = [ ]
@property
def wt ( self ) :
return self . work and self . train
def __dict__ ( self ) - > Dict [ str , Union [ bool , str , List [ str ] ] ] :
return dict (
email = self . email ,
password = self . password ,
work = self . work ,
train = self . train ,
wam = self . wam ,
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 ,
boosters = self . boosters ,
travel_to_fight = self . travel_to_fight ,
epic_hunt = self . epic_hunt ,
epic_hunt_ebs = self . epic_hunt_ebs ,
rw_def_side = self . rw_def_side ,
interactive = self . interactive ,
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 ,
)
def reset ( self ) :
self . work = True
self . train = True
self . wam = False
self . auto_sell = list ( )
self . auto_sell_all = False
self . employees = False
self . fight = False
self . air = False
self . ground = False
self . all_in = False
self . next_energy = False
self . boosters = False
self . travel_to_fight = False
self . always_travel = False
self . epic_hunt = False
self . epic_hunt_ebs = False
self . rw_def_side = False
self . interactive = True
self . continuous_fighting = False
self . auto_buy_raw = False
self . force_wam = False
self . sort_battles_time = True
self . force_travel = False
self . telegram = True
self . telegram_chat_id = 0
self . telegram_token = " "
class Energy :
@ -332,7 +349,7 @@ class Energy:
@property
def food_fights ( self ) :
return ( self . recoverable + self . recovered ) / / 10
return self . available / / 10
@property
def reference_time ( self ) :
@ -344,7 +361,7 @@ class Energy:
@property
def is_recoverable_full ( self ) :
return self . recoverable > = self . limit - self . interval
return self . recoverable > = self . limit - 5 * self . interval
@property
def is_recovered_full ( self ) :
@ -358,24 +375,14 @@ class Energy:
def available ( self ) :
return self . recovered + self . recoverable
@property
def __dict__ ( self ) :
return dict (
limit = self . limit ,
interval = self . interval ,
recoverable = self . recoverable ,
recovered = self . recovered ,
reference_time = self . reference_time
)
class Details ( object ) :
class Details :
xp = 0
cc = 0
pp = 0
pin = None
gold = 0
next_pp : List [ int ] = [ ]
next_pp : List [ int ] = None
citizen_id = 0
citizenship = 0
current_region = 0
@ -386,6 +393,9 @@ class Details(object):
daily_task_reward = False
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 , }
def __init__ ( self ) :
self . next_pp = [ ]
@property
def xp_till_level_up ( self ) :
if self . xp > = 10000 :
@ -423,7 +433,7 @@ class Politics:
is_country_president : bool = False
class House ( object ) :
class House :
quality = None
unactivated_count = 0
active_untill = utils . good_timedelta ( utils . now ( ) , - datetime . timedelta ( days = 1 ) )
@ -448,31 +458,34 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
"""
self . _req = SlowRequests ( )
def post ( self , url : str , * args , * * kwargs ) - > Response :
return self . _req . post ( url , * args , * * kwargs )
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_article_json ( self , article_id : int ) - > Response :
def _get_main_ article_json ( self , article_id : int ) - > Response :
return self . get ( " {} /main/articleJson/ {} " . format ( self . url , article_id ) )
def _get_battlefield_choose_side ( self , battle : int , side : int ) - > Response :
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_citizen_hovercard ( self , citizen : int ) - > Response :
def _get_main_ citizen_hovercard ( self , citizen : int ) - > Response :
return self . get ( " {} /main/citizen-hovercard/ {} " . format ( self . url , citizen ) )
def _get_citizen_profile ( self , player_id : int ) - > Response :
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_citizen_daily_assistant ( self ) - > Response :
def _get_main_ citizen_daily_assistant ( self ) - > Response :
return self . get ( " {} /main/citizenDailyAssistant " . format ( self . url ) )
def _get_city_data_residents ( self , city : int , page : int = 1 , params : Mapping [ str , Any ] = None ) - > Response :
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 } )
@ -495,96 +508,98 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
def _get_economy_my_market_offers ( self ) - > Response :
return self . get ( " {} /economy/myMarketOffers " . format ( self . url ) )
def _get_job_data ( self ) - > Response :
def _get_main_ job_data ( self ) - > Response :
return self . get ( " {} /main/job-data " . format ( self . url ) )
def _get_leaderboards_damage_aircraft_rankings ( self , country : int , weeks : int = 0 , mu : int = 0 ) - > Response :
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_leaderboards_damage_rankings ( self , country : int , weeks : int = 0 , mu : int = 0 , div : int = 0 ) - > Response :
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_leaderboards_kills_aircraft_rankings ( self , country : int , weeks : int = 0 , mu : int = 0 ) - > Response :
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_leaderboards_kills_rankings ( self , country : int , weeks : int = 0 , mu : int = 0 , div : int = 0 ) - > Response :
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_messages ( self , page : int = 1 ) - > Response :
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_show_weapons ( self , battle_id : int ) - > Response :
params = { " _token " : self . token , " battleId " : battle_id }
return self . get ( " {} /military/show-weapons " . format ( self . url ) , params = params )
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_money_donation_accept ( self , donation_id : int ) - > Response :
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_money_donation_reject ( self , donation_id : int ) - > Response :
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_notifications_ajax_community ( self , page : int = 1 ) - > Response :
def _get_main_ notifications_ajax_community ( self , page : int = 1 ) - > Response :
return self . get ( " {} /main/notificationsAjax/community/ {} " . format ( self . url , page ) )
def _get_notifications_ajax_system ( self , page : int = 1 ) - > Response :
def _get_main_ notifications_ajax_system ( self , page : int = 1 ) - > Response :
return self . get ( " {} /main/notificationsAjax/system/ {} " . format ( self . url , page ) )
def _get_notifications_ajax_report ( self , page : int = 1 ) - > Response :
def _get_main_ notifications_ajax_report ( self , page : int = 1 ) - > Response :
return self . get ( " {} /main/notificationsAjax/report/ {} " . format ( self . url , page ) )
def _get_party_members ( self , party : int ) - > Response :
def _get_main_ party_members ( self , party : int ) - > Response :
return self . get ( " {} /main/party-members/ {} " . format ( self . url , party ) )
def _get_rankings_parties ( self , country : int ) - > Response :
def _get_main_ rankings_parties ( self , country : int ) - > Response :
return self . get ( " {} /main/rankings-parties/1/ {} " . format ( self . url , country ) )
def _get_training_grounds_json ( self ) - > Response :
def _get_main_ training_grounds_json ( self ) - > Response :
return self . get ( " {} /main/training-grounds-json " . format ( self . url ) )
def _get_weekly_challenge_data ( self ) - > Response :
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_activate_battle_effect ( self , battle : int , kind : str , citizen_id : int ) - > Response :
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_article_comments ( self , article : int , page : int = 1 ) - > Response :
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_article_comments_create ( self , message : str , article : int , parent : int = 0 ) - > Response :
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_battle_console ( self , battle : int , zone : int , round _id: int , division : int , page : int ,
damage : bool ) - > Response :
data = dict ( battleId = battle , zoneId = zone , action = " battleStatistics " , round = round_id , division = division ,
leftPage = page , rightPage = page , _token = self . token )
if damage :
data . update ( { " type " : " damage " } )
else :
data . update ( { " type " : " kills " } )
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 )
return self . post ( " {} /military/battle-console " . format ( self . url ) , data = data )
def _post_buy_gold_items ( self , currency : str , item : str , amount : int ) - > Response :
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 )
@ -592,7 +607,7 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
data = dict ( _token = self . token , presentation = presentation )
return self . post ( " {} /candidate-for-congress " . format ( self . url ) , data = data )
def _post_citizen_add_remove_friend ( self , citizen : int , add : bool ) - > Response :
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 " } )
@ -600,18 +615,22 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
data . update ( { " action " : " removeFriend " } )
return self . post ( " {} /main/citizen-addRemoveFriend " . format ( self . url ) , data = data )
def _post_collect_anniversary_reward ( self ) - > Response :
def _post_main_ collect_anniversary_reward ( self ) - > Response :
return self . post ( " {} /main/collect-anniversary-reward " . format ( self . url ) , data = { " _token " : self . token } )
def _post_country_donate ( self , country : int , action : str , value : Union [ int , float ] ,
quality : int = None ) - > Response :
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_daily_task_reward ( self ) - > Response :
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_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 )
@ -687,18 +706,18 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
data = { " _token " : self . token , " action_type " : " resign " } )
def _post_economy_sell_company ( self , factory : int , pin : int = None , sell : bool = True ) - > Response :
url = " {} /economy/sell-company/ {} " . format ( self . url , factory )
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 ( url , data = d ata , headers = { " Referer " : url } )
return self . post ( " {} /economy/sell-company/ {} " . form at( 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_training_grounds_json ( )
return self . _get_main_ training_grounds_json ( )
else :
for idx , tg_id in enumerate ( tg_ids ) :
data [ " grounds[ %i ][id] " % idx ] = tg_id
@ -746,19 +765,23 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
data = dict ( _token = self . token , email = email , commit = " Reset password " )
return self . post ( " {} /forgot-password " . format ( self . url ) , data = data )
def _post_fight_activate_booster ( self , battle : int , quality : int , duration : int , kind : str ) - > Response :
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_messages_alert ( self , notification_ids : List [ int ] ) - > Response :
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_messages_compose ( self , subject : str , body : str , citizens : List [ int ] ) - > Response :
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 )
@ -768,43 +791,52 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
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 ' ) , )
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_change_weapon ( self , battle_id : int , battle_zone_id : int , customization_level : int ) - > Response :
data = dict ( _token = self . token , battleZoneId = battle_zone_id , battleId = battle_id ,
customizationLevel = customization_level )
return self . post ( " {} /military/change-weapon " . 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 ) - > Response :
data = dict ( sideId = side_id , battleId = battle_id , _token = self . token )
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 ) - > Response :
data = dict ( sideId = side_id , battleId = battle_id , _token = self . token )
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_travel ( self , check : str , * * kwargs ) - > Response :
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_travel_data ( self , * * kwargs ) - > Response :
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_weekly_challenge_reward ( self , reward_id : int ) - > Response :
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_write_article ( self , title : str , content : str , location : int , kind : int ) - > Response :
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 )
@ -812,95 +844,127 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
# Wall Posts
# ## Country
def _post_country_comment_retrieve ( self , post_id : int ) - > Response :
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_country_comment_create ( self , post_id : int , comment_message : str ) - > Response :
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_country_post_create ( self , body : str , post_as : int ) - > Response :
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_country_post_retrieve ( self ) - > Response :
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_military_unit_comment_retrieve ( self , post_id : int ) - > Response :
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_military_unit_comment_create ( self , post_id : int , comment_message : str ) - > Response :
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_military_unit_post_create ( self , body : str , post_as : int ) - > Response :
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_military_unit_post_retrieve ( self ) - > Response :
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_party_comment_retrieve ( self , post_id : int ) - > Response :
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_party_comment_create ( self , post_id : int , comment_message : str ) - > Response :
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_party_post_create ( self , body : str ) - > Response :
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_party_post_retrieve ( self ) - > Response :
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_wall_comment_retrieve ( self , post_id : int ) - > Response :
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_wall_comment_create ( self , post_id : int , comment_message : str ) - > Response :
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_wall_post_create ( self , body : str ) - > Response :
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_wall_post_retrieve ( self ) - > Response :
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 ] ] = [ ]
__to_update : List [ Dict [ Any , Any ] ] = None
name : str = " "
email : str = " "
citizen_id : int = 0
key : str = " "
allowed : bool = False
@property
def __dict__ ( self ) :
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 ) :
self . _req = Session ( )
self . url = " https://api.erep.lv "
self . _req . headers . update ( { " user-agent " : " Bot reporter v2 " } )
self . __to_update = [ ]
self . __registered : bool = False
@property
def __dict__ ( self ) :
return dict ( allowed = self . allowed , __to_update = self . __to_update )
def do_init ( self , name : str = " " , email : str = " " , citizen_id : int = 0 ) :
self . name : str = name
self . email : str = email
@ -962,11 +1026,12 @@ class Reporter:
class MyJSONEncoder ( JSONEncoder ) :
def default ( self , o ) :
from erepublik . citizen import Citizen
if isinstance ( o , decimal . Decimal ) :
return float ( " {:.02f} " . format ( o ) )
elif isinstance ( o , datetime . datetime ) :
return dict ( __type__ = ' datetime ' , year = o . year , month = o . month , day = o . day , hour = o . hour , minute = o . minute ,
second = o . second , microsecond = o . microsecond )
second = o . second , microsecond = o . microsecond , tzinfo = o . tzinfo . zone if o . tzinfo else None )
elif isinstance ( o , datetime . date ) :
return dict ( __type__ = ' date ' , year = o . year , month = o . month , day = o . day )
elif isinstance ( o , datetime . timedelta ) :
@ -976,8 +1041,10 @@ class MyJSONEncoder(JSONEncoder):
return dict ( headers = o . headers . __dict__ , url = o . url , text = o . text )
elif hasattr ( o , ' __dict__ ' ) :
return o . __dict__
elif isinstance ( o , deque ) :
elif isinstance ( o , ( deque , set ) ):
return list ( o )
elif isinstance ( o , Citizen ) :
return o . to_json ( )
return super ( ) . default ( o )
@ -997,30 +1064,39 @@ class BattleSide:
class BattleDivision :
end : datetime . datetime
epic : bool
dom_pts : Dict [ str , int ] = None
wall : Dict [ str , Union [ int , float ] ] = None
dom_pts : Dict [ str , int ]
wall : Dict [ str , Union [ int , float ] ]
battle_zone_id : int
def_medal : Dict [ str , int ]
inv_medal : Dict [ str , int ]
@property
def div_end ( self ) - > bool :
return utils . now ( ) > = self . end
def __init__ ( self , end : datetime . datetime , epic : bool , inv_pts : int , def_pts : int , wall_for : int , wall_dom : float ) :
def __init__ (
self , div_id : int , end : datetime . datetime , epic : bool , inv_pts : int , def_pts : int ,
wall_for : int , wall_dom : float , def_medal : Tuple [ int , int ] , inv_medal : Tuple [ int , int ]
) :
self . battle_zone_id = div_id
self . end = end
self . epic = epic
self . dom_pts = dict ( { " inv " : inv_pts , " def " : def_pts } )
self . wall = dict ( { " for " : wall_for , " dom " : wall_dom } )
self . def_medal = { " id " : def_medal [ 0 ] , " dmg " : def_medal [ 1 ] }
self . inv_medal = { " id " : inv_medal [ 0 ] , " dmg " : inv_medal [ 1 ] }
class Battle ( object ) :
id : int = 0
war_id : int = 0
zone_id : int = 0
is_rw : bool = False
is_dict_lib : bool = False
start : datetime . datetime = None
invader : BattleSide = None
defender : BattleSide = None
div : Dict [ int , BattleDivision ] = None
class Battle :
id : int
war_id : int
zone_id : int
is_rw : bool
is_dict_lib : bool
start : datetime . datetime
invader : BattleSide
defender : BattleSide
div : Dict [ int , BattleDivision ]
@property
def is_air ( self ) - > bool :
@ -1051,11 +1127,18 @@ class Battle(object):
else :
end = utils . localize_dt ( datetime . datetime . max - datetime . timedelta ( days = 1 ) )
battle_div = B attleDivision (
end = end , epic = data . get ( ' epic_type ' ) in [ 1 , 5 ] ,
inv_pts = data . get ( ' dom_pts ' ) . get ( " inv " ) , def_pts = data . get ( ' dom_pts ' ) . get ( " def " ) ,
wall_for = data . get ( ' wall ' ) . get ( " for " ) , wall_dom = data . get ( ' wall ' ) . get ( " dom " )
)
if not d ata [ ' stats ' ] [ ' def ' ] :
def_medal = ( 0 , 0 )
else :
def_medal = ( data [ ' stats ' ] [ ' def ' ] [ ' citizenId ' ] , data [ ' stats ' ] [ ' def ' ] [ ' damage ' ] )
if not data [ ' stats ' ] [ ' inv ' ] :
inv_medal = ( 0 , 0 )
else :
inv_medal = ( data [ ' stats ' ] [ ' inv ' ] [ ' citizenId ' ] , data [ ' stats ' ] [ ' inv ' ] [ ' damage ' ] )
battle_div = BattleDivision ( end = end , epic = data . get ( ' epic_type ' ) in [ 1 , 5 ] , div_id = data . get ( ' id ' ) ,
inv_pts = data . get ( ' dom_pts ' ) . get ( " inv " ) , def_pts = data . get ( ' dom_pts ' ) . get ( " def " ) ,
wall_for = data . get ( ' wall ' ) . get ( " for " ) , wall_dom = data . get ( ' wall ' ) . get ( " dom " ) ,
def_medal = def_medal , inv_medal = inv_medal )
self . div . update ( { div : battle_div } )
@ -1066,9 +1149,10 @@ class Battle(object):
time_part = " {} " . format ( now - self . start )
else :
time_part = " - {} " . format ( self . start - now )
return " Battle {} | {:>21.21} : {:<21.21} | Round {:2} | Start {} " . format (
self . id , utils . COUNTRIES [ self . invader . id ] , utils . COUNTRIES [ self . defender . id ] , self . zone_id , time_part
)
return f " Battle {self . id } | " \
f " { utils . COUNTRIES [ self . invader . id ] : >21.21 } : { utils . COUNTRIES [ self . defender . id ] : <21.21 } | " \
f " Round { self . zone_id : 2 } | " \
f " Time since start { time_part } "
class EnergyToFight :
@ -1100,3 +1184,91 @@ class EnergyToFight:
if 0 < new_energy < self . energy :
self . energy = new_energy
return self . energy
class TelegramBot :
__initialized = False
__queue : List [ str ]
chat_id = 0
api_url = " "
player_name = " "
__thread_stopper : threading . Event
_last_time : datetime . datetime
_last_full_energy_report : datetime . datetime
_next_time : datetime . datetime
_threads : List [ threading . Thread ]
def __init__ ( self , stop_event : threading . Event = None ) :
self . _threads = [ ]
self . __queue = [ ]
self . __thread_stopper = threading . Event ( ) if stop_event is None else stop_event
def __dict__ ( self ) :
return dict ( 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 = bool ( len ( self . _threads ) ) )
def do_init ( self , chat_id : int , token : str , player_name : str = " " ) :
self . chat_id = chat_id
self . api_url = " https://api.telegram.org/bot {} /sendMessage " . format ( token )
self . player_name = player_name
self . __initialized = True
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 ( " \n \n – – – – – – – – – – – – – – – – – – – – – – \n \n " . join ( self . __queue ) )
def send_message ( self , message : str ) - > bool :
self . __queue . append ( message )
if not self . __initialized :
return True
self . _threads = [ t for t in self . _threads if t . is_alive ( ) ]
self . _next_time = utils . good_timedelta ( utils . now ( ) , datetime . timedelta ( minutes = 1 ) )
if not self . _threads :
name = " telegram_ {} send " . format ( f " { self . player_name } _ " if self . player_name else " " )
send_thread = threading . Thread ( target = self . __send_messages , name = name )
send_thread . start ( )
self . _threads . append ( send_thread )
return True
def report_free_bhs ( self , battles : List [ Tuple [ int , int , int , int , datetime . timedelta ] ] ) :
battle_links = [ ]
for battle_id , side_id , against_id , damage , time_left in battles :
total_seconds = int ( time_left . total_seconds ( ) )
time_start = " "
hours , remainder = divmod ( total_seconds , 3600 )
if hours :
time_start = f " { hours } h "
minutes , seconds = divmod ( remainder , 60 )
time_start + = f " { minutes : 02 } m { seconds : 02 } s "
damage = " {:,} " . format ( damage ) . replace ( ' , ' , ' ' )
battle_links . append ( f " * { damage } *dmg bh for [ { utils . COUNTRIES [ side_id ] } vs { utils . COUNTRIES [ against_id ] } ] "
f " (https://www.erepublik.com/en/military/battlefield/ { battle_id } ) "
f " _time since start { time_start } _ " )
self . send_message ( " Free BHs: \n " + " \n " . join ( battle_links ) )
def report_full_energy ( self , available : int , limit : int , interval : int ) :
if ( utils . now ( ) - self . _last_full_energy_report ) . total_seconds ( ) > = 30 * 60 :
self . _last_full_energy_report = utils . now ( )
message = f " Full energy ( { available } hp/ { limit } hp + { interval } hp/6min) "
self . send_message ( message )
def report_medal ( self , msg ) :
self . send_message ( f " New award: * { msg } * " )
def __send_messages ( self ) :
while self . _next_time > utils . now ( ) :
if self . __thread_stopper . is_set ( ) :
break
self . __thread_stopper . wait ( utils . get_sleep_seconds ( self . _next_time ) )
message = " \n \n – – – – – – – – – – – – – – – – – – – – – – \n \n " . join ( self . __queue )
if self . player_name :
message = f " Player * { self . player_name } * \n " + message
response = post ( self . api_url , json = dict ( chat_id = self . chat_id , text = message , parse_mode = " Markdown " ) )
self . _last_time = utils . now ( )
if response . json ( ) . get ( ' ok ' ) :
self . __queue = [ ]
return True
return False