Pouca conveniência na vida estudantil

Uma história simples sobre como eu me envergonhava de pedir constantemente informações aos colegas de classe e decidi facilitar nossa vida um pouco.

Acredito que muitos de meus colegas estão familiarizados com a situação quando em uma sala de bate-papo geral, onde informações importantes são frequentemente exibidas, existem cerca de 30 interlocutores ativos que carregam constantemente seus bancos de dados Vkontakte com suas mensagens. Sob tais condições, é improvável que todos vejam essas informações importantes. Isso acontece comigo também. Há um ano, foi decidido corrigir esse mal-entendido.

Aqueles que estão prontos para não ficar indignados com o próximo artigo sobre o bot, pergunto em gato.

Como eu sou um estudante de primeiro nível, exemplos estarão relacionados a este tópico.
Portanto, existe uma tarefa: tornar conveniente a transferência de informações do diretor para os alunos, tanto para o diretor quanto para os alunos. Graças aos recursos relativamente novos do Vkontakte (a saber, mensagens pessoais das comunidades), a decisão chamou minha atenção imediatamente. Um bot sentado em um grupo deve receber mensagens do ancião (idosos, se houver muitos grupos no fluxo) e enviá-los para as partes interessadas (estudantes).

A tarefa está definida, prossiga.

Vamos precisar de:

1. biblioteca vk_api para usar vk api
2. peewee orm para trabalhar com o banco de dados
3. e módulos internos python

Além disso, antes de ler, proponho atualizar a memória dos padrões "Observer" ( Habr , Wiki ) e "Facade" ( Habr , Wiki )

Parte 1. "Prazer em conhecê-lo, camarada Bot."

Primeiro você precisa ensinar nosso bot a entender-se como uma comunidade. Crie uma classe chamada grupo. Como argumentos, aceite um objeto de sessão e um objeto representativo do banco de dados (Proxy).

class Group(BaseCommunicateVK): def __init__(self, vksession, storage): super().__init__(vksession) self.storage = storage 

 class BaseCommunicateVK: longpoll = None def __init__(self, vksession): self.session = vksession self.api = vksession.get_api() if BaseCommunicateVK.longpoll is None: BaseCommunicateVK.longpoll = VkLongPoll(self.session) def get_api(self): return self.api def get_longpoll(self): return self.longpoll def method(self, func, args): return self.api.method(func, args) @staticmethod def create_session(token=None, login=None, password=None, api_v='5.85'): try: if token: session = vk_api.VkApi(token=token, api_version=api_v) elif login and password: session = vk_api.VkApi(login, password, api_version=api_v) else: raise vk_api.AuthError("Define login and password or token.") return session except vk_api.ApiError as error: logging.info(error) def get_last_message(self, user_id): return self.api.messages.getHistory( peer_id=user_id, count=1)["items"][0] @staticmethod def get_attachments(last_message): if not last_message or "attachments" not in last_message: return "" attachments = last_message["attachments"] attach_strings = [] for attach in attachments: attach_type = attach["type"] attach_info = attach[attach_type] attach_id = attach_info["id"] attach_owner_id = attach_info["owner_id"] if "access_key" in attach_info: access_key = attach_info["access_key"] attach_string = "{}{}_{}_{}".format(attach_type, attach_owner_id, attach_id, access_key) else: attach_string = "{}{}_{}".format(attach_type, attach_owner_id, attach_id) attach_strings.append(attach_string) return ",".join(attach_strings) @staticmethod def get_forwards(attachments, last_message): if not attachments or "fwd_count" not in attachments: return "" if len(last_message["fwd_messages"]) == int(attachments["fwd_count"]): return last_message["id"] def send(self, user_id, message, attachments=None, **kwargs): send_to = int(user_id) if "last_message" in kwargs: last_message = kwargs["last_message"] else: last_message = None p_attachments = self.get_attachments(last_message) p_forward = self.get_forwards(attachments, last_message) if message or p_attachments or p_forward: self.api.messages.send( user_id=send_to, message=message, attachment=p_attachments, forward_messages=p_forward) if destroy: accept_msg_id = self.api.messages \ .getHistory(peer_id=user_id, count=1) \ .get('items')[0].get('id') self.delete(accept_msg_id, destroy_type=destroy_type) def delete(self, msg_id, destroy_type=1): self.api.messages.delete(message_id=msg_id, delete_for_all=destroy_type) 

Crie um método para atualizar membros da comunidade. Divida-os imediatamente em administradores e participantes e salve-os no banco de dados.

• self.api é configurado ao criar a classe base do grupo (BaseCommunicateVK)

 def update_members(self): fields = 'domain, sex' admins = self.api.groups.getMembers(group_id=self.group_id, fields=fields, filter='managers') self.save_members(self._configure_users(admins)) members = self.api.groups.getMembers(group_id=self.group_id, fields=fields) self.save_members(self._configure_users(members)) return self def save_members(self, members): self.storage.update(members) @staticmethod def _configure_users(items, exclude=None): if exclude is None: exclude = [] users = [] for user in items.get('items'): if user.get('id') not in exclude: member = User() member.configure(**user) users.append(member) return users 

Essa classe também deve poder enviar mensagens para os destinatários, portanto, o próximo método no estúdio. Nos parâmetros: lista de endereçamento, texto da mensagem e aplicativo. Essa coisa toda começa em um thread separado para que o bot possa receber mensagens de outros participantes.
As mensagens são recebidas no modo síncrono, portanto, com um aumento no número de clientes ativos, a velocidade da resposta obviamente diminuirá.

  def broadcast(self, uids, message, attachments=None, **kwargs): report = BroadcastReport() def send_all(): users_ids = uids if not isinstance(users_ids, list): users_ids = list(users_ids) report.should_be_sent = len(users_ids) for user_id in users_ids: try: self.send(user_id, message, attachments, **kwargs) if message or attachments: report.sent += 1 except vk_api.VkApiError as error: report.errors.append('vk.com/id{}: {}'.format(user_id, error)) except ValueError: continue for uid in self.get_member_ids(admins=True, moders=True): self.send(uid, str(report)) broadcast_thread = Thread(target=send_all) broadcast_thread.start() broadcast_thread.join() 

 class BroadcastReport: def __init__(self): self.should_be_sent = 0 self.sent = 0 self.errors = [] def __str__(self): res = "#  #" res += "\n: {}  ".format(self.should_be_sent) res += "\n: {} ".format(self.sent) if self.errors: res += "\n:" for i in self.errors: res += "\n- {}".format(i) return res 

Parece que a abstração do grupo acabou. Nos reunimos com todos os membros da comunidade, agora precisamos aprender a entendê-los.

Parte 2. "Psh ... bem-vindo .."

Deixe o bot ouvir todas as mensagens dos membros da nossa comunidade.
Para fazer isso, crie uma classe ChatHandler, que fará isso
Nos parâmetros:

• group_manager é uma instância da classe da comunidade que acabamos de escrever
• command_observer reconhece comandos conectados (mas mais sobre isso na terceira parte)

 class ChatHandler(Handler): def __init__(self, group_manager, command_observer): super().__init__() self.longpoll = group_manager.get_longpoll() self.group = group_manager self.api = group_manager.get_api() self.command_observer = command_observer 

Além disso, de fato, ouvimos mensagens de usuários e reconhecemos comandos.

 def listen(self): try: for event in self.longpoll.listen(): if event.user_id and event.type == VkEventType.MESSAGE_NEW and event.to_me: self.group.api.messages.markAsRead(peer_id=event.user_id) self.handle(event.user_id, event.text, event.attachments, message_id=event.message_id) except ConnectionError: logging.error("I HAVE BEEN DOWNED AT {}".format(datetime.datetime.today())) self.longpoll.update_longpoll_server() def handle(self, user_id, message, attachments, **kwargs): member = self.group.get_member(user_id) self.group.update_members() self.command_observer.execute(member, message, attachments, self.group, **kwargs) def run(self): self.listen() 

Parte 3. "O que você escreveu sobre o meu ..?"

O reconhecimento de comandos é tratado por um subsistema separado implementado através do padrão Observer.
Atenção CommandObserver:

 class CommandObserver(AbstractObserver): def execute(self, member, message, attachments, group, **kwargs): for command in self.commands: for trigger in command.triggers: body = command.get_body(trigger, message) if body is not None: group.api.messages.setActivity(user_id=member.id, type="typing") if command.system: kwargs.update({"trigger": trigger, "commands": self.commands}) else: kwargs.update({"trigger": trigger}) return command.proceed(member, body, attachments, group, **kwargs) 

 class AbstractObserver(metaclass=ABCMeta): def __init__(self): self.commands = [] def add(self, *args): for arg in args: self.commands.append(arg) @abstractmethod def execute(self, *args, **kwargs): pass 

Mas o que esse observador reconhecerá?
Então chegamos à parte mais interessante - a equipe.
Cada equipe é uma classe independente, um descendente da classe Command de base.
Tudo o que é necessário para o comando é executar o método continue () se sua palavra-chave for encontrada no início da mensagem do usuário. As palavras-chave de comando são definidas na variável triggers da classe de comando (linha ou lista de linhas)

 class Command(metaclass=ABCMeta): def __init__(self): self.triggers = [] self.description = "Empty description." self.system = False self.privilege = False self.activate_times = [] self.activate_days = set() self.autostart_func = self.proceed def proceed(self, member, message, attachments, group, **kwargs): raise NotImplementedError() @staticmethod def get_body(kw, message): if not isinstance(kw, list): kw = [kw, ] for i in kw: reg = '^ *(\\{}) *'.format(i) if re.search(reg, message): return re.sub(reg, '', message).strip(' ') 

Como você pode ver na assinatura do método continue (), cada comando recebe um link para uma instância de um membro do grupo, sua mensagem (sem uma palavra-chave), aplicativos e um link para uma instância do grupo. Ou seja, toda interação com um membro do grupo é da equipe. Eu acho que essa é a solução mais correta, pois dessa forma é possível criar um shell (Shell) para maior interatividade.
(Na verdade, para fazer isso, você precisará adicionar assíncrono, porque o processamento é síncrono ou cada mensagem recebida deve ser processada em um novo encadeamento, que não é de todo lucrativo)

Exemplos de implementação de comandos:

 class BroadcastCommand(Command): def __init__(self): super().__init__() self.triggers = ['.mb'] self.privilege = True self.description = "    ." def proceed(self, member, message, attachments, group, **kwargs): if member.id not in group.get_member_ids(admins=True, editors=True): group.send(member.id, "You cannot do this ^_^") return True last_message = group.get_last_message(member.id) group.broadcast(group.get_member_ids(), message, attachments, last_message=last_message, **kwargs) return True 

 class HelpCommand(Command): def __init__(self): super().__init__() self.commands = [] self.triggers = ['.h', '.help'] self.system = True self.description = "  ." def proceed(self, member, message, attachments, group, **kwargs): commands = kwargs["commands"] help = "  :\n\n" admins = group.get_member_ids(admins=True, moders=True) i = 0 for command in commands: if command.privilege and member.id not in admins: continue help += "{}) {}\n\n".format(i + 1, command.name()) i += 1 group.send(member.id, help) return True 

Parte 4. "Somos uma grande equipe."

Agora todos esses módulos e manipuladores precisam ser combinados e configurados.
Outra aula, por favor!
Crie uma fachada que personalizará nosso bot.

 class VKManage: def __init__(self, token=None, login=None, password=None): self.session = BaseCommunicateVK.create_session(token, login, password, api_version) self.storage = DBProxy(DatabaseORM) self.group = Group(self.session, self.storage).setup().update_members() self.chat = ChatHandler(self.group, CommandObserver.get_observer()) def start(self): self.chat.run() def get_command(self, command_name): return { " ": BroadcastCommand(), " ": AdminBroadcastCommand(), "": HelpCommand(), " ": SkippedLectionsCommand(), "": TopicTimetableCommand().setup_account(self.bot.api), }.get(command_name) def connect_command(self, command_name): command = self.get_command(str(command_name).lower()) if command: self.chat.command_observer.add(command) return self def connect_commands(self, command_names): for i in command_names.split(','): self.connect_command(i.strip()) return self 

A última etapa é o lançamento. Sempre o mais desagradável, porque pode surgir algum tipo de surpresa. Não desta vez.

• O ConfigParser é importado do core.settings.ConfigParser. De fato, apenas lê a configuração.

• O caminho do projeto é importado do módulo de configurações na raiz do projeto.

  
  

   if __name__ == '__main__': config = ConfigParser(project_path) VKManage(token=config['token'], login=config['login'], password=config['password'])\ .connect_commands(",  ,  ,  ")\ .start() 

  
  

Isso parece ser tudo.

No momento, este programa beneficiou pelo menos três grupos e, espero, trará a você também.

Você pode implantá-lo gratuitamente no Heroku, mas isso é outra história.

Referências:

• repositório no github