From 22dc18d80dbc2952d02e9ccc4bbd994a661d0fa1 Mon Sep 17 00:00:00 2001 From: Eriks Karls Date: Thu, 27 Feb 2020 16:03:58 +0200 Subject: [PATCH] Leaderboards are now directly available from Citizen instance --- erepublik/__init__.py | 2 +- erepublik/citizen.py | 270 +++++++++++++++++++++++------------------- 2 files changed, 147 insertions(+), 125 deletions(-) diff --git a/erepublik/__init__.py b/erepublik/__init__.py index de73290..4c09eaf 100644 --- a/erepublik/__init__.py +++ b/erepublik/__init__.py @@ -5,7 +5,7 @@ __author__ = """Eriks Karls""" __email__ = 'eriks@72.lv' __version__ = '0.20.0' -__commit_id__ = "bf5f3d7" +__commit_id__ = "772c09a" from erepublik import classes, utils from erepublik.citizen import Citizen diff --git a/erepublik/citizen.py b/erepublik/citizen.py index eb0649a..748e4e4 100644 --- a/erepublik/citizen.py +++ b/erepublik/citizen.py @@ -731,129 +731,6 @@ class CitizenTravel(BaseCitizen): return return_list -class CitizenTasks(BaseCitizen): - tg_contract: dict = {} - ot_points: int = 0 - next_ot_time: datetime = None - - def eat(self): - """ Eat food """ - self._eat("blue") - - def work(self): - if self.energy.food_fights >= 1: - response = self._post_economy_work("work") - js = response.json() - good_msg = ["already_worked", "captcha"] - if not js.get("status") and not js.get("message") in good_msg: - if js.get('message') == 'employee': - self.find_new_job() - self.update_citizen_info() - self.work() - else: - self.reporter.report_action("WORK", json_val=js) - else: - self._eat("blue") - if self.energy.food_fights < 1: - seconds = (self.energy.reference_time - self.now).total_seconds() - self.write_log("I don't have energy to work. Will sleep for {}s".format(seconds)) - self.sleep(seconds) - self._eat("blue") - self.work() - - def train(self): - r = self._get_main_training_grounds_json() - tg_json = r.json() - self.details.gold = tg_json["page_details"]["gold"] - self.tg_contract.update(free_train=tg_json["hasFreeTrain"]) - if tg_json["contracts"]: - self.tg_contract.update(**tg_json["contracts"][0]) - - tgs = [] - for data in sorted(tg_json["grounds"], key=lambda k: k["cost"]): - if data["default"] and not data["trained"]: - tgs.append(data["id"]) - if tgs: - if self.energy.food_fights >= len(tgs): - response = self._post_economy_train(tgs) - if not response.json().get("status"): - self.update_citizen_info() - self.train() - else: - self.reporter.report_action("TRAIN", response.json()) - else: - self._eat("blue") - if self.energy.food_fights < len(tgs): - large = max(self.energy.reference_time, self.now) - small = min(self.energy.reference_time, self.now) - self.write_log("I don't have energy to train. Will sleep for {} seconds".format( - (large - small).seconds)) - self.sleep(int((large - small).total_seconds())) - self._eat("blue") - self.train() - - def work_ot(self): - # I"m not checking for 1h cooldown. Beware of nightshift work, if calling more than once every 60min - self.update_job_info() - if self.ot_points >= 24 and self.energy.food_fights > 1: - r = self._post_economy_work_overtime() - if not r.json().get("status") and r.json().get("message") == "money": - self.resign_from_employer() - self.find_new_job() - else: - if r.json().get('message') == 'employee': - self.find_new_job() - self.reporter.report_action("WORK_OT", r.json()) - elif self.energy.food_fights < 1 and self.ot_points >= 24: - self._eat("blue") - if self.energy.food_fights < 1: - large = max(self.energy.reference_time, self.now) - small = min(self.energy.reference_time, self.now) - self.write_log("I don't have energy to work OT. Will sleep for {}s".format((large - small).seconds)) - self.sleep(int((large - small).total_seconds())) - self._eat("blue") - self.work_ot() - - def resign_from_employer(self) -> bool: - self.update_job_info() - if self.r.json().get("isEmployee"): - self.reporter.report_action("RESIGN", self.r.json()) - self._post_economy_resign() - return True - return False - - def buy_tg_contract(self) -> Response: - ret = self._post_main_buy_gold_items('gold', "TrainingContract2", 1) - self.reporter.report_action("BUY_TG_CONTRACT", ret.json()) - return ret - - def find_new_job(self) -> Response: - r = self._get_economy_job_market_json(self.details.current_country) - jobs = r.json().get("jobs") - data = dict(citizen=0, salary=10) - for posting in jobs: - salary = posting.get("salary") - limit = posting.get("salaryLimit", 0) - userid = posting.get("citizen").get("id") - - if (not limit or salary * 3 < limit) and salary > data["salary"]: - data.update({"citizen": userid, "salary": salary}) - self.reporter.report_action("APPLYING_FOR_JOB", jobs, str(data['citizen'])) - return self._post_economy_job_market_apply(**data) - - def apply_to_employer(self, employer_id: int, salary: float) -> bool: - data = dict(citizenId=employer_id, salary=salary) - self.reporter.report_action("APPLYING_FOR_JOB", data) - r = self._post_economy_job_market_apply(employer_id, salary) - return bool(r.json().get('status')) - - def update_job_info(self): - ot = self._get_main_job_data().json().get("overTime", {}) - if ot: - self.next_ot_time = utils.localize_timestamp(int(ot.get("nextOverTime", 0))) - self.ot_points = ot.get("points", 0) - - class CitizenEconomy(CitizenTravel): food: Dict[str, int] = {"q1": 0, "q2": 0, "q3": 0, "q4": 0, "q5": 0, "q6": 0, "q7": 0, "total": 0} inventory: Dict[str, int] = {"used": 0, "total": 0} @@ -1366,6 +1243,20 @@ class CitizenEconomy(CitizenTravel): return r.json().get('status') or not r.json().get('error') +class CitizenLeaderboard(BaseCitizen): + def get_aircraft_damage_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Dict[str, any]: + return self._get_main_leaderboards_damage_aircraft_rankings(country, weeks, mu).json() + + def get_ground_damage_rankings(self, country: int, weeks: int = 0, mu: int = 0, div: int = 4) -> Dict[str, any]: + return self._get_main_leaderboards_damage_rankings(country, weeks, mu, div).json() + + def get_aircraft_kill_rankings(self, country: int, weeks: int = 0, mu: int = 0) -> Dict[str, any]: + return self._get_main_leaderboards_kills_aircraft_rankings(country, weeks, mu).json() + + def get_ground_kill_rankings(self, country: int, weeks: int = 0, mu: int = 0, div: int = 4) -> Dict[str, any]: + return self._get_main_leaderboards_kills_rankings(country, weeks, mu, div).json() + + class CitizenMedia(BaseCitizen): def endorse_article(self, article_id: int, amount: int) -> bool: if amount in (5, 50, 100): @@ -2105,8 +1996,139 @@ class CitizenSocial(BaseCitizen): else: self.report_error(f"Unsupported notification kind: \"{kind}\"!") + def get_citizen_profile(self, player_id: int = None): + if player_id is None: + player_id = self.details.citizen_id + return self._get_main_citizen_profile_json(player_id).json() -class Citizen(CitizenAnniversary, CitizenEconomy, CitizenMedia, CitizenMilitary, CitizenPolitics, CitizenSocial, CitizenTasks): + +class CitizenTasks(BaseCitizen): + tg_contract: dict = {} + ot_points: int = 0 + next_ot_time: datetime = None + + def eat(self): + """ Eat food """ + self._eat("blue") + + def work(self): + if self.energy.food_fights >= 1: + response = self._post_economy_work("work") + js = response.json() + good_msg = ["already_worked", "captcha"] + if not js.get("status") and not js.get("message") in good_msg: + if js.get('message') == 'employee': + self.find_new_job() + self.update_citizen_info() + self.work() + else: + self.reporter.report_action("WORK", json_val=js) + else: + self._eat("blue") + if self.energy.food_fights < 1: + seconds = (self.energy.reference_time - self.now).total_seconds() + self.write_log("I don't have energy to work. Will sleep for {}s".format(seconds)) + self.sleep(seconds) + self._eat("blue") + self.work() + + def train(self): + r = self._get_main_training_grounds_json() + tg_json = r.json() + self.details.gold = tg_json["page_details"]["gold"] + self.tg_contract.update(free_train=tg_json["hasFreeTrain"]) + if tg_json["contracts"]: + self.tg_contract.update(**tg_json["contracts"][0]) + + tgs = [] + for data in sorted(tg_json["grounds"], key=lambda k: k["cost"]): + if data["default"] and not data["trained"]: + tgs.append(data["id"]) + if tgs: + if self.energy.food_fights >= len(tgs): + response = self._post_economy_train(tgs) + if not response.json().get("status"): + self.update_citizen_info() + self.train() + else: + self.reporter.report_action("TRAIN", response.json()) + else: + self._eat("blue") + if self.energy.food_fights < len(tgs): + large = max(self.energy.reference_time, self.now) + small = min(self.energy.reference_time, self.now) + self.write_log("I don't have energy to train. Will sleep for {} seconds".format( + (large - small).seconds)) + self.sleep(int((large - small).total_seconds())) + self._eat("blue") + self.train() + + def work_ot(self): + # I"m not checking for 1h cooldown. Beware of nightshift work, if calling more than once every 60min + self.update_job_info() + if self.ot_points >= 24 and self.energy.food_fights > 1: + r = self._post_economy_work_overtime() + if not r.json().get("status") and r.json().get("message") == "money": + self.resign_from_employer() + self.find_new_job() + else: + if r.json().get('message') == 'employee': + self.find_new_job() + self.reporter.report_action("WORK_OT", r.json()) + elif self.energy.food_fights < 1 and self.ot_points >= 24: + self._eat("blue") + if self.energy.food_fights < 1: + large = max(self.energy.reference_time, self.now) + small = min(self.energy.reference_time, self.now) + self.write_log("I don't have energy to work OT. Will sleep for {}s".format((large - small).seconds)) + self.sleep(int((large - small).total_seconds())) + self._eat("blue") + self.work_ot() + + def resign_from_employer(self) -> bool: + self.update_job_info() + if self.r.json().get("isEmployee"): + self.reporter.report_action("RESIGN", self.r.json()) + self._post_economy_resign() + return True + return False + + def buy_tg_contract(self) -> Response: + ret = self._post_main_buy_gold_items('gold', "TrainingContract2", 1) + self.reporter.report_action("BUY_TG_CONTRACT", ret.json()) + return ret + + def find_new_job(self) -> Response: + r = self._get_economy_job_market_json(self.details.current_country) + jobs = r.json().get("jobs") + data = dict(citizen=0, salary=10) + for posting in jobs: + salary = posting.get("salary") + limit = posting.get("salaryLimit", 0) + userid = posting.get("citizen").get("id") + + if (not limit or salary * 3 < limit) and salary > data["salary"]: + data.update({"citizen": userid, "salary": salary}) + self.reporter.report_action("APPLYING_FOR_JOB", jobs, str(data['citizen'])) + return self._post_economy_job_market_apply(**data) + + def apply_to_employer(self, employer_id: int, salary: float) -> bool: + data = dict(citizenId=employer_id, salary=salary) + self.reporter.report_action("APPLYING_FOR_JOB", data) + r = self._post_economy_job_market_apply(employer_id, salary) + return bool(r.json().get('status')) + + def update_job_info(self): + ot = self._get_main_job_data().json().get("overTime", {}) + if ot: + self.next_ot_time = utils.localize_timestamp(int(ot.get("nextOverTime", 0))) + self.ot_points = ot.get("points", 0) + + +class Citizen( + CitizenAnniversary, CitizenEconomy, CitizenLeaderboard, CitizenMedia, CitizenMilitary, + CitizenPolitics, CitizenSocial, CitizenTasks +): debug: bool = False def __init__(self, email: str = "", password: str = "", auto_login: bool = True):