Flake8

Battle.div is no longer a defaultdict, but should raise error if trying to access division which isn't assigned

erepublik/__init__.py

@@ -5,7 +5,7 @@
 __author__ = """Eriks Karls"""
 __email__ = 'eriks@72.lv'
 __version__ = '0.20.1.3'
-__commit_id__ = "2f69090"
+__commit_id__ = "b49e4ab"
 
 from erepublik import classes, utils
 from erepublik.citizen import Citizen

erepublik/citizen.py

@@ -12,8 +12,8 @@ from requests import HTTPError, RequestException, Response
 
 from erepublik import utils
 from erepublik.access_points import CitizenAPI
-from erepublik.classes import (Battle, BattleDivision, Config, Details, Energy, ErepublikException,
+from erepublik.classes import Battle, BattleDivision, Config, Details, Energy, \
-                               MyCompanies, MyJSONEncoder, OfferItem, Politics, Reporter, TelegramBot)
+    ErepublikException, MyCompanies, MyJSONEncoder, OfferItem, Politics, Reporter, TelegramBot
 
 
 class BaseCitizen(CitizenAPI):
@@ -664,7 +664,16 @@ class BaseCitizen(CitizenAPI):
         return bool(re.search(r'body id="error"|Internal Server Error|'
                               r'CSRF attack detected|meta http-equiv="refresh"|not_authenticated', response.text))
 
-    def _report_action(self, action: str, msg: str, **kwargs):
+    def _report_action(self, action: str, msg: str, **kwargs: Optional[Dict[str, Any]]):
+        """ Report action to all available reporting channels
+
+        :type action: str
+        :type msg: str
+        :type kwargs: Optional[Dict[str, Any]]
+        :param action: Action taken
+        :param msg: Message about the action
+        :param kwargs: Extra information regarding action
+        """
         kwargs = utils.json.loads(utils.json.dumps(kwargs or {}, cls=MyJSONEncoder))
         action = action[:32]
         self.write_log(msg)
@@ -838,8 +847,8 @@ class CitizenCompanies(BaseCitizen):
                 raw_factories = True
 
             free_inventory = self.inventory["total"] - self.inventory["used"]
-            wam_list = self.my_companies.get_holding_wam_companies(wam_holding_id,
+            wam_list = self.my_companies.get_holding_wam_companies(
-                                                                   raw_factory=raw_factories)[:self.energy.food_fights]
+                wam_holding_id, raw_factory=raw_factories)[:self.energy.food_fights]
             has_space = False
             while not has_space and wam_list:
                 extra_needed = self.my_companies.get_needed_inventory_usage(companies=wam_list)
@@ -859,7 +868,8 @@ class CitizenCompanies(BaseCitizen):
             if sum(employ_factories.values()) > self.my_companies.work_units:
                 employ_factories = {}
 
-            response = self._post_economy_work("production", wam=[c.id for c in wam_list], employ=employ_factories).json()
+            response = self._post_economy_work("production", wam=[c.id for c in wam_list],
+                                               employ=employ_factories).json()
             return response
 
     def update_companies(self):
@@ -898,8 +908,8 @@ class CitizenCompanies(BaseCitizen):
                         18: "HRM q1", 19: "HRM q2", 20: "HRM q3", 21: "HRM q4", 22: "HRM q5",
                         24: "ARM q1", 25: "ARM q2", 26: "ARM q3", 27: "ARM q4", 28: "ARM q5", }[industry_id]
         if building_type == 2:
-            company_name = f"Storage"
+            company_name = "Storage"
-        self.write_log(f"{company_name} created!")
+        self.write_log(f'{company_name} created!')
         return self._post_economy_create_company(industry_id, building_type)
 
     def dissolve_factory(self, factory_id: int) -> Response:
@@ -1149,7 +1159,8 @@ class CitizenEconomy(CitizenTravel):
             self.stop_threads.wait()
             return False
         else:
-            self._report_action("BUY_GOLD", f"New amount {self.details.cc}cc, {self.details.gold}g", kwargs=response.json())
+            self._report_action('BUY_GOLD', f'New amount {self.details.cc}cc, {self.details.gold}g',
+                                kwargs=response.json())
             return True
 
     def donate_money(self, citizen_id: int = 1620414, amount: float = 0.0, currency: int = 62) -> bool:
@@ -1177,22 +1188,22 @@ class CitizenEconomy(CitizenTravel):
             self._report_action("DONATE_ITEMS", msg, success=True)
             return amount
         elif re.search('You must wait 5 seconds before donating again', response.text):
-            self.write_log(f"Previous donation failed! Must wait at least 5 seconds before next donation!")
+            self.write_log('Previous donation failed! Must wait at least 5 seconds before next donation!')
             self.sleep(5)
             return self.donate_items(citizen_id, int(amount), industry_id, quality)
         else:
-            if re.search(r"You do not have enough items in your inventory to make this donation", response.text):
+            if re.search(r'You do not have enough items in your inventory to make this donation', response.text):
                 self._report_action("DONATE_ITEMS",
                                     f"Unable to donate {amount}q{quality} "
                                     f"{self.get_industry_name(industry_id)}, not enough left!", success=False)
                 return 0
             available = re.search(
-                rf"Cannot transfer the items because the user has only (\d+) free slots in (his|her) storage.",
+                r'Cannot transfer the items because the user has only (\d+) free slots in (his|her) storage.',
                 response.text
             ).group(1)
-            self._report_action("DONATE_ITEMS",
+            self._report_action('DONATE_ITEMS',
-                                f"Unable to donate {amount}q{quality}{self.get_industry_name(industry_id)}"
+                                f'Unable to donate {amount}q{quality}{self.get_industry_name(industry_id)}'
-                                f", receiver has only {available} storage left!", success=False)
+                                f', receiver has only {available} storage left!', success=False)
             self.sleep(5)
             return self.donate_items(citizen_id, int(available), industry_id, quality)
 
@@ -1204,7 +1215,7 @@ class CitizenEconomy(CitizenTravel):
         data = dict(country=country_id, action='currency', value=amount)
         r = self._post_main_country_donate(**data)
         if r.json().get('status') or not r.json().get('error'):
-            self._report_action("CONTRIBUTE_CC", f"Contributed {amount}cc to {utils.COUNTRIES[country_id]}'s treasury",
+            self._report_action("CONTRIBUTE_CC", f'Contributed {amount}cc to {utils.COUNTRIES[country_id]}\'s treasury',
                                 success=True)
             return True
         else:
@@ -1402,7 +1413,8 @@ class CitizenMilitary(CitizenTravel):
         battle_zone = battle.div[11 if battle.is_air else self.division].battle_zone_id
         r = self._post_military_change_weapon(battle_id, battle_zone, quality)
         influence = r.json().get('weaponInfluence')
-        self._report_action("MILITARY_WEAPON", f"Switched to q{quality} weapon, new influence {influence}", kwargs=r.json())
+        self._report_action("MILITARY_WEAPON", f"Switched to q{quality} weapon,"
+                                               f" new influence {influence}", kwargs=r.json())
         return influence
 
     def check_epic_battles(self):
@@ -1776,11 +1788,11 @@ class CitizenMilitary(CitizenTravel):
         return quality
 
     def activate_battle_effect(self, battle_id: int, kind: str) -> Response:
-        self._report_action("MILITARY_BOOSTER", f"Activated {kind} booster")
+        self._report_action('MILITARY_BOOSTER', f'Activated {kind} booster')
         return self._post_main_activate_battle_effect(battle_id, kind, self.details.citizen_id)
 
     def activate_pp_booster(self, battle_id: int) -> Response:
-        self._report_action("MILITARY_BOOSTER", f"Activated PrestigePoint booster")
+        self._report_action('MILITARY_BOOSTER', 'Activated PrestigePoint booster')
         return self._post_military_fight_activate_booster(battle_id, 1, 180, "prestige_points")
 
     def _rw_choose_side(self, battle_id: int, side_id: int) -> Response:
@@ -1957,29 +1969,29 @@ class CitizenPolitics(BaseCitizen):
         return ret
 
     def candidate_for_congress(self, presentation: str = "") -> Response:
-        self._report_action("POLITIC_CONGRESS", f"Applied for congress elections")
+        self._report_action('POLITIC_CONGRESS', 'Applied for congress elections')
         return self._post_candidate_for_congress(presentation)
 
     def candidate_for_party_presidency(self) -> Response:
-        self._report_action("POLITIC_PARTY_PRESIDENT", f"Applied for party president elections")
+        self._report_action('POLITIC_PARTY_PRESIDENT', 'Applied for party president elections')
         return self._get_candidate_party(self.politics.party_slug)
 
 
 class CitizenSocial(BaseCitizen):
     def send_mail_to_owner(self):
         if not self.details.citizen_id == 1620414:
-            self.send_mail("Started", "time {}".format(self.now.strftime("%Y-%m-%d %H-%M-%S")), [1620414, ])
+            self.send_mail('Started', f'time {self.now.strftime("%Y-%m-%d %H-%M-%S")}', [1620414, ])
             self.sleep(1)
-            msg_id = re.search(r"<input type=\"hidden\" value=\"(\d+)\" "
+            msg_id = re.search(r'<input type="hidden" value="(\d+)" '
-                               r"id=\"delete_message_(\d+)\" name=\"delete_message\[]\">", self.r.text).group(1)
+                               r'id="delete_message_(\d+)" name="delete_message\[]">', self.r.text).group(1)
             self._post_delete_message([msg_id])
 
     def send_mail(self, subject: str, msg: str, ids: List[int] = None):
         if ids is None:
             ids = [1620414, ]
         for player_id in ids:
-            self._report_action("SOCIAL_MESSAGE", f"Sent a message to {player_id}", kwargs=dict(subject=subject, msg=msg,
+            self._report_action('SOCIAL_MESSAGE', f'Sent a message to {player_id}',
-                                                                                                id=player_id))
+                                kwargs=dict(subject=subject, msg=msg, id=player_id))
             self._post_main_messages_compose(subject, msg, [player_id])
 
     def write_on_country_wall(self, message: str) -> bool:
@@ -1988,16 +2000,16 @@ class CitizenSocial(BaseCitizen):
                                      self.r.text, re.S | re.M)
         r = self._post_main_country_post_create(message, max(post_to_wall_as, key=int) if post_to_wall_as else 0)
 
-        self._report_action("SOCIAL_WRITE_WALL_COUNTRY", f"Wrote a message to the country wall", kwargs=message)
+        self._report_action('SOCIAL_WRITE_WALL_COUNTRY', 'Wrote a message to the country wall', kwargs=message)
         return r.json()
 
     def add_friend(self, player_id: int) -> Response:
         resp = self._get_main_citizen_hovercard(player_id)
         r_json = resp.json()
-        if not any([r_json["isBanned"], r_json["isDead"], r_json["isFriend"], r_json["isOrg"], r_json["isSelf"]]):
+        if not any([r_json['isBanned'], r_json['isDead'], r_json['isFriend'], r_json['isOrg'], r_json['isSelf']]):
             r = self._post_main_citizen_add_remove_friend(int(player_id), True)
             self.write_log(f"{r_json['name']:<64} (id:{player_id:>11}) added as friend")
-            self._report_action("SOCIAL_ADD_FRIEND", f"{r_json['name']:<64} (id:{player_id:>11}) added as friend")
+            self._report_action('SOCIAL_ADD_FRIEND', f"{r_json['name']:<64} (id:{player_id:>11}) added as friend")
             return r
         return resp
 
@@ -2164,7 +2176,7 @@ class CitizenTasks(BaseCitizen):
     def resign_from_employer(self) -> bool:
         r = self.update_job_info()
         if r.json().get("isEmployee"):
-            self._report_action("ECONOMY_RESIGN", f"Resigned from employer!", kwargs=r.json())
+            self._report_action('ECONOMY_RESIGN', 'Resigned from employer!', kwargs=r.json())
             self._post_economy_resign()
             return True
         return False
@@ -2175,7 +2187,7 @@ class CitizenTasks(BaseCitizen):
             extra = ret.json()
         except:  # noqa
             extra = {}
-        self._report_action("ECONOMY_TG_CONTRACT", f"Bought TG Contract", kwargs=extra)
+        self._report_action('ECONOMY_TG_CONTRACT', 'Bought TG Contract', kwargs=extra)
         return ret
 
     def find_new_job(self) -> Response:
@@ -2449,7 +2461,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
         if response is None:
             return
         if response.get("status"):
-            self._report_action("WORK_AS_MANAGER", f"Worked as manager", kwargs=response)
+            self._report_action('WORK_AS_MANAGER', 'Worked as manager', kwargs=response)
             if self.config.auto_sell:
                 for kind, data in response.get("result", {}).get("production", {}).items():
                     if data and kind in self.config.auto_sell:
@@ -2527,7 +2539,7 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
             self.travel_to_residence()
             return bool(wam_count)
         else:
-            self.write_log("Did not WAM because I would mess up levelup!")
+            self.write_log('Did not WAM because I would mess up levelup!')
             self.my_companies.ff_lockdown = 0
 
         self.update_companies()
@@ -2550,4 +2562,4 @@ class Citizen(CitizenAnniversary, CitizenCompanies, CitizenEconomy, CitizenLeade
                         continue
                 self.stop_threads.wait(90)
             except:  # noqa
-                self.report_error("Command central has broken")
+                self.report_error('Command central is broken')

erepublik/classes.py

@@ -1,9 +1,8 @@
 import datetime
 import hashlib
 import threading
-from collections import defaultdict
 from decimal import Decimal
-from typing import Any, Dict, List, NamedTuple, Tuple, Union, Optional
+from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union
 
 from requests import Response, Session, post
 
@@ -296,8 +295,8 @@ class MyCompanies:
 
     @property
     def __dict__(self):
-        return dict(name=str(self), work_units=self.work_units, next_ot_time=self.next_ot_time, ff_lockdown=self.ff_lockdown, holdings=self.holdings,
+        return dict(name=str(self), work_units=self.work_units, next_ot_time=self.next_ot_time,
-                    company_count=len(self.companies))
+                    ff_lockdown=self.ff_lockdown, holdings=self.holdings, company_count=len(self.companies))
 
 
 class Config:
@@ -643,21 +642,28 @@ class BattleDivision:
     def div_end(self) -> bool:
         return utils.now() >= self.end
 
-    def __init__(self, **kwargs):
+    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]):
         """Battle division helper class
 
-        :param kwargs: must contain keys:
+        :type div_id: int
-            div_id: int, end: datetime.datetime, epic: bool, inv_pts: int, def_pts: int,
+        :type end: datetime.datetime
-            wall_for: int, wall_dom: float, def_medal: Tuple[int, int], inv_medal: Tuple[int, int]
+        :type epic: bool
+        :type inv_pts: int
+        :type def_pts: int
+        :type wall_for: int
+        :type wall_dom: float
+        :type def_medal: Tuple[int, int]
+        :type inv_medal: Tuple[int, int]
         """
 
-        self.battle_zone_id = kwargs.get("div_id", 0)
+        self.battle_zone_id = div_id
-        self.end = kwargs.get("end", 0)
+        self.end = end
-        self.epic = kwargs.get("epic", 0)
+        self.epic = epic
-        self.dom_pts = dict({"inv": kwargs.get("inv_pts", 0), "def": kwargs.get("def_pts", 0)})
+        self.dom_pts = dict({"inv": inv_pts, "def": def_pts})
-        self.wall = dict({"for": kwargs.get("wall_for", 0), "dom": kwargs.get("wall_dom", 0)})
+        self.wall = dict({"for": wall_for, "dom": wall_dom})
-        self.def_medal = {"id": kwargs.get("def_medal", 0)[0], "dmg": kwargs.get("def_medal", 0)[1]}
+        self.def_medal = {"id": def_medal[0], "dmg": def_medal[1]}
-        self.inv_medal = {"id": kwargs.get("inv_medal", 0)[0], "dmg": kwargs.get("inv_medal", 0)[1]}
+        self.inv_medal = {"id": inv_medal[0], "dmg": inv_medal[1]}
 
     @property
     def id(self):
@@ -705,7 +711,7 @@ class Battle:
             [row.get('id') for row in battle.get('def', {}).get('ally_list') if row['deployed']]
         )
 
-        self.div = defaultdict(BattleDivision)
+        self.div = {}
         for div, data in battle.get('div', {}).items():
             div = int(data.get('div'))
             if data.get('end'):
@@ -721,7 +727,7 @@ class Battle:
                 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'),
+            battle_div = BattleDivision(div_id=data.get('id'), 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"),
                                         def_medal=def_medal, inv_medal=inv_medal)

erepublik/utils.py

@@ -9,7 +9,7 @@ import traceback
 import unicodedata
 from decimal import Decimal
 from pathlib import Path
-from typing import Any, List, Mapping, Optional, Union, Dict
+from typing import Any, Dict, List, Mapping, Optional, Union
 
 import pytz
 import requests
@@ -323,7 +323,8 @@ def send_email(name: str, content: List[Any], player=None, local_vars: Mapping[A
         if "state_thread" in local_vars:
             local_vars.pop('state_thread', None)
         from erepublik.classes import MyJSONEncoder
-        files.append(('file', ("local_vars.json", json.dumps(local_vars, cls=MyJSONEncoder, sort_keys=True), "application/json")))
+        files.append(('file', ("local_vars.json", json.dumps(local_vars, cls=MyJSONEncoder, sort_keys=True),
+                               "application/json")))
     if isinstance(player, Citizen):
         files.append(('file', ("instance.json", player.to_json(indent=True), "application/json")))
     requests.post('https://pasts.72.lv', data=data, files=files)

setup.cfg

@@ -18,8 +18,25 @@ replace = __version__ = '{new_version}'
 universal = 1
 
 [flake8]
-exclude = docs
+exclude = docs,.tox,.git,log,debug,venv
 max-line-length = 120
-ignore = E722 F401
+ignore = D100,D101,D102,D103
+;old_ignore = E722 F401
 
-[aliases]
+[pycodestyle]
+max-line-length = 120
+exclude = .tox,.git,log,debug,venv, build
+
+[mypy]
+python_version = 3.7
+check_untyped_defs = True
+ignore_missing_imports = False
+warn_unused_ignores = True
+warn_redundant_casts = True
+warn_unused_configs = True
+plugins = mypy_django_plugin.main
+
+[isort]
+multi_line_output = 2
+line_length = 120
+not_skip = __init__.py

setup.py

@@ -3,7 +3,7 @@
 
 """The setup script."""
 
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 with open('README.rst') as readme_file:
     readme = readme_file.read()

tests/test_erepublik_script.py

@@ -3,10 +3,10 @@
 
 """Tests for `erepublik` package."""
 
-import unittest
-
 from erepublik import Citizen
 
+import unittest
+
 
 class TestErepublik(unittest.TestCase):
     """Tests for `erepublik` package."""