Poca conveniencia en la vida estudiantil.

Una historia simple sobre cómo me daba vergüenza preguntar constantemente a los compañeros por información faltante y decidí hacer nuestra vida un poco más fácil.

Creo que muchos de mis compañeros están familiarizados con la situación cuando en una sala de chat general, donde a menudo se muestra información importante, hay unos 30 interlocutores activos que constantemente cargan las bases de datos Vkontakte con sus mensajes. En tales condiciones, es poco probable que todos vean esta información importante. A mí también me pasa. Hace un año, se decidió corregir este malentendido.

Pregunto a los que están listos para no indignarse con el próximo artículo sobre el bot.

Como soy un estudiante de primer nivel, los ejemplos estarán relacionados con este tema.
Entonces, hay una tarea: hacer que la transferencia de información del director a los estudiantes sea conveniente tanto para el director como para los estudiantes. Gracias a las características relativamente nuevas de Vkontakte (es decir, mensajes personales de las comunidades), la decisión me llamó la atención de inmediato. Un bot sentado en un grupo debe recibir mensajes del anciano (ancianos si hay muchos grupos en la transmisión) y enviarlos a las partes interesadas (estudiantes).

La tarea está establecida, proceda.

Necesitaremos:

1. vk_api biblioteca para usar vk api
2. orbe orm para trabajar con la base de datos
3. y módulos integrados de python

Además, antes de leer, propongo actualizar la memoria de los patrones "Observador" ( Habr , Wiki ) y "Fachada" ( Habr , Wiki )

Parte 1. "Encantado de conocerte, camarada Bot".

Primero debes enseñarle a nuestro bot a entenderse como una comunidad. Crea una clase llamada Grupo. Como argumentos, permita que acepte un objeto de sesión y un objeto representativo de la base de datos (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) 

Cree un método para actualizar a los miembros de la comunidad. Divídalos inmediatamente en administradores y participantes y guárdelos en la base de datos.

• self.api se configura al crear la clase base de 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 

Esta clase también debería poder enviar mensajes a los destinatarios, por lo que el siguiente método en el estudio. En los parámetros: lista de correo, mensaje de texto y aplicación. Todo esto comienza en un hilo separado para que el bot pueda recibir mensajes de otros participantes.
Los mensajes se reciben en modo síncrono, por lo que con un aumento en el número de clientes activos, la velocidad de respuesta obviamente disminuirá.

  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 

En esto, al parecer, la abstracción del grupo ha terminado. Nos reunimos con todos los miembros de la comunidad, ahora necesitamos aprender cómo entenderlos.

Parte 2. "Psh ... bienvenido .."

Deje que el bot escuche todos los mensajes de los miembros de nuestra comunidad.
Para hacer esto, cree una clase ChatHandler, que hará esto
En los parámetros:

• group_manager es una instancia de la clase comunitaria que acabamos de escribir
• command_observer reconoce los comandos conectados (pero más sobre eso en la tercera 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 

Además, de hecho, escuchamos mensajes de usuarios y reconocemos 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. "¿Qué escribiste sobre el mío ...?"

El reconocimiento de comandos es manejado por un subsistema separado implementado a través del patrón Observador.
Atención 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 

Pero, ¿qué reconocerá este observador?
Así que llegamos a la parte más interesante: el equipo.
Cada equipo es una clase independiente, un descendiente de la clase de Comando base.
Todo lo que se requiere del comando es ejecutar el método proceder () si su palabra clave se encuentra al comienzo del mensaje del usuario. Las palabras clave de comando se definen en la variable desencadenantes de la clase de comando (cadena o lista de cadenas)

 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 puede ver en la firma del método Proceder (), cada comando recibe un enlace a una instancia de un miembro del grupo, su mensaje (sin una palabra clave), aplicaciones y un enlace a una instancia de grupo. Es decir, toda interacción con un miembro del grupo descansa en el equipo. Creo que esta es la solución más correcta, ya que de esta manera es posible crear un shell (Shell) para una mayor interactividad.
(En verdad, para hacer esto, necesitará agregar asíncrono, porque el procesamiento es sincrónico, o cada mensaje recibido debe procesarse en un nuevo hilo, que no es rentable)

Ejemplos de implementación 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 un gran equipo".

Ahora todos estos módulos y controladores deben combinarse y configurarse.
Otra clase por favor!
Crea una fachada que personalizará nuestro 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 

La última etapa es el lanzamiento. Siempre el más desagradable, porque puede surgir algún tipo de sorpresa. Esta vez no

• ConfigParser se importa desde core.settings.ConfigParser. De hecho, solo lee la configuración.

• project_path se importa desde el módulo de configuración en la raíz del proyecto.

  
  

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

  
  

Eso parece ser todo.

Por el momento, este programa ha beneficiado al menos a tres grupos y espero que también lo traiga a usted.

Puedes implementarlo gratis en Heroku, pero esa es otra historia.

Referencias

• repositorio en github