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

Telegram integration

Browse Source
This commit is contained in:
Eriks Karls 2019-09-26 15:42:16 +03:00
parent 8db4ab1f0f
commit 20bba4b9f9
2 changed files with 67 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
erepublik/citizen.py
View File

@ -11,6 +11,7 @@ import requests
from requests import Response, RequestException
from erepublik import classes, utils
from erepublik.classes import TelegramBot
class Citizen(classes.CitizenAPI):
@ -49,8 +50,9 @@ class Citizen(classes.CitizenAPI):
debug = False
__registered = False
logged_in = False
telegram = None
def __init__(self, email: str = "", password: str = "", auto_login: bool = True):
def __init__(self, email: str = "", password: str = "", auto_login: bool = True, telegram: Dict[str, Any] = None):
super().__init__()
self.commit_id = utils.COMMIT_ID
self.config = classes.Config()
@ -66,6 +68,11 @@ class Citizen(classes.CitizenAPI):
if auto_login:
self.login()
if telegram is None:
self.telegram = TelegramBot(620981703, "864251270:AAFzZZdjspI-kIgJVk4gF3TViGFoHnf8H4o", self.name)
else:
self.telegram = TelegramBot(telegram['chat_id'], telegram['token'])
def login(self):
self.get_csrf_token()
self.update_citizen_info()
@ -261,7 +268,9 @@ class Citizen(classes.CitizenAPI):
msgs = ["{count} x {kind}, totaling {} {currency}\n"
"{about}".format(d["count"] * d["reward"], **d) for d in data.values()]
self.write_log("Found awards: {}".format("\n".join(msgs)))
msgs = "\n".join(msgs)
self.telegram.report_medal(msgs)
self.write_log("Found awards: {}".format(msgs))
for info in data.values():
self.reporter.report_action("NEW_MEDAL", info)
@ -270,6 +279,7 @@ class Citizen(classes.CitizenAPI):
level = levelup.group(1)
msg = "Level up! Current level {}".format(level)
self.write_log(msg)
self.telegram.report_medal(f"Level *{level}*")
self.reporter.report_action("LEVEL_UP", value=level)
def update_all(self, force_update=False):
@ -337,6 +347,8 @@ class Citizen(classes.CitizenAPI):
self.energy.limit = citizen.get("energyToRecover", 0)
self.energy.recovered = citizen.get("energy", 0)
self.energy.recoverable = citizen.get("energyFromFoodRemaining", 0)
if self.energy.is_energy_full:
self.telegram.report_full_energy(self.energy.available, self.energy.limit, self.energy.interval)
self.details.current_region = citizen.get("regionLocationId", 0)
self.details.current_country = citizen.get("countryLocationId", 0) # country where citizen is located
@ -1740,6 +1752,7 @@ class Citizen(classes.CitizenAPI):
def launch_attack(self, war_id: int, region_id: int, region_name: str):
self._post_wars_attack_region(war_id, region_id, region_name)
self.telegram.send_message(f"Battle for *{region_name}* queued")
def state_update_repeater(self):
try:
@ -1848,6 +1861,7 @@ class Citizen(classes.CitizenAPI):
if self.details.cc < amount or amount < 20:
return False
data = dict(country=71, action='currency', value=amount)
self.telegram.send_message(f"Donated {amount}cc to {utils.COUNTRIES[71]}")
self.reporter.report_action("CONTRIBUTE_CC", data)
r = self._post_main_country_donate(**data)
return r.json().get('status') or not r.json().get('error')

58
erepublik/classes.py
View File

@ -5,9 +5,9 @@ import random
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
@ -122,7 +122,7 @@ class MyCompanies:
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
-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', {})
@ -335,7 +335,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):
@ -507,7 +507,8 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
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:
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))
@ -515,7 +516,8 @@ Class for unifying eRepublik known endpoints and their required/optional paramet
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:
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))
@ -774,7 +776,7 @@ 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)
@ -1139,3 +1141,45 @@ class EnergyToFight:
if 0 < new_energy < self.energy:
self.energy = new_energy
return self.energy
class TelegramBot:
chat_id = 0
api_url = ""
def __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
def send_message(self, message: str) -> bool:
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"))
return response.json().get('ok')
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):
message = ""
if self.player_name:
message = f"Player *{self.player_name}*\n"
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}*")