royalnet/royalpack/commands/abstract/linker.py

from typing import *
import royalnet.commands as rc
import royalnet.utils as ru
import royalnet.serf.telegram as rst
import royalnet.backpack.tables as rbt
import abc
import logging
import asyncio as aio
from ...types import Updatable


log = logging.getLogger(__name__)


class LinkerCommand(rc.Command, metaclass=abc.ABCMeta):

    def __init__(self, interface: rc.CommandInterface):
        super().__init__(interface)

        self.updater_task = None
        if self.enabled():
            self.updater_task = self.loop.create_task(self.run_updater())

    async def run(self, args: rc.CommandArgs, data: rc.CommandData) -> None:
        author = await data.get_author(error_if_none=True)
        if len(args) == 0:
            message = []
            for obj in await self.get_updatables_of_user(session=data.session, user=author):
                async def change(attribute: str, value: Any):
                    """A shortcut for self.__change."""
                    await self._change(session=data.session,
                                       obj=obj,
                                       attribute=attribute,
                                       new=value)

                await self.update(session=data.session, obj=obj, change=change)
                message.append(self.describe(obj))
            if len(message) == 0:
                raise rc.UserError("Nessun account connesso.")
            await data.session_commit()
            await data.reply("\n".join(message))
        else:
            created = await self.create(session=data.session, user=author, args=args, data=data)
            await data.session_commit()
            if created is not None:
                message = ["🔗 Account collegato!", "", self.describe(created)]
                await data.reply("\n".join(message))

    def describe(self, obj: Updatable) -> str:
        """The text that should be appended to the report message for a given Updatable."""
        return str(obj)

    @abc.abstractmethod
    async def get_updatables_of_user(self, session, user: rbt.User) -> List[Updatable]:
        """Get the updatables of a specific user."""
        ...

    @abc.abstractmethod
    async def get_updatables(self, session) -> List[Updatable]:
        """Return a list of all objects that should be updated at this updater cycle."""
        ...

    @abc.abstractmethod
    async def create(self,
                     session,
                     user: rbt.User,
                     args: rc.CommandArgs,
                     data: Optional[rc.CommandData] = None) -> Optional[Updatable]:
        """Create a new updatable object for a user.

        This function is responsible for adding the object to the session."""
        ...

    @abc.abstractmethod
    async def update(self, session, obj, change: Callable[[str, Any], Awaitable[None]]):
        """Update a single updatable object. Use the change method to change values on the object!"""
        ...

    @abc.abstractmethod
    async def on_increase(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
        """Called when the attribute has increased from the old value."""
        ...

    @abc.abstractmethod
    async def on_unchanged(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
        """Called when the attribute stayed the same as the old value."""
        ...

    @abc.abstractmethod
    async def on_decrease(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
        """Called when the attribute has decreased from the old value."""
        ...

    @abc.abstractmethod
    async def on_first(self, session, obj: Updatable, attribute: str, old: None, new: Any) -> None:
        """Called when the attribute changed from None."""
        ...

    @abc.abstractmethod
    async def on_reset(self, session, obj: Updatable, attribute: str, old: Any, new: None) -> None:
        """Called when the attribute changed to None."""
        ...

    async def _change(self,
                      session,
                      obj,
                      attribute: str,
                      new) -> None:
        """Set the value of an attribute of an object to a value, and call the corresponding method."""
        old = obj.__getattribute__(attribute)
        if new == old:
            await self.on_unchanged(session=session,
                                    obj=obj,
                                    attribute=attribute,
                                    old=old,
                                    new=new)
        else:
            if old is None:
                await self.on_first(session=session,
                                    obj=obj,
                                    attribute=attribute,
                                    old=old,
                                    new=new)
            elif new is None:
                await self.on_reset(session=session,
                                    obj=obj,
                                    attribute=attribute,
                                    old=old,
                                    new=new)
            elif new > old:
                await self.on_increase(session=session,
                                       obj=obj,
                                       attribute=attribute,
                                       old=old,
                                       new=new)
            else:
                await self.on_decrease(session=session,
                                       obj=obj,
                                       attribute=attribute,
                                       old=old,
                                       new=new)
        obj.__setattr__(attribute, new)

    def enabled(self) -> bool:
        """Whether the updater is enabled or not."""
        return self.config[self.name]["updater"]["enabled"] and self.interface.name == "telegram"

    def period(self) -> int:
        """The time between two updater cycles."""
        return self.config[self.name]["updater"]["period"]

    def delay(self) -> int:
        """The time between two object updates."""
        return self.config[self.name]["updater"]["delay"]

    def target(self) -> int:
        """The id of the Telegram chat where notifications should be sent."""
        return self.config[self.name]["updater"]["target"]

    async def run_updater(self):
        log.info(f"Starting updater: {self.name}")

        while True:
            log.debug(f"Updater cycle: {self.name}")
            session = self.alchemy.Session()
            objects = await self.get_updatables(session)

            for obj in objects:
                log.debug(f"Updating: {obj} ({self.name})")

                async def change(attribute: str, value: Any):
                    """A shortcut for self.__change."""
                    await self._change(session=session,
                                       obj=obj,
                                       attribute=attribute,
                                       new=value)

                try:
                    await self.update(session=session,
                                      obj=obj,
                                      change=change)
                except Exception as e:
                    ru.sentry_exc(e)

                delay = self.delay()
                log.debug(f"Waiting for: {delay} seconds (delay)")
                await aio.sleep(delay)

            log.debug(f"Committing updates: {self.name}")
            await ru.asyncify(session.commit)
            session.close()

            period = self.period()
            log.debug(f"Waiting for: {period} seconds (period)")
            await aio.sleep(period)

    async def notify(self, message):
        await self.serf.api_call(self.serf.client.send_message,
                                 chat_id=self.target(),
                                 text=rst.escape(message),
                                 parse_mode="HTML",
                                 disable_webpage_preview=True)