diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 99c76edb..6e38193a 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -10,12 +10,13 @@ jobs: # You can use PyPy versions in python-version. # For example, pypy2 and pypy3 matrix: - python-version: [3.8] +# python-version: ["3.7.13", "3.8.13", "3.9.13", "3.10"] + python-version: [3.8] steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 + uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} # You can test your matrix by printing the current Python version @@ -24,7 +25,6 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install discord.py==1.5.0 pip install -r requirements.txt - name: Running examples as tests env: @@ -35,7 +35,6 @@ jobs: LINKS_API_PASSWORD: ${{ secrets.LINKS_API_PASSWORD }} RUNNING_TESTS: true run: | - python -m examples.discord_bot python -m examples.discord_links - python -m examples.events + python -m examples.events_example python -m examples.war_logs diff --git a/.gitignore b/.gitignore index f1e5497d..bff38fa9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ # Byte-compiled / optimized / DLL files +.idea/ __pycache__/ *.py[cod] *$py.class @@ -82,11 +83,11 @@ celerybeat-schedule *.sage.py # Environments -.env +examples/.env .venv -env/ +examples/.env/ venv/ -ENV/ +examples/.env/ env.bak/ venv.bak/ @@ -107,4 +108,4 @@ venv.bak/ examples/creds.py # vscode -.vscode/ \ No newline at end of file +.vscode/ diff --git a/README.rst b/README.rst index 97519056..aec3a3de 100644 --- a/README.rst +++ b/README.rst @@ -26,7 +26,7 @@ Getting Started Installing ----------- -**Python 3.5 or higher is required** +**Python 3.7 or higher is required** .. code:: sh @@ -47,26 +47,39 @@ This example will get a player with a certain tag, and search for 5 clans with a .. code:: py + import asyncio import coc - client = coc.login('email', 'password') async def main(): + coc_client = coc.Client() + try: + await coc_client.login("email", "password") + except coc.invalidcredentials as error: + exit(error) + player = await client.get_player("tag") - print("{0.name} has {0.trophies} trophies!".format(player)) + print(f"{player.name} has {player.trophies} trophies!") - clans = await client.search_clans(name="Best Clan Ever", limit=5) + clans = await client.search_clans(name="best clan ever", limit=5) for clan in clans: - print("{0.name} ({0.tag}) has {0.member_count} members".format(clan)) + print(f"{clan.name} ({clan.tag}) has {clan.member_count} members") try: war = await client.get_current_war("#clantag") - print("{0.clan_tag} is currently in {0.state} state.".format(war)) - except coc.PrivateWarLog: - print("Uh oh, they have a private war log!") + print(f"{war.clan_tag} is currently in {war.state} state.") + except coc.privatewarlog: + print("uh oh, they have a private war log!") - client.loop.run_until_complete(main()) - client.close() + # make sure to close the session or you will get asyncio + # task pending errors + await client.close() + + if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass Basic Events Example --------------------- @@ -75,24 +88,52 @@ whenever someone joins the clan or a member of the clan donates troops. .. code:: py + import asyncio + import logging + import coc - client = coc.login('email', 'password', client=coc.EventsClient) - @client.event - @coc.ClanEvents.member_join(tags=["#clantag", "#clantag2"]) + @coc.ClanEvents.member_join() async def foo(player, clan): - print("{0.name} ({0.tag}) just joined {1.name} ({1.tag})!".format(player, clan)) + print(f"{player.name} ({player.tag}) just joined {clan.name} ({clan.tag})") + - @client.event - @coc.ClanEvents.member_donations(tags=["#clantag", "#clantag2"]) + @coc.ClanEvents.member_donations() async def bar(old_member, member): troops_donated = member.donations - old_member.donations - print("{0} just donated {1} troops!".format(member.name, troops_donated)) + print(f"{member.name} just donated {troops_donated} troops!") + - client.run_forever() + async def main(): + coc_client = coc.EVentsClient() + try: + await coc.login("email", "password") + except coc.InvalidCredentials as error: + exit(error) + + # Register all the clans you want to monitor + list_of_clan_tags = ["tag1", "tag2", "tag3"] + coc_client.add_clan_updates(*list_of_clan_tags) + + # Register the callbacks for each of the events you are monitoring + coc_client.add_events( + foo, + bar + ) + if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + log = logging.getLogger() + + loop = asyncio.get_event_loop() + try: + loop.run_until_complete(main()) + loop.run_forever() + except KeyboardInterrupt: + pass + For more examples see the examples directory Contributing diff --git a/coc/__init__.py b/coc/__init__.py index a7d35572..ba4d290b 100644 --- a/coc/__init__.py +++ b/coc/__init__.py @@ -22,7 +22,7 @@ SOFTWARE. """ -__version__ = "2.0.1" +__version__ = "2.1.0" from .abc import BasePlayer, BaseClan from .clans import RankedClan, Clan diff --git a/coc/abc.py b/coc/abc.py index baf11287..45240aa9 100644 --- a/coc/abc.py +++ b/coc/abc.py @@ -105,7 +105,11 @@ def get_detailed_members(self, cls: Type["Player"] = None, load_game_data: bool if load_game_data and not isinstance(load_game_data, bool): raise TypeError("load_game_data must be either True or False.") - return PlayerIterator(self._client, (p.tag for p in self.members), cls=cls, load_game_data=load_game_data) + return PlayerIterator(self._client, + (p.tag for p in self.members), + cls=cls, + load_game_data=load_game_data, + members=self.members_dict) class BasePlayer: diff --git a/coc/clans.py b/coc/clans.py index 4148810f..3bbdf223 100644 --- a/coc/clans.py +++ b/coc/clans.py @@ -25,7 +25,7 @@ from .players import ClanMember -from .miscmodels import try_enum, ChatLanguage, Location, Label, WarLeague +from .miscmodels import try_enum, ChatLanguage, Location, Label, WarLeague, CapitalDistrict from .utils import get, cached_property, correct_tag from .abc import BaseClan @@ -129,6 +129,9 @@ class Clan(BaseClan): member_cls: :class:`coc.ClanMember` The type which the members found in :attr:`Clan.members` will be of. Ensure any overriding of this inherits from :class:`coc.ClanMember`. + capital_district_cls: :class:`coc.CapitalDistrict` + The type which the clan capital districts found in :attr:`Clan.capital_districts` will be of. + Ensure any overriding of this inherits from :class:`coc.CapitalDistrict`. war_league: :class:`coc.WarLeague` The clan's CWL league. """ @@ -149,24 +152,28 @@ class Clan(BaseClan): "member_count", "_labels", "_members", + "_districts", "_client", "label_cls", "member_cls", + "capital_district_cls", "war_league", "chat_language", "_cs_labels", "_cs_members", + "_cs_members_dict", + "_cs_capital_districts", "_iter_labels", "_iter_members", + "_iter_capital_districts" ) def __init__(self, *, data, client, **_): super().__init__(data=data, client=client) self.label_cls = Label self.member_cls = ClanMember - - self._members = None # type: typing.Optional[typing.Dict[str, ClanMember]] + self.capital_district_cls = CapitalDistrict self._from_data(data) @@ -200,6 +207,13 @@ def _from_data(self, data: dict) -> None: member_cls(data=mdata, client=self._client, clan=self) for mdata in data_get("memberList", []) ) + capital_district_cls = self.capital_district_cls + if data_get("clanCapital"): + self._iter_capital_districts = (capital_district_cls(data=cddata, client=self._client) for cddata in + data_get("clanCapital")["districts"]) + else: + self._iter_capital_districts = () + @cached_property("_cs_labels") def labels(self) -> typing.List[Label]: """List[:class:`Label`]: A :class:`List` of :class:`Label` that the clan has.""" @@ -208,8 +222,18 @@ def labels(self) -> typing.List[Label]: @cached_property("_cs_members") def members(self) -> typing.List[ClanMember]: """List[:class:`ClanMember`]: A list of members that belong to the clan.""" - dict_members = self._members = {m.tag: m for m in self._iter_members} - return list(dict_members.values()) + return list(self.members_dict.values()) + + @cached_property("_cs_members_dict") + def members_dict(self) -> typing.Dict[str, ClanMember]: + """Dict[str, :class:`ClanMember`]: A dict of members that belong to the clan.""" + return {m.tag: m for m in self._iter_members} + + + @cached_property("_cs_capital_districts") + def capital_districts(self) -> typing.List[CapitalDistrict]: + """List[:class:`CapitalDistrict`]: A :class:`List` of :class:`CapitalDistrict` that the clan has.""" + return list(self._iter_capital_districts) def get_member(self, tag: str) -> typing.Optional[ClanMember]: """Return a :class:`ClanMember` with the tag provided. Returns ``None`` if not found. @@ -226,11 +250,11 @@ def get_member(self, tag: str) -> typing.Optional[ClanMember]: The member who matches the tag provided: Optional[:class:`ClanMember`] """ tag = correct_tag(tag) - if not self._members: + if not self.members_dict: _ = self.members try: - return self._members[tag] + return self.members_dict[tag] except KeyError: return None diff --git a/coc/client.py b/coc/client.py index 7e561ce8..0e527c0e 100644 --- a/coc/client.py +++ b/coc/client.py @@ -21,7 +21,6 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ - import asyncio import logging @@ -258,8 +257,8 @@ async def login(self, email: str, password: str) -> None: self.http = http = self._create_client(email, password) await http.create_session(self.connector, self.timeout) await http.initialise_keys() - self._create_holders() + self._create_holders() LOG.debug("HTTP connection created. Client is ready for use.") def login_with_keys(self, *keys: str) -> None: @@ -278,13 +277,10 @@ def login_with_keys(self, *keys: str) -> None: LOG.debug("HTTP connection created. Client is ready for use.") - def close(self) -> None: - """Closes the HTTP connection - """ - LOG.info("Clash of Clans client logging out...") - self.dispatch("on_client_close") - self.loop.run_until_complete(self.http.close()) - self.loop.close() + async def close(self) -> None: + """Closes the HTTP connection from within a loop function such as + async def main()""" + await self.http.close() def dispatch(self, event_name: str, *args, **kwargs) -> None: """Dispatches an event listener matching the `event_name` parameter.""" diff --git a/coc/errors.py b/coc/errors.py index 072b6f01..6031b665 100644 --- a/coc/errors.py +++ b/coc/errors.py @@ -81,7 +81,7 @@ def __init__(self, response=None, data=None): self.reason = None self.message = response - fmt = "Unknown Error Occured: {0}" + fmt = "Error Occurred: {0}" super().__init__(fmt.format(self.message)) @@ -102,6 +102,8 @@ class InvalidCredentials(HTTPException): were passed. This is when your email/password pair is incorrect. Subclass of :exc:`HTTPException` """ + def __init__(self, response="Invalid Credentials"): + super().__init__(response=response) class Forbidden(HTTPException): diff --git a/coc/events.py b/coc/events.py index fd730b2b..247d8a66 100644 --- a/coc/events.py +++ b/coc/events.py @@ -41,7 +41,10 @@ class Event: - """Object that is created for an event. This contains runner functions, tags and type.""" + """ + Object that is created for an event. This contains runner functions, + tags and type. + """ __slots__ = ("runner", "callback", "tags", "type") @@ -719,13 +722,6 @@ def run_forever(self): except KeyboardInterrupt: self.close() - def close(self): - """Closes the client and all running tasks.""" - tasks = {t for t in asyncio.Task.all_tasks(loop=self.loop) if not t.done()} - for task in tasks: - task.cancel() - super().close() - def dispatch(self, event_name: str, *args, **kwargs): # pylint: disable=broad-except registered = self._listeners["client"].get(event_name) diff --git a/coc/events.pyi b/coc/events.pyi index 2409f044..9bd3e3f2 100644 --- a/coc/events.pyi +++ b/coc/events.pyi @@ -185,4 +185,3 @@ class EventsClient(Client): def add_events(self, *events: Callable) -> None: ... def remove_events(self, *events: Callable) -> None: ... def run_forever(self) -> None: ... - def close(self) -> None: ... diff --git a/coc/ext/discordlinks/__init__.py b/coc/ext/discordlinks/__init__.py index a97a2b7b..5dce07c2 100644 --- a/coc/ext/discordlinks/__init__.py +++ b/coc/ext/discordlinks/__init__.py @@ -44,7 +44,7 @@ def extract_expiry_from_jwt_token(token): return None -def login(username: str, password: str, loop: asyncio.AbstractEventLoop = None) -> "DiscordLinkClient": +async def login(username: str, password: str) -> "DiscordLinkClient": """Eases logging into the API client. For more information on this project, please join the discord server - @@ -66,9 +66,8 @@ def login(username: str, password: str, loop: asyncio.AbstractEventLoop = None) raise TypeError("username and password must both be a string") if not username or not password: raise ValueError("username or password must not be an empty string.") - if loop and not isinstance(loop, asyncio.AbstractEventLoop): - raise TypeError("loop must be of type asyncio.AbstractEventLoop, or None.") + loop = asyncio.get_running_loop() return DiscordLinkClient(username, password, loop) @@ -106,6 +105,10 @@ def __init__(self, username: str, password: str, loop: asyncio.AbstractEventLoop self.http_session = aiohttp.ClientSession(loop=self.loop) + async def close(self): + """Close the client session established""" + await self.http_session.close() + async def _request(self, method, url, *, token_request: bool = False, **kwargs): url = self.BASE_URL + url diff --git a/coc/http.py b/coc/http.py index fcfeae60..2ce229fd 100644 --- a/coc/http.py +++ b/coc/http.py @@ -415,64 +415,66 @@ def get_player_labels(self, **kwargs): async def initialise_keys(self): LOG.debug("Initialising keys from the developer site.") self.initialising_keys.clear() - session = aiohttp.ClientSession() - body = {"email": self.email, "password": self.password} - resp = await session.post("https://developer.clashofclans.com/api/login", json=body) - if resp.status == 403: - raise InvalidCredentials(resp) - - LOG.info("Successfully logged into the developer site.") - - resp_paylaod = await resp.json() - ip = json_loads(base64_b64decode(resp_paylaod["temporaryAPIToken"].split(".")[1] + "====").decode("utf-8"))["limits"][1]["cidrs"][0].split("/")[0] - - LOG.info("Found IP address to be %s", ip) - - resp = await session.post("https://developer.clashofclans.com/api/apikey/list") - keys = (await resp.json())["keys"] - self._keys.extend(key["key"] for key in keys if key["name"] == self.key_names and ip in key["cidrRanges"]) - - LOG.info("Retrieved %s valid keys from the developer site.", len(self._keys)) - - if len(self._keys) < self.key_count: - for key in (k for k in keys if k["name"] == self.key_names and ip not in k["cidrRanges"]): - LOG.info( - "Deleting key with the name %s and IP %s (not matching our current IP address).", - self.key_names, key["cidrRanges"], + # Use context manager to automatically clean up after ourselves + async with aiohttp.ClientSession() as session: + body = {"email": self.email, "password": self.password} + resp = await session.post("https://developer.clashofclans.com/api/login", json=body) + if resp.status == 403: + LOG.error("Invalid credentials used when attempting to log in") + await self.close() + raise InvalidCredentials() + + LOG.info("Successfully logged into the developer site.") + + resp_paylaod = await resp.json() + ip = json_loads(base64_b64decode(resp_paylaod["temporaryAPIToken"].split(".")[1] + "====").decode("utf-8"))["limits"][1]["cidrs"][0].split("/")[0] + + LOG.info("Found IP address to be %s", ip) + + resp = await session.post("https://developer.clashofclans.com/api/apikey/list") + keys = (await resp.json())["keys"] + self._keys.extend(key["key"] for key in keys if key["name"] == self.key_names and ip in key["cidrRanges"]) + + LOG.info("Retrieved %s valid keys from the developer site.", len(self._keys)) + + if len(self._keys) < self.key_count: + for key in (k for k in keys if k["name"] == self.key_names and ip not in k["cidrRanges"]): + LOG.info( + "Deleting key with the name %s and IP %s (not matching our current IP address).", + self.key_names, key["cidrRanges"], + ) + await session.post("https://developer.clashofclans.com/api/apikey/revoke", json={"id": key["id"]}) + + while len(self._keys) < self.key_count and len(keys) < KEY_MAXIMUM: + data = { + "name" : self.key_names, + "description": "Created on {}".format(datetime.now().strftime("%c")), + "cidrRanges" : [ip], + "scopes" : [self.key_scopes], + } + + LOG.info("Creating key with data %s.", str(data)) + + resp = await session.post("https://developer.clashofclans.com/api/apikey/create", json=data) + key = await resp.json() + self._keys.append(key["key"]["key"]) + + if len(keys) == 10 and len(self._keys) < self.key_count: + LOG.critical("%s keys were requested to be used, but a maximum of %s could be " + "found/made on the developer site, as it has a maximum of 10 keys per account. " + "Please delete some keys or lower your `key_count` level." + "I will use %s keys for the life of this client.", + self.key_count, len(self._keys), len(self._keys)) + + if len(self._keys) == 0: + await self.close() + raise RuntimeError( + "There are {} API keys already created and none match a key_name of '{}'." + "Please specify a key_name kwarg, or go to 'https://developer.clashofclans.com' to delete " + "unused keys.".format(len(keys), self.key_names) ) - await session.post("https://developer.clashofclans.com/api/apikey/revoke", json={"id": key["id"]}) - - while len(self._keys) < self.key_count and len(keys) < KEY_MAXIMUM: - data = { - "name" : self.key_names, - "description": "Created on {}".format(datetime.now().strftime("%c")), - "cidrRanges" : [ip], - "scopes" : [self.key_scopes], - } - - LOG.info("Creating key with data %s.", str(data)) - - resp = await session.post("https://developer.clashofclans.com/api/apikey/create", json=data) - key = await resp.json() - self._keys.append(key["key"]["key"]) - - if len(keys) == 10 and len(self._keys) < self.key_count: - LOG.critical("%s keys were requested to be used, but a maximum of %s could be " - "found/made on the developer site, as it has a maximum of 10 keys per account. " - "Please delete some keys or lower your `key_count` level." - "I will use %s keys for the life of this client.", - self.key_count, len(self._keys), len(self._keys)) - - if len(self._keys) == 0: - await self.close() - raise RuntimeError( - "There are {} API keys already created and none match a key_name of '{}'." - "Please specify a key_name kwarg, or go to 'https://developer.clashofclans.com' to delete " - "unused keys.".format(len(keys), self.key_names) - ) - - await session.close() + self.keys = cycle(self._keys) self.initialising_keys.set() LOG.info("Successfully initialised keys for use.") diff --git a/coc/miscmodels.py b/coc/miscmodels.py index 483aabf8..77fff952 100644 --- a/coc/miscmodels.py +++ b/coc/miscmodels.py @@ -580,6 +580,40 @@ def __init__(self, *, data, client): self.badge = try_enum(Icon, data=data.get("iconUrls"), client=self._client) +class CapitalDistrict: + """Represents a Clan Capital District. + + Attributes + ----------- + id: + :class:`int`: The district's unique ID as given by the API. + name: + :class:`str`: The district's name. + hall_level: + :class:`int`: The district's hall level + """ + + __slots__ = ("id", "name", "hall_level") + + def __str__(self): + return self.name + + def __repr__(self): + attrs = [("id", self.id), ("name", self.name)] + return "<%s %s>" % (self.__class__.__name__, " ".join("%s=%r" % t for t in attrs),) + + def __eq__(self, other): + return isinstance(other, self.__class__) and \ + self.id == other.id and \ + self.hall_level == other.hall_level + + def __init__(self, *, data, client): + # pylint: disable=invalid-name + self.id: int = data.get("id") + self.name: str = data.get("name") + self.hall_level: int = data.get("districtHallLevel") + + class WarLeague: """Represents a clan's CWL league. Attributes diff --git a/coc/players.py b/coc/players.py index 2ccfaf6f..6faf9c16 100644 --- a/coc/players.py +++ b/coc/players.py @@ -208,6 +208,8 @@ class Player(ClanMember): The player's best versus trophy count. versus_attack_wins: :class:`int` The number of versus attacks the player has won + clan_capital_contributions: :class:`int` + The player's total contribution to clan capitals legend_statistics: Optional[:class:`LegendStatistics`] The player's legend statistics, or ``None`` if they have never been in the legend league. war_opted_in: Optional[:class:`bool`] @@ -226,6 +228,7 @@ class Player(ClanMember): "builder_hall", "best_versus_trophies", "versus_attack_wins", + "clan_capital_contributions", "legend_statistics", "war_opted_in", "_achievements", @@ -309,6 +312,7 @@ def _from_data(self, data: dict) -> None: self.builder_hall: int = data_get("builderHallLevel", 0) self.best_versus_trophies: int = data_get("bestVersusTrophies") self.versus_attack_wins: int = data_get("versusBattleWins") + self.clan_capital_contributions: int = data_get("clanCapitalContributions") self.legend_statistics = try_enum(LegendStatistics, data=data_get("legendStatistics")) try: diff --git a/examples/discord_bot.py b/examples/discord_bot.py index 774d20e5..a230aee4 100644 --- a/examples/discord_bot.py +++ b/examples/discord_bot.py @@ -1,29 +1,22 @@ -# this example assumes you have discord.py > v1.5 +# this example assumes you have discord.py > v2.0 # installed via `python -m pip install -U discord.py` # for more info on using discord.py, see the docs at: # https://discordpy.readthedocs.io/en/latest - +import asyncio import logging import os import traceback -import coc import discord +from discord.ext import commands +import coc from coc import utils -from discord.ext import commands INFO_CHANNEL_ID = 761848043242127370 # some discord channel ID clan_tags = ["#20090C9PR", "#202GG92Q", "#20C8G0RPL"] bot = commands.Bot(command_prefix="?", intents=discord.Intents.all()) -coc_client = coc.login( - os.environ["DEV_SITE_EMAIL"], - os.environ["DEV_SITE_PASSWORD"], - key_names="coc.py tests", - client=coc.EventsClient, -) -logging.basicConfig(level=logging.ERROR) @bot.event @@ -31,7 +24,6 @@ async def on_ready(): print("Logged in!!") -@coc_client.event @coc.ClanEvents.member_join(tags=clan_tags) async def on_clan_member_join(member, clan): await bot.get_channel(INFO_CHANNEL_ID).send( @@ -40,7 +32,6 @@ async def on_clan_member_join(member, clan): ) -@coc_client.event @coc.ClanEvents.member_name(tags=clan_tags) async def member_name_change(old_player, new_player): await bot.get_channel(INFO_CHANNEL_ID).send( @@ -49,12 +40,12 @@ async def member_name_change(old_player, new_player): ) -@coc_client.event @coc.ClientEvents.event_error() async def on_event_error(exception): if isinstance(exception, coc.PrivateWarLog): return # lets ignore private war log errors - print("Uh oh! Something went wrong in coc.py events... printing traceback for you.") + print( + "Uh oh! Something went wrong in coc.py events... printing traceback for you.") traceback.print_exc() @@ -65,7 +56,7 @@ async def player_heroes(ctx, player_tag): return try: - player = await coc_client.get_player(player_tag) + player = await ctx.bot.coc_client.get_player(player_tag) except coc.NotFound: await ctx.send("This player doesn't exist!") return @@ -80,7 +71,7 @@ async def player_heroes(ctx, player_tag): @bot.command() async def parse_army(ctx, army_link: str): - troops, spells = coc_client.parse_army_link(army_link) + troops, spells = ctx.bot.coc_client.parse_army_link(army_link) print(troops, spells) parsed_link_output = '' if troops or spells: # checking if troops or spells is present in link @@ -96,9 +87,10 @@ async def parse_army(ctx, army_link: str): parsed_link_output += "Invalid Link!" await ctx.send(parsed_link_output) + @bot.command() async def create_army(ctx): - link = coc_client.create_army_link( + link = ctx.bot.coc_client.create_army_link( barbarian=10, archer=20, hog_rider=30, @@ -108,6 +100,42 @@ async def create_army(ctx): ) await ctx.send(link) + +@bot.command() +async def member_stat(ctx, player_tag): + if not utils.is_valid_tag(player_tag): + await ctx.send("You didn't give me a proper tag!") + return + + try: + player = await ctx.bot.coc_client.get_player(player_tag) + except coc.NotFound: + await ctx.send("This clan doesn't exist!") + return + + frame = '' + if player.town_hall > 11: + frame += f"`{'TH Weapon LvL:':<15}` `{player.town_hall_weapon:<15}`\n" + role = player.role if player.role else 'None' + clan = player.clan.name if player.clan else 'None' + frame += ( + f"`{'Role:':<15}` `{role:<15}`\n" + f"`{'Player Tag:':<15}` `{player.tag:<15}`\n" + f"`{'Current Clan:':<15}` `{clan:<15.15}`\n" + f"`{'League:':<15}` `{player.league.name:<15.15}`\n" + f"`{'Trophies:':<15}` `{player.trophies:<15}`\n" + f"`{'Best Trophies:':<15}` `{player.best_trophies:<15}`\n" + f"`{'War Stars:':<15}` `{player.war_stars:<15}`\n" + f"`{'Attack Wins:':<15}` `{player.attack_wins:<15}`\n" + f"`{'Defense Wins:':<15}` `{player.defense_wins:<15}`\n" + f"`{'Castle Contrib':<15}` `{player.clan_capital_contributions:<15}`\n" + ) + e = discord.Embed(colour=discord.Colour.green(), + description=frame) + e.set_thumbnail(url=player.clan.badge.url) + await ctx.send(embed=e) + + @bot.command() async def clan_info(ctx, clan_tag): if not utils.is_valid_tag(clan_tag): @@ -115,7 +143,7 @@ async def clan_info(ctx, clan_tag): return try: - clan = await coc_client.get_clan(clan_tag) + clan = await ctx.bot.coc_client.get_clan(clan_tag) except coc.NotFound: await ctx.send("This clan doesn't exist!") return @@ -126,9 +154,11 @@ async def clan_info(ctx, clan_tag): log = "Public" e = discord.Embed(colour=discord.Colour.green()) + e.set_thumbnail(url=clan.badge.url) e.add_field(name="Clan Name", - value=f"{clan.name}({clan.tag})\n[Open in game]({clan.share_link})", inline=False) + value=f"{clan.name}({clan.tag})\n[Open in game]({clan.share_link})", + inline=False) e.add_field(name="Clan Level", value=clan.level, inline=False) e.add_field(name="Description", value=clan.description, inline=False) e.add_field(name="Leader", value=clan.get_member_by( @@ -154,7 +184,15 @@ async def clan_info(ctx, clan_tag): value=f"Won - {clan.war_wins}\nLost - {clan.war_losses}\n Draw - {clan.war_ties}", inline=False ) - await ctx.send(embed=e) + + frame = "" + for district in clan.capital_districts: + frame += (f"`{f'{district.name}:':<20}` `{district.hall_level:<15}`\n") + + e2 = discord.Embed(colour=discord.Colour.green(), description=frame, + title="Capital Distracts") + + await ctx.send(embeds=[e, e2]) @bot.command() @@ -164,7 +202,7 @@ async def clan_member(ctx, clan_tag): return try: - clan = await coc_client.get_clan(clan_tag) + clan = await ctx.bot.coc_client.get_clan(clan_tag) except coc.NotFound: await ctx.send("This clan does not exist!") return @@ -172,6 +210,7 @@ async def clan_member(ctx, clan_tag): member = "" for i, a in enumerate(clan.members, start=1): member += f"`{i}.` {a.name}\n" + embed = discord.Embed(colour=discord.Colour.red(), title=f"Members of {clan.name}", description=member) embed.set_thumbnail(url=clan.badge.url) @@ -188,7 +227,7 @@ async def current_war_status(ctx, clan_tag): e = discord.Embed(colour=discord.Colour.blue()) try: - war = await coc_client.get_current_war(clan_tag) + war = await ctx.bot.coc_client.get_current_war(clan_tag) except coc.PrivateWarLog: return await ctx.send("Clan has a private war log!") @@ -204,34 +243,40 @@ async def current_war_status(ctx, clan_tag): e.add_field(name=war.clan.name, value=war.clan.tag) e.add_field( - name="Opponent:", value=f"{war.opponent.name}\n" f"{war.opponent.tag}", inline=False) + name="Opponent:", + value=f"{war.opponent.name}\n" f"{war.opponent.tag}", inline=False) e.add_field(name="War End Time:", - value=f"{hours} hours {minutes} minutes {seconds} seconds", inline=False) + value=f"{hours} hours {minutes} minutes {seconds} seconds", + inline=False) await ctx.send(embed=e) -async def run_tests_and_quit(): - # ignore this; it is purely for the benefit of being able to run the examples as tests. - import sys - - class Msg: - _state = bot._connection +async def main(): + logging.basicConfig(level=logging.ERROR) - await bot.wait_until_ready() - ctx = commands.Context(prefix=None, message=Msg, bot=bot) + coc_client = coc.Client() - async def _mock_get_channel(): - return bot.get_channel(INFO_CHANNEL_ID) - ctx._get_channel = _mock_get_channel + # Attempt to log into CoC API using your credentials. + try: + await coc_client.login(os.environ.get("DEV_SITE_EMAIL"), + os.environ.get("DEV_SITE_PASSWORD")) + except coc.InvalidCredentials as error: + exit(error) - await ctx.invoke(player_heroes, "#JY9J2Y99") - for command in (clan_info, clan_member, current_war_status): - await ctx.invoke(command, clan_tags[0]) + # Add the client session to the bot + bot.coc_client = coc_client - sys.exit(0) + try: + # Run the bot + await bot.start(os.environ.get("DISCORD_BOT_TOKEN")) + finally: + # When done, do not forget to clean up after yourself! + await coc_client.close() -if os.environ.get("RUNNING_TESTS"): - bot.loop.create_task(run_tests_and_quit()) -bot.run(os.environ["DISCORD_BOT_TOKEN"]) +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass diff --git a/examples/discord_bot_with_cogs.py b/examples/discord_bot_with_cogs.py new file mode 100644 index 00000000..2e387601 --- /dev/null +++ b/examples/discord_bot_with_cogs.py @@ -0,0 +1,257 @@ +import asyncio +import os + +import aiohttp +import discord +from discord.ext import commands + +import coc + +# Clan tags we want to listen for +clan_tags = ["#P222C9Y", "#9VPR98RG", "#9G2QU8YG", "#80Y8L0QY", "#2Y28CGP8"] + + +class CoCBot(commands.Bot): + """Inherit from commands.Bot so that you can easily configure the bots + behaviour""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.coc_client: coc.EventsClient = kwargs.get("coc_client") + + async def on_ready(self): + print("Logged in!!") + + async def setup_hook(self) -> None: + # Add command cog class to start listening for those commands + await self.add_cog(CocCommands(self)) + + # Register your CoC Events + self.coc_client.add_clan_updates(*clan_tags) + + # Register all the players you want to track + async for clan in self.coc_client.get_clans(clan_tags): + self.coc_client.add_player_updates( + *[member.tag for member in clan.members]) + + # Register the callback functions for the evens you are listening for + self.coc_client.add_events( + on_clan_member_donation, + on_clan_member_donation_receive, + on_clan_trophy_change, + ) + + +class CocCommands(commands.Cog): + """This class holds the commands supported by the bot""" + def __init__(self, bot: CoCBot): + self.bot = bot + + @commands.command() + async def member_stat(self, ctx, player_tag): + if not coc.utils.is_valid_tag(player_tag): + await ctx.send("You didn't give me a proper tag!") + return + + try: + player = await self.bot.coc_client.get_player(player_tag) + except coc.NotFound: + await ctx("This clan doesn't exist!") + return + + frame = "" + if player.town_hall > 11: + frame += f"`{'TH Weapon LvL:':<15}` `{player.town_hall_weapon:<15}`\n" + role = player.role if player.role else 'None' + clan = player.clan.name if player.clan else 'None' + frame += ( + f"`{'Role:':<15}` `{role:<15}`\n" + f"`{'Player Tag:':<15}` `{player.tag:<15}`\n" + f"`{'Current Clan:':<15}` `{clan:<15.15}`\n" + f"`{'League:':<15}` `{player.league.name:<15.15}`\n" + f"`{'Trophies:':<15}` `{player.trophies:<15}`\n" + f"`{'Best Trophies:':<15}` `{player.best_trophies:<15}`\n" + f"`{'War Stars:':<15}` `{player.war_stars:<15}`\n" + f"`{'Attack Wins:':<15}` `{player.attack_wins:<15}`\n" + f"`{'Defense Wins:':<15}` `{player.defense_wins:<15}`\n" + f"`{'Castle Contrib':<15}` `{player.clan_capital_contributions:<15}`\n" + ) + e = discord.Embed(colour=discord.Colour.green(), + description=frame) + e.set_thumbnail(url=player.clan.badge.url) + await ctx.send(embed=e) + + @commands.command() + async def clan_info(self, ctx, clan_tag): + if not coc.utils.is_valid_tag(clan_tag): + await ctx.send("You didn't give me a proper tag!") + return + + try: + clan = await self.bot.coc_client.get_clan(clan_tag) + except coc.NotFound: + await ctx.send("This clan doesn't exist!") + return + + if clan.public_war_log is False: + log = "Private" + else: + log = "Public" + + e = discord.Embed(colour=discord.Colour.green()) + e.set_thumbnail(url=clan.badge.url) + + e.add_field(name="Clan Name", + value=f"{clan.name}({clan.tag})\n[Open in game]({clan.share_link})", + inline=False) + + e.add_field(name="Clan Level", + value=clan.level, + inline=False) + + e.add_field(name="Description", + value=clan.description, + inline=False) + + e.add_field(name="Leader", + value=clan.get_member_by(role=coc.Role.leader), + inline=False) + + e.add_field(name="Clan Type", + value=clan.type, + inline=False) + + e.add_field(name="Location", + value=clan.location, + inline=False) + + e.add_field(name="Total Clan Trophies", + value=clan.points, + inline=False) + + e.add_field(name="Total Clan Versus Trophies", + value=clan.versus_points, + inline=False) + + e.add_field(name="WarLog Type", + value=log, + inline=False) + + e.add_field(name="Required Trophies", + value=clan.required_trophies, + inline=False) + + e.add_field(name="War Win Streak", + value=clan.war_win_streak, + inline=False) + + e.add_field(name="War Frequency", + value=clan.war_frequency, + inline=False) + + e.add_field(name="Clan War League Rank", + value=clan.war_league, + inline=False) + + e.add_field(name="Clan Labels", + value="\n".join(label.name for label in clan.labels), + inline=False) + + e.add_field(name="Member Count", + value=f"{clan.member_count}/50", + inline=False) + + e.add_field( + name="Clan Record", + value=f"Won - {clan.war_wins}\nLost - {clan.war_losses}\n " + f"Draw - {clan.war_ties}", + inline=False + ) + + frame = "" + for district in clan.capital_districts: + frame += (f"`{f'{district.name}:':<20}` " + f"`{district.hall_level:<15}`\n") + + e2 = discord.Embed(colour=discord.Colour.green(), description=frame, + title="Capital Distracts") + + await ctx.send(embeds=[e, e2]) + + +async def send_via_webhook(msg: str) -> None: + """Webhooks allow you to send messages to discord without having to + have a bot session. Keep in mind that you need to create the webhook + URL from Discord""" + webhook_url = os.environ.get("DISCORD_WEBHOOK") + if webhook_url is None: + print("No URL webhook found") + return + + async with aiohttp.ClientSession() as session: + webhook = discord.Webhook.from_url(os.environ.get("DISCORD_WEBHOOK"), + session=session) + await webhook.send(msg) + + +# The following decorators are listeners. Keep in mind that they must first +# be registered before they work. The registration process happens during +# the bots `setup_hook` +@coc.ClanEvents.member_donations() +async def on_clan_member_donation(old_member, new_member): + final_donated_troops = new_member.donations - old_member.donations + msg =(f"{new_member} of {new_member.clan} just donated " + f"{final_donated_troops} troops.") + print(msg) + await send_via_webhook(msg) + + +@coc.ClanEvents.member_received() +async def on_clan_member_donation_receive(old_member, new_member): + final_received_troops = new_member.received - old_member.received + msg = (f"{new_member} of {new_member.clan} just received " + f"{final_received_troops} troops.") + print(msg) + await send_via_webhook(msg) + + +@coc.ClanEvents.points() +async def on_clan_trophy_change(old_clan, new_clan): + msg =(f"{new_clan.name} total trophies changed from {old_clan.points} " + f"to {new_clan.points}") + print(msg) + await send_via_webhook(msg) + + +async def main(): + coc_client = coc.EventsClient() + + # Attempt to log into CoC API using your credentials. To enable events, + # you must use the coc.EventsClient class + try: + await coc_client.login(os.environ.get("DEV_SITE_EMAIL"), + os.environ.get("DEV_SITE_PASSWORD")) + except coc.InvalidCredentials as error: + exit(error) + + # Instantiate your custom bot class that inherits from the discord bot + # notice that we added coc_client into the bot. This will give us access + # to coc_client from all our discord bot commands + bot = CoCBot(command_prefix="?", + intents=discord.Intents.all(), + coc_client=coc_client) + + try: + # Run the bot + await bot.start(os.environ.get("DISCORD_BOT_TOKEN")) + finally: + # When done, do not forget to clean up after yourself! + await coc_client.close() + + +if __name__ == "__main__": + try: + # It is recommended to use the high level asyncio.run wrapper to + # properly cleanup after the loops are done + asyncio.run(main()) + except KeyboardInterrupt: + pass diff --git a/examples/discord_links.py b/examples/discord_links.py index a7d23afe..3dedd827 100644 --- a/examples/discord_links.py +++ b/examples/discord_links.py @@ -3,46 +3,50 @@ from coc.ext import discordlinks -client = discordlinks.login(os.environ["LINKS_API_USERNAME"], os.environ["LINKS_API_PASSWORD"]) - async def main(): + client = await discordlinks.login(os.environ["LINKS_API_USERNAME"], + os.environ["LINKS_API_PASSWORD"]) + player_tag = "#JY9J2Y99" discord_id = 230214242618441728 # add a link await client.add_link(player_tag, discord_id) - print("Player Tag {} is now linked to discord id {}".format(player_tag, discord_id)) + print(f"Player Tag {player_tag} is now linked to discord id {discord_id}") # get a link by tag discord_id = await client.get_link(player_tag) - print("Player Tag {} is linked to discord id {}".format(player_tag, discord_id)) + print(f"Player Tag {player_tag} is linked to discord id {discord_id}") # update a link new_discord_id = 230214242618441728 await client.delete_link(player_tag) await client.add_link(player_tag, new_discord_id) - print("Link for player tag {} has been updated to have discord id {}".format(player_tag, new_discord_id)) + print(f"Link for player tag {player_tag} has been updated to " + f"have discord id {new_discord_id}") # delete a link await client.delete_link(player_tag) - print("Link for player tag {} has been removed from the database.".format(player_tag)) + print(f"Link for player tag {player_tag} has been removed from " + f"the database.") # batch get links by tag player_tags = ["#JY9J2Y99", "#2GV0QY8G8", "#PP9L22C8", "#2LPC9J8L"] links = await client.get_links(*player_tags) for tag, discord_id in links: if discord_id is None: - print("Player tag {} doesn't have any links.".format(tag)) + print(f"Player tag {tag} doesn't have any links.") else: - print("Player tag {} is linked to discord id {}".format(tag, discord_id)) + print(f"Player tag {tag} is linked to discord id {discord_id}") # batch get links by id discord_ids = [246286410946969610, 230214242618441728, 267057699856842753] links = await client.get_many_linked_players(*discord_ids) for tag, discord_id in links: - print("Discord ID {} is linked to {}".format(tag, discord_id)) + print(f"Discord ID {tag} is linked to {discord_id}") + await client.close() -loop = asyncio.get_event_loop() -loop.run_until_complete(main()) +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/events.py b/examples/events.py deleted file mode 100644 index 7ff64ca4..00000000 --- a/examples/events.py +++ /dev/null @@ -1,139 +0,0 @@ -import logging -import os - -import coc - -from coc import utils - -client = coc.login( - os.environ["DEV_SITE_EMAIL"], - os.environ["DEV_SITE_PASSWORD"], - key_names="coc.py tests", - client=coc.EventsClient, -) -logging.basicConfig(level=logging.INFO) -log = logging.getLogger() - - -clan_tags = ["#P222C9Y", "#9VPR98RG", "#9G2QU8YG", "#80Y8L0QY", "#9UJ8JRUP"] -player_tags = ["#YQ2QYLGJ", "#QYJCVGL", "#2LURLC9V", "#QCQR298V", "#82CVC2V8", "#29U09V8J", "#8GYGL22P"] - - -"""Clan Events""" - - -@client.event # Pro Tip : if you don't have @client.event then your events won't run! Don't forget it! -@coc.ClanEvents.member_donations(tags=clan_tags) -async def on_clan_member_donation(old_member, new_member): - final_donated_troops = new_member.donations - old_member.donations - log.info(f"{new_member} of {new_member.clan} just donated {final_donated_troops} troops.") - - -@client.event -@coc.ClanEvents.member_received(tags=clan_tags) -async def on_clan_member_donation_receive(old_member, new_member): - final_received_troops = new_member.received - old_member.received - log.info(f"{new_member} of {new_member.clan} just received {final_received_troops} troops.") - - -@client.event -@coc.ClanEvents.member_join(tags=clan_tags) -async def on_clan_member_join(member, clan): - log.info(f"{member.name} has joined {clan.name}") - - -@client.event -@coc.ClanEvents.member_leave(tags=clan_tags) -async def on_clan_member_leave(member, clan): - log.info(f"{member.name} has left {clan.name}") - - -@client.event -@coc.ClanEvents.points(tags=clan_tags) -async def on_clan_trophy_change(old_clan, new_clan): - log.info(f"{new_clan.name} total trophies changed from {old_clan.points} to {new_clan.points}") - - -@client.event -@coc.ClanEvents.member_versus_trophies(tags=clan_tags) -async def clan_member_versus_trophies_changed(old_member, new_member): - log.info(f"{new_member} versus trophies changed from {old_member.versus_trophies} to {new_member.versus_trophies}") - - -"""War Events""" - - -@client.event -@coc.WarEvents.war_attack(tags=clan_tags) -async def current_war_stats(attack, war): - log.info(f"Attack number {attack.order}\n({attack.attacker.map_position}).{attack.attacker} of {attack.attacker.clan} " - f"attacked ({attack.defender.map_position}).{attack.defender} of {attack.defender.clan}") - - -"""Player Events""" - - -@client.event -@coc.PlayerEvents.donations(tags=player_tags) -async def on_player_donation(old_player, new_player): - final_donated_troops = new_player.donations - old_player.donations - log.info(f"{new_player} of {new_player.clan} just donated {final_donated_troops} troops.") - - -@client.event -@coc.PlayerEvents.received(tags=player_tags) -async def on_player_donation_receive(old_player, new_player): - final_received_troops = new_player.received - old_player.received - log.info(f"{new_player} of {new_player.clan} just received {final_received_troops} troops.") - - -@client.event -@coc.PlayerEvents.trophies(tags=player_tags) -async def on_player_trophy_change(old_player, new_player): - log.info(f"{new_player} trophies changed from {old_player.trophies} to {new_player.trophies}") - - -@client.event -@coc.PlayerEvents.versus_trophies(tags=player_tags) -async def on_player_versus_trophy_change(old_player, new_player): - log.info(f"{new_player} versus trophies changed from {old_player.trophies} to {new_player.trophies}") - - -"""Client Events""" - - -@client.event -@coc.ClientEvents.maintenance_start() -async def on_maintenance(): - log.info("Maintenace Started") - - -@client.event -@coc.ClientEvents.maintenance_completion() -async def on_maintenance_completion(time_started): - log.info("Maintenace Ended; started at %s", time_started) - - -@client.event -@coc.ClientEvents.new_season_start() -async def season_started(): - log.info("New season started, and will finish at %s", str(utils.get_season_end())) - - -async def add_clan_players(): - async for clan in client.get_clans(clan_tags): - client.add_player_updates(*[member.tag for member in clan.members]) - -if os.environ.get("RUNNING_TESTS"): - # ignore this; it's just for running github action tests. - import sys - - class Handler(logging.Handler): - def emit(self, record) -> None: - sys.exit(0) - log.addHandler(Handler()) - # we don't wanna wait forever for an event, so if it sets up OK lets call it quits. - client.loop.call_later(20.0, sys.exit, 0) - -client.loop.run_until_complete(add_clan_players()) -client.loop.run_forever() diff --git a/examples/events_example.py b/examples/events_example.py new file mode 100644 index 00000000..963d900c --- /dev/null +++ b/examples/events_example.py @@ -0,0 +1,175 @@ +import asyncio +import logging +import os + +import coc +from coc import utils + +clan_tags = ["#P222C9Y", "#9VPR98RG", "#9G2QU8YG", "#80Y8L0QY", "#2Y28CGP8"] + +"""Clan Events""" + + +@coc.ClanEvents.member_donations() +async def on_clan_member_donation(old_member, new_member): + final_donated_troops = new_member.donations - old_member.donations + log.info( + f"{new_member} of {new_member.clan} just donated {final_donated_troops} troops.") + + +@coc.ClanEvents.member_received() +async def on_clan_member_donation_receive(old_member, new_member): + final_received_troops = new_member.received - old_member.received + log.info( + f"{new_member} of {new_member.clan} just received {final_received_troops} troops.") + + +@coc.ClanEvents.member_join() +async def on_clan_member_join(member, clan): + log.info(f"{member.name} has joined {clan.name}") + + +@coc.ClanEvents.member_leave() +async def on_clan_member_leave(member, clan): + log.info(f"{member.name} has left {clan.name}") + + +@coc.ClanEvents.points() +async def on_clan_trophy_change(old_clan, new_clan): + log.info( + f"{new_clan.name} total trophies changed from {old_clan.points} to {new_clan.points}") + + +@coc.ClanEvents.member_versus_trophies() +async def clan_member_versus_trophies_changed(old_member, new_member): + log.info( + f"{new_member} versus trophies changed from {old_member.versus_trophies} to {new_member.versus_trophies}") + + +"""War Events""" + + +@coc.WarEvents.war_attack() +async def current_war_stats(attack, war): + log.info( + f"Attack number {attack.order}\n({attack.attacker.map_position}).{attack.attacker} of {attack.attacker.clan} " + f"attacked ({attack.defender.map_position}).{attack.defender} of {attack.defender.clan}") + + +"""Player Events""" + + +@coc.PlayerEvents.donations() +async def on_player_donation(old_player, new_player): + final_donated_troops = new_player.donations - old_player.donations + log.info( + f"{new_player} of {new_player.clan} just donated {final_donated_troops} troops.") + + +@coc.PlayerEvents.received() +async def on_player_donation_receive(old_player, new_player): + final_received_troops = new_player.received - old_player.received + log.info( + f"{new_player} of {new_player.clan} just received {final_received_troops} troops.") + + +@coc.PlayerEvents.trophies() +async def on_player_trophy_change(old_player, new_player): + log.info( + f"{new_player} trophies changed from {old_player.trophies} to {new_player.trophies}") + + +@coc.PlayerEvents.versus_trophies() +async def on_player_versus_trophy_change(old_player, new_player): + log.info( + f"{new_player} versus trophies changed from {old_player.trophies} to {new_player.trophies}") + + +"""Client Events""" + + +@coc.ClientEvents.maintenance_start() +async def on_maintenance(): + log.info("Maintenace Started") + + +@coc.ClientEvents.maintenance_completion() +async def on_maintenance_completion(time_started): + log.info("Maintenace Ended; started at %s", time_started) + + +@coc.ClientEvents.new_season_start() +async def season_started(): + log.info("New season started, and will finish at %s", + str(utils.get_season_end())) + + +async def main() -> None: + coc_client = coc.EventsClient() + + # Attempt to log into CoC API using your credentials. You must use the + # coc.EventsClient to enable event listening + try: + await coc_client.login(os.environ.get("DEV_SITE_EMAIL"), + os.environ.get("DEV_SITE_PASSWORD")) + except coc.InvalidCredentials as error: + exit(error) + + # Register all the clans you want to track + coc_client.add_clan_updates(*clan_tags) + + # Register all the players you want to track + async for clan in coc_client.get_clans(clan_tags): + coc_client.add_player_updates(*[member.tag for member in clan.members]) + + # Register all the callback functions that are triggered when a + # event if fired. + coc_client.add_events( + on_clan_member_donation, + on_clan_member_donation_receive, + on_clan_member_join, + on_clan_member_leave, + on_clan_trophy_change, + clan_member_versus_trophies_changed, + current_war_stats, + on_player_donation, + on_player_donation_receive, + on_player_trophy_change, + on_player_versus_trophy_change, + on_maintenance, + on_maintenance_completion, + season_started + ) + + if os.environ.get("RUNNING_TESTS"): + # ignore this; it's just for running github action tests. + import sys + + class Handler(logging.Handler): + def emit(self, record) -> None: + sys.exit(0) + + log.addHandler(Handler()) + # we don't wanna wait forever for an event, so if + # it sets up OK lets call it quits. + await asyncio.sleep(20) + _loop = asyncio.get_event_loop() + _loop.stop() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + log = logging.getLogger() + + # Unlike the other examples that use `asyncio.run()`, in order to run + # events forever you must set the event loop to run forever so we will use + # the lower level function calls to handle this. + loop = asyncio.get_event_loop() + + try: + # Using the loop context, run the main function then set the loop + # to run forever so that it continuously monitors for events + loop.run_until_complete(main()) + loop.run_forever() + except KeyboardInterrupt: + pass diff --git a/examples/war_logs.py b/examples/war_logs.py index 9e8c9846..72eae883 100644 --- a/examples/war_logs.py +++ b/examples/war_logs.py @@ -3,14 +3,12 @@ import coc -# email and password is your login credentials used at https://developer.clashofclans.com -client = coc.login(os.environ["DEV_SITE_EMAIL"], os.environ["DEV_SITE_PASSWORD"], key_names="coc.py tests") - -async def get_warlog_for_clans(clan_tags: list): +async def get_warlog_for_clans(client: coc.Client, clan_tags: list): war_logs = {} for tag in clan_tags: - # if we're not allowed to view warlog (private in game), this will raise an exception + # if we're not allowed to view warlog (private in game), + # this will raise an exception try: warlog = await client.get_warlog(tag) except coc.PrivateWarLog: @@ -18,33 +16,44 @@ async def get_warlog_for_clans(clan_tags: list): war_logs[tag] = warlog - # return a dict of list of war logs: {'tag': [list_of_warlog_objects]} return war_logs -async def get_clan_tags_names(name: str, limit: int): +async def get_clan_tags_names(client: coc.Client, name: str, limit: int): clans = await client.search_clans(name=name, limit=limit) - # return a list of tuples of name/tag pair ie. [(name, tag), (another name, another tag)] + # return a list of tuples of name/tag pair ie. + # [(name, tag), (another name, another tag)] return [(n.name, n.tag) for n in clans] -async def get_warlog_opponents_from_clan_name(name: str, no_of_clans: int): - clan_tags_names = await get_clan_tags_names(name, no_of_clans) +async def get_warlog_opponents_from_clan_name(client: coc.Client, name: str, no_of_clans: int): + clan_tags_names = await get_clan_tags_names(client, name, no_of_clans) # search for war logs with clan tags found - war_logs = await get_warlog_for_clans([n[1] for n in clan_tags_names]) + war_logs = await get_warlog_for_clans(client, [n[1] for n in clan_tags_names]) for name, tag in clan_tags_names: # iterate over the wars for war in war_logs[tag]: - # if it is a league war we will error below because it does not return a WarLog object, - # and thus no opponent + # if it is a league war we will error below because it does not + # return a WarLog object, and thus no opponent if war.is_league_entry: print("League War Season - No opponent info available") else: - print("War: {} vs {}".format(war.clan.name, war.opponent.name)) + print(f"War: {war.clan.name} vs {war.opponent.name}") + + +async def main(): + coc_client = coc.Client() + try: + await coc_client.login(os.environ.get("DEV_SITE_EMAIL"), + os.environ.get("DEV_SITE_PASSWORD")) + except coc.InvalidCredentials as error: + exit(error) + + await get_warlog_opponents_from_clan_name(coc_client, "Reddit Zulu", 5) + await coc_client.close() if __name__ == "__main__": - client.loop.run_until_complete(get_warlog_opponents_from_clan_name("name", 5)) - client.close() + asyncio.run(main()) diff --git a/setup.py b/setup.py index 717c09b4..fbcc73d8 100644 --- a/setup.py +++ b/setup.py @@ -7,7 +7,7 @@ with open(os.path.join(os.getcwd(), "requirements.txt")) as f: REQUIREMENTS = f.read().splitlines() -VERSION = "2.0.1" +VERSION = "2.0.2" if "a" in VERSION: VERSION += "+" + subprocess.check_output(["git", "rev-parse", "--short", "HEAD"]).decode("utf-8").strip() @@ -56,9 +56,10 @@ def run(self): include_package_data=True, extras_require={"docs": ["sphinx", "sphinx_rtd_theme", "sphinxcontrib_trio", "autodocsumm"]}, classifiers={ - "Programming Language :: Python :: 3.5", - "Programming Language :: Python :: 3.6", "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", }, cmdclass={"lint": LintCommand}, )