Wenig Bequemlichkeit im Studentenleben

Eine einfache Geschichte darüber, wie ich mich schämte, Klassenkameraden ständig nach fehlenden Informationen zu fragen, und ich beschloss, unser Leben ein wenig einfacher zu gestalten.

Ich glaube, dass viele meiner Kollegen mit der Situation vertraut sind, wenn in einem allgemeinen Chatroom, in dem wichtige Informationen häufig geflasht werden, etwa 30 aktive Gesprächspartner ständig Vkontakte-Datenbanken mit ihren Nachrichten laden. Unter solchen Bedingungen ist es unwahrscheinlich, dass jeder diese wichtigen Informationen sieht. Das passiert mir auch. Vor einem Jahr wurde beschlossen, dieses Missverständnis zu korrigieren.

Diejenigen, die bereit sind, sich nicht über den nächsten Artikel über den Bot zu empören, frage ich unter Katze.

Da ich ein Student der ersten Stufe bin, werden Beispiele zu diesem Thema verwendet.
Es gibt also eine Aufgabe: Die Übertragung von Informationen vom Schulleiter an die Schüler ist sowohl für den Schulleiter als auch für die Schüler bequem. Dank der relativ neuen Funktionen von Vkontakte (nämlich persönliche Nachrichten von Communities) fiel mir die Entscheidung sofort auf. Ein Bot, der in einer Gruppe sitzt, sollte Nachrichten vom Ältesten (Älteste, wenn sich viele Gruppen im Stream befinden) empfangen und an interessierte Parteien (Studenten) senden.

Die Aufgabe ist eingestellt, fahren Sie fort.

Wir werden brauchen:

1. vk_api Bibliothek zur Verwendung von vk api
2. peewee orm , um mit der Datenbank zu arbeiten
3. und in Python integrierte Module

Außerdem schlage ich vor dem Lesen vor, die Erinnerung an die Muster "Observer" ( Habr , Wiki ) und "Facade" ( Habr , Wiki ) zu aktualisieren .

Teil 1. "Schön, Sie kennenzulernen, Genosse Bot."

Zuerst müssen Sie unserem Bot beibringen, sich selbst als Gemeinschaft zu verstehen. Erstellen Sie eine Klasse namens Group. Lassen Sie es als Argumente ein Sitzungsobjekt und ein datenbankrepräsentatives Objekt (Proxy) akzeptieren.

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) 

Erstellen Sie eine Methode zum Aktualisieren von Community-Mitgliedern. Teilen Sie sie sofort in Administratoren und Teilnehmer auf und speichern Sie sie in der Datenbank.

• self.api wird beim Erstellen der Gruppenbasisklasse (BaseCommunicateVK) konfiguriert.

 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 

Diese Klasse sollte auch in der Lage sein, Nachrichten an Empfänger zu senden, also die nächste Methode im Studio. In den Parametern: Mailingliste, Nachrichtentext und Anwendung. Das Ganze beginnt in einem separaten Thread, damit der Bot Nachrichten von anderen Teilnehmern empfangen kann.
Nachrichten werden im synchronen Modus empfangen, sodass mit zunehmender Anzahl aktiver Clients die Antwortgeschwindigkeit offensichtlich abnimmt.

  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 

Damit scheint die Abstraktion der Gruppe vorbei zu sein. Wir haben uns mit allen Mitgliedern der Community getroffen, jetzt müssen wir lernen, wie man sie versteht.

Teil 2. "Psh ... willkommen .."

Lassen Sie den Bot alle Nachrichten von Mitgliedern unserer Community abhören.
Erstellen Sie dazu eine Klasse ChatHandler, die dies ausführt
In den Parametern:

• group_manager ist eine Instanz der Community-Klasse, die wir gerade geschrieben haben
• command_observer erkennt verbundene Befehle (mehr dazu im dritten Teil)

 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 

Außerdem hören wir Nachrichten von Benutzern ab und erkennen Befehle.

 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() 

Teil 3. "Was hast du über meine geschrieben?"

Die Befehlserkennung wird von einem separaten Subsystem ausgeführt, das über das Observer-Muster implementiert ist.
Achtung 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 

Aber was wird dieser Beobachter erkennen?
Also kamen wir zum interessantesten Teil - dem Team.
Jedes Team ist eine unabhängige Klasse, ein Nachkomme der Basisbefehlsklasse.
Für den Befehl muss lediglich die Methode continue () ausgeführt werden, wenn das Schlüsselwort am Anfang der Nachricht des Benutzers gefunden wird. Befehlsschlüsselwörter werden in der Auslöservariablen der Befehlsklasse (Zeichenfolge oder Liste von Zeichenfolgen) definiert.

 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(' ') 

Wie aus der Signatur der Methode continue () hervorgeht, erhält jeder Befehl eine Verknüpfung zu einer Instanz eines Gruppenmitglieds, seine Nachricht (ohne Schlüsselwort), Anwendungen und eine Verknüpfung zu einer Gruppeninstanz. Das heißt, die gesamte Interaktion mit einem Gruppenmitglied liegt beim Team. Ich denke, dies ist die richtigste Lösung, da auf diese Weise eine Shell (Shell) für eine größere Interaktivität erstellt werden kann.
(In Wahrheit müssen Sie dazu entweder asynchron hinzufügen, da die Verarbeitung synchron ist, oder jede empfangene Nachricht sollte in einem neuen Thread verarbeitet werden, was überhaupt nicht rentabel ist.)

Beispiele für die Implementierung von Befehlen:

 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 

Teil 4. "Wir sind ein großes Team."

Jetzt müssen alle diese Module und Handler kombiniert und konfiguriert werden.
Noch eine Klasse bitte!
Erstellen Sie eine Fassade, die unseren Bot anpasst.

 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 

Die letzte Phase ist der Start. Immer die böse, weil eine Art Überraschung herauskommen kann. Nicht dieses Mal.

• ConfigParser wird aus core.settings.ConfigParser importiert. Tatsächlich liest es nur die Konfiguration.

• project_path wird aus dem Einstellungsmodul im Projektstamm importiert.

  
  

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

  
  

Das scheint alles zu sein.

Im Moment hat dieses Programm mindestens drei Gruppen geholfen und ich hoffe, Sie werden es auch mitbringen.

Sie können es kostenlos auf Heroku bereitstellen, aber das ist eine andere Geschichte.

Referenzen:

• Repository auf Github