str.format() -> f-string
This commit is contained in:
Eriks K
parent
c0122eccdf
commit
c9971f3570
@ -81,8 +81,7 @@ class SlowRequests(Session):
|
||||
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)
|
||||
body = f"[{utils.now().strftime('%F %T')}]\tURL: '{url}'\tMETHOD: {method}\tARGS: {args}\n"
|
||||
with open(self.request_log_name, 'ab') as file:
|
||||
file.write(body.encode("UTF-8"))
|
||||
pass
|
||||
@ -95,20 +94,18 @@ class SlowRequests(Session):
|
||||
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 ""
|
||||
}
|
||||
fd_path = 'debug/requests'
|
||||
fd_time = self.last_time.strftime('%Y/%m/%d/%H-%M-%S')
|
||||
fd_name = utils.slugify(url[len(Citizen.url):])
|
||||
fd_extra = "_REDIRECT" if redirect else ""
|
||||
|
||||
try:
|
||||
utils.json.loads(resp.text)
|
||||
file_data.update({"ext": "json"})
|
||||
fd_ext = 'json'
|
||||
except utils.json.JSONDecodeError:
|
||||
file_data.update({"ext": "html"})
|
||||
fd_ext = 'html'
|
||||
|
||||
filename = 'debug/requests/{time}_{name}{extra}.{ext}'.format(**file_data)
|
||||
filename = f'{fd_path}/{fd_time}_{fd_name}{fd_extra}.{fd_ext}'
|
||||
utils.write_file(filename, resp.text)
|
||||
pass
|
||||
|
||||
@ -483,7 +480,7 @@ class ErepublikPoliticsAPI(CitizenBaseAPI):
|
||||
class ErepublikPresidentAPI(CitizenBaseAPI):
|
||||
def _post_wars_attack_region(self, war_id: int, region_id: int, region_name: str) -> Response:
|
||||
data = {'_token': self.token, 'warId': war_id, 'regionName': region_name, 'regionNameConfirm': region_name}
|
||||
return self.post('{}/wars/attack-region/{}/{}'.format(self.url, war_id, region_id), data=data)
|
||||
return self.post(f'{self.url}/wars/attack-region/{war_id}/{region_id}', data=data)
|
||||
|
||||
def _post_new_war(self, self_country_id: int, attack_country_id: int, debate: str = "") -> Response:
|
||||
data = dict(requirments=1, _token=self.token, debate=debate,
|
||||
|
||||
@ -607,8 +607,8 @@ class BaseCitizen(access_points.CitizenAPI):
|
||||
except AttributeError:
|
||||
continue
|
||||
if data:
|
||||
msgs = ["{count} x {kind}, totaling {} {currency}\n"
|
||||
"{about}".format(d["count"] * d["reward"], **d) for d in data.values()]
|
||||
msgs = [f"{d['count']} x {d['kind']}, totaling {d['count'] * d['reward']} "
|
||||
f"{d['currency']}" for d in data.values()]
|
||||
|
||||
msgs = "\n".join(msgs)
|
||||
if self.config.telegram:
|
||||
@ -1471,9 +1471,8 @@ class CitizenMedia(BaseCitizen):
|
||||
article_id = 0
|
||||
return article_id
|
||||
else:
|
||||
raise classes.ErepublikException("Article kind must be one of:\n{}\n'{}' is not supported".format(
|
||||
"\n".join(["{}: {}".format(k, v) for k, v in kinds.items()]), kind
|
||||
))
|
||||
kinds = "\n".join([f"{k}: {v}" for k, v in kinds.items()])
|
||||
raise classes.ErepublikException(f"Article kind must be one of:\n{kinds}\n'{kind}' is not supported")
|
||||
|
||||
def get_article(self, article_id: int) -> Dict[str, Any]:
|
||||
return self._get_main_article_json(article_id).json()
|
||||
@ -1812,7 +1811,7 @@ class CitizenMilitary(CitizenTravel):
|
||||
else:
|
||||
self._eat('blue')
|
||||
if self.energy.recovered < 50 or error_count >= 10 or count <= 0:
|
||||
self.write_log("Hits: {:>4} | Damage: {}".format(total_hits, total_damage))
|
||||
self.write_log(f"Hits: {total_hits:>4} | Damage: {total_damage}")
|
||||
ok_to_fight = False
|
||||
if total_damage:
|
||||
self.reporter.report_fighting(battle, not side.is_defender, division, total_damage, total_hits)
|
||||
@ -2317,7 +2316,7 @@ class CitizenTasks(BaseCitizen):
|
||||
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.write_log(f"I don't have energy to work. Will sleep for {seconds}s")
|
||||
self.sleep(seconds)
|
||||
self._eat("blue")
|
||||
self.work()
|
||||
@ -2450,7 +2449,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
self.reporter.do_init()
|
||||
if self.config.telegram and self.config.telegram_chat_id:
|
||||
self.telegram.do_init(self.config.telegram_chat_id,
|
||||
self.config.telegram_token or "864251270:AAFzZZdjspI-kIgJVk4gF3TViGFoHnf8H4o",
|
||||
self.config.telegram_token,
|
||||
self.name)
|
||||
self.telegram.send_message(f"*Started* {utils.now():%F %T}")
|
||||
|
||||
@ -2508,8 +2507,8 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
data[(title, reward)]['count'] += count
|
||||
self._post_main_global_alerts_close(medal.get('id'))
|
||||
if data:
|
||||
msgs = ["{count} x {kind},"
|
||||
" totaling {} {currency}".format(d["count"] * d["reward"], **d) for d in data.values()]
|
||||
msgs = [f"{d['count']} x {d['kind']}, totaling {d['count'] * d['reward']} "
|
||||
f"{d['currency']}" for d in data.values()]
|
||||
|
||||
msgs = "\n".join(msgs)
|
||||
if self.config.telegram:
|
||||
@ -2642,7 +2641,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
if not amount:
|
||||
inv_resp = self._get_economy_inventory_items().json()
|
||||
category = "rawMaterials" if kind.endswith("Raw") else "finalProducts"
|
||||
item = "{}_{}".format(constants.INDUSTRIES[kind], quality)
|
||||
item = f"{constants.INDUSTRIES[kind]}_{quality}"
|
||||
amount = inv_resp.get("inventoryItems").get(category).get("items").get(item).get("amount", 0)
|
||||
|
||||
if amount >= 1:
|
||||
@ -2671,7 +2670,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
elif kind.endswith("Raw"):
|
||||
self.sell_produced_product(kind, 1)
|
||||
else:
|
||||
raise classes.ErepublikException("Unknown kind produced '{kind}'".format(kind=kind))
|
||||
raise classes.ErepublikException(f"Unknown kind produced '{kind}'")
|
||||
elif self.config.auto_buy_raw and re.search(r"not_enough_[^_]*_raw", response.get("message")):
|
||||
raw_kind = re.search(r"not_enough_(\w+)_raw", response.get("message"))
|
||||
if raw_kind:
|
||||
@ -2717,7 +2716,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
self._report_action("WORK_AS_MANAGER", "Not enough money to work as manager!", kwargs=response)
|
||||
self.write_log("Not enough money to work as manager!")
|
||||
else:
|
||||
msg = "I was not able to wam and or employ because:\n{}".format(response)
|
||||
msg = f"I was not able to wam and or employ because:\n{response}"
|
||||
self._report_action("WORK_AS_MANAGER", f"Worked as manager failed: {msg}", kwargs=response)
|
||||
self.write_log(msg)
|
||||
|
||||
@ -2756,7 +2755,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
|
||||
|
||||
wam_count = self.my_companies.get_total_wam_count()
|
||||
if wam_count:
|
||||
self.write_log("Wam ff lockdown is now {}, was {}".format(wam_count, self.my_companies.ff_lockdown))
|
||||
self.write_log(f"Wam ff lockdown is now {wam_count}, was {self.my_companies.ff_lockdown}")
|
||||
self.my_companies.ff_lockdown = wam_count
|
||||
self.travel_to_residence()
|
||||
return bool(wam_count)
|
||||
|
||||
@ -419,7 +419,7 @@ class Energy:
|
||||
self._recovery_time = utils.now()
|
||||
|
||||
def __repr__(self):
|
||||
return "{:4}/{:4} + {:4}, {:3}hp/6min".format(self.recovered, self.limit, self.recoverable, self.interval)
|
||||
return f"{self.recovered:4}/{self.limit:4} + {self.recoverable:4}, {self.interval:3}hp/6min"
|
||||
|
||||
def set_reference_time(self, recovery_time: datetime.datetime):
|
||||
self._recovery_time = recovery_time.replace(microsecond=0)
|
||||
@ -598,10 +598,10 @@ class Reporter:
|
||||
for unreported_data in self.__to_update:
|
||||
unreported_data.update(player_id=self.citizen_id, key=self.key)
|
||||
unreported_data = utils.json.loads(utils.json.dumps(unreported_data, cls=MyJSONEncoder))
|
||||
self._req.post("{}/bot/update".format(self.url), json=unreported_data)
|
||||
self._req.post(f"{self.url}/bot/update", json=unreported_data)
|
||||
self.__to_update.clear()
|
||||
data = utils.json.loads(utils.json.dumps(data, cls=MyJSONEncoder))
|
||||
r = self._req.post("{}/bot/update".format(self.url), json=data)
|
||||
r = self._req.post(f"{self.url}/bot/update", json=data)
|
||||
return r
|
||||
|
||||
def register_account(self):
|
||||
@ -609,8 +609,8 @@ class Reporter:
|
||||
try:
|
||||
r = self.__bot_update(dict(key=self.key, check=True, player_id=self.citizen_id))
|
||||
if not r.json().get("status"):
|
||||
self._req.post("{}/bot/register".format(self.url), json=dict(name=self.name, email=self.email,
|
||||
player_id=self.citizen_id))
|
||||
self._req.post(f"{self.url}/bot/register", json=dict(name=self.name, email=self.email,
|
||||
player_id=self.citizen_id))
|
||||
finally:
|
||||
self.__registered = True
|
||||
self.allowed = True
|
||||
@ -895,9 +895,9 @@ class Battle:
|
||||
time_now = utils.now()
|
||||
is_started = self.start < utils.now()
|
||||
if is_started:
|
||||
time_part = " {}".format(time_now - self.start)
|
||||
time_part = f" {time_now - self.start}"
|
||||
else:
|
||||
time_part = "-{}".format(self.start - time_now)
|
||||
time_part = f"-{self.start - time_now}"
|
||||
|
||||
return (f"Battle {self.id} for {self.region_name[:16]:16} | "
|
||||
f"{self.invader} : {self.defender} | Round time {time_part} | {'R'+str(self.zone_id):>3}")
|
||||
@ -981,7 +981,7 @@ class TelegramBot:
|
||||
self._threads = [t for t in self._threads if t.is_alive()]
|
||||
self._next_time = utils.good_timedelta(utils.now(), datetime.timedelta(seconds=20))
|
||||
if not self._threads:
|
||||
name = "telegram_{}send".format(f"{self.player_name}_" if self.player_name else "")
|
||||
name = f"telegram_{f'{self.player_name}_' if self.player_name else ''}send"
|
||||
send_thread = threading.Thread(target=self.__send_messages, name=name)
|
||||
send_thread.start()
|
||||
self._threads.append(send_thread)
|
||||
|
||||
@ -93,7 +93,7 @@ def interactive_sleep(sleep_seconds: int):
|
||||
# seconds = seconds % 30 if seconds % 30 else 30
|
||||
else:
|
||||
seconds = 1
|
||||
sys.stdout.write("\rSleeping for {:4} more seconds".format(sleep_seconds))
|
||||
sys.stdout.write(f"\rSleeping for {sleep_seconds:4} more seconds")
|
||||
sys.stdout.flush()
|
||||
time.sleep(seconds)
|
||||
sleep_seconds -= seconds
|
||||
@ -105,7 +105,7 @@ silent_sleep = time.sleep
|
||||
|
||||
def _write_log(msg, timestamp: bool = True, should_print: bool = False):
|
||||
erep_time_now = now()
|
||||
txt = "[{}] {}".format(erep_time_now.strftime('%F %T'), msg) if timestamp else msg
|
||||
txt = f"[{erep_time_now.strftime('%F %T')}] {msg}" if timestamp else msg
|
||||
txt = "\n".join(["\n".join(textwrap.wrap(line, 120)) for line in txt.splitlines()])
|
||||
if not os.path.isdir('log'):
|
||||
os.mkdir('log')
|
||||
@ -172,10 +172,10 @@ def write_request(response: requests.Response, is_error: bool = False):
|
||||
ext = "html"
|
||||
|
||||
if not is_error:
|
||||
filename = "debug/requests/{}_{}.{}".format(now().strftime('%F_%H-%M-%S'), name, ext)
|
||||
filename = f"debug/requests/{now().strftime('%F_%H-%M-%S')}_{name}.{ext}"
|
||||
write_file(filename, html)
|
||||
else:
|
||||
return {"name": "{}_{}.{}".format(now().strftime('%F_%H-%M-%S'), name, ext),
|
||||
return {"name": f"{now().strftime('%F_%H-%M-%S')}_{name}.{ext}",
|
||||
"content": html.encode('utf-8'),
|
||||
"mimetype": "application/json" if ext == "json" else "text/html"}
|
||||
|
||||
@ -196,14 +196,14 @@ def send_email(name: str, content: List[Any], player=None, local_vars: Dict[str,
|
||||
if promo:
|
||||
resp = {"name": "%s.html" % name, "mimetype": "text/html",
|
||||
"content": file_content_template.format(title="Promo", body="<br/>".join(content))}
|
||||
subject = "[eBot][{}] Promos: {}".format(now().strftime('%F %T'), name)
|
||||
subject = f"[eBot][{now().strftime('%F %T')}] Promos: {name}"
|
||||
|
||||
elif captcha:
|
||||
resp = {"name": "%s.html" % name, "mimetype": "text/html",
|
||||
"content": file_content_template.format(title="ReCaptcha", body="<br/>".join(content))}
|
||||
subject = "[eBot][{}] RECAPTCHA: {}".format(now().strftime('%F %T'), name)
|
||||
subject = f"[eBot][{now().strftime('%F %T')}] RECAPTCHA: {name}"
|
||||
else:
|
||||
subject = "[eBot][%s] Bug trace: %s" % (now().strftime('%F %T'), name)
|
||||
subject = f"[eBot][{now().strftime('%F %T')}] Bug trace: {name}"
|
||||
|
||||
body = "".join(traceback.format_stack()) + \
|
||||
"\n\n" + \
|
||||
|
||||
@ -97,7 +97,7 @@ def main():
|
||||
player.set_debug(CONFIG.get('debug', False))
|
||||
player.login()
|
||||
if CONFIG.get('battle_launcher'):
|
||||
name = "{}-battle_launcher-{}".format(player.name, threading.active_count() - 1)
|
||||
name = f"{player.name}-battle_launcher-{threading.active_count() - 1}"
|
||||
state_thread = threading.Thread(target=_battle_launcher, args=(player,), name=name)
|
||||
state_thread.start()
|
||||
|
||||
|
||||
@ -94,15 +94,15 @@ def main():
|
||||
closest_next_time = dt_max
|
||||
next_tasks = []
|
||||
for task, next_time in sorted(tasks.items(), key=lambda s: s[1]):
|
||||
next_tasks.append("{}: {}".format(next_time.strftime('%F %T'), task))
|
||||
next_tasks.append(f"{next_time.strftime('%F %T')}: {task}")
|
||||
if next_time < closest_next_time:
|
||||
closest_next_time = next_time
|
||||
sleep_seconds = int(utils.get_sleep_seconds(closest_next_time))
|
||||
if sleep_seconds <= 0:
|
||||
player.write_log(f"Loop detected! Offending task: '{next_tasks[0]}'")
|
||||
player.write_log("My next Tasks and there time:\n" + "\n".join(sorted(next_tasks)))
|
||||
player.write_log("Sleeping until (eRep): {} (sleeping for {}s)".format(
|
||||
closest_next_time.strftime("%F %T"), sleep_seconds))
|
||||
player.write_log(f"Sleeping until (eRep): {closest_next_time.strftime('%F %T')}"
|
||||
f" (sleeping for {sleep_seconds}s)")
|
||||
seconds_to_sleep = sleep_seconds if sleep_seconds > 0 else 0
|
||||
player.sleep(seconds_to_sleep)
|
||||
except Exception as e:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user