القليل من الراحة في حياة الطلاب

قصة بسيطة عن شعوري بالخجل من أن أطلب باستمرار من زملائي المعلومات المفقودة وقررت أن نجعل حياتنا أسهل قليلاً.

أعتقد أن العديد من زملائي على دراية بالحالة عندما يكون في غرفة الدردشة العامة ، حيث غالبًا ما تومض المعلومات المهمة ، هناك حوالي 30 محاورًا نشطًا يقومون بتحميل قواعد بيانات فكونتاكتي باستمرار مع رسائلهم. في ظل هذه الظروف ، من غير المحتمل أن يرى الجميع هذه المعلومات الهامة. يحدث لي أيضا. قبل عام ، تقرر تصحيح سوء الفهم هذا.

أولئك الذين هم على استعداد لعدم السخط من المقال التالي حول الروبوت ، أسأل تحت القط.

نظرًا لأنني طالب من المستوى الأول ، فستكون الأمثلة مرتبطة بهذا الموضوع.
لذلك ، هناك مهمة: لجعل نقل المعلومات من headman إلى الطلاب مناسبًا لكل من headman والطلاب. بفضل الميزات الجديدة نسبيًا في فكونتاكتي (أي الرسائل الشخصية من المجتمعات) ، لفت انتباهي القرار على الفور. يجب أن يتلقى البوت الجالس في مجموعة رسائل من كبار السن (كبار السن ، إذا كان هناك العديد من المجموعات في ساحة المشاركات) وإرسالها إلى الأطراف المعنية (الطلاب).

تم تعيين المهمة ، تابع.

سنحتاج إلى:

1. مكتبة vk_api لاستخدام vk api
2. ore peewee للعمل مع قاعدة البيانات
3. وبيثون وحدات مدمجة

أيضا ، قبل القراءة ، أقترح تحديث أنماط "الأوبزرفر" ( هبر ، ويكي ) و "واجهة" ( هبر ، ويكي )

الجزء 1. "سعدت بلقائك ، الرفيق بوت."

تحتاج أولاً إلى تعليم الروبوت الخاص بنا لفهم نفسه كمجتمع. قم بإنشاء فصل يسمى Group. كوسيطة ، دعها تقبل كائن جلسة وكائن ممثل قاعدة بيانات (وكيل).

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) 

إنشاء طريقة لتحديث أفراد المجتمع. قسمهم على الفور إلى إداريين ومشاركين واحفظهم في قاعدة البيانات.

• يتم تكوين self.api عند إنشاء الفئة الأساسية للمجموعة (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 

يجب أن يكون هذا الفصل قادرًا أيضًا على إرسال الرسائل إلى المستلمين ، وبالتالي فإن الطريقة التالية في الاستوديو. في المعلمات: القائمة البريدية ونص الرسالة والتطبيق. يبدأ هذا الأمر برمته في سلسلة محادثات منفصلة بحيث يمكن للبوت استقبال رسائل من المشاركين الآخرين.
يتم تلقي الرسائل في وضع متزامن ، لذلك مع زيادة عدد العملاء النشطين ، ستنخفض سرعة الاستجابة بشكل واضح.

  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 

على هذا ، يبدو أن تجريد المجموعة قد انتهى. التقينا مع جميع أفراد المجتمع ، والآن نحن بحاجة إلى معرفة كيفية فهمهم.

الجزء 2. "Psh ... مرحبا .."

دع البوت يستمع إلى جميع الرسائل من أعضاء مجتمعنا.
للقيام بذلك ، قم بإنشاء فئة ChatHandler ، والتي ستقوم بذلك
في المعلمات:

• group_manager هو مثال على فئة المجتمع التي كتبناها للتو
• يتعرف command_observer على الأوامر المتصلة (ولكن المزيد عن ذلك في الجزء الثالث)

 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 

علاوة على ذلك ، في الواقع ، نستمع إلى رسائل من المستخدمين ونتعرف على الأوامر.

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

الجزء 3. "ماذا كتبت عن ملكي ..؟"

يتم التعامل مع التعرف على الأوامر من خلال نظام فرعي منفصل يتم تنفيذه من خلال نمط المراقب.
انتباه 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 

ولكن ماذا سيعرف هذا المراقب؟
لذلك وصلنا إلى الجزء الأكثر إثارة للاهتمام - الفريق.
كل فريق هو فئة مستقلة ، سليل من فئة القيادة الأساسية.
كل ما هو مطلوب للأمر هو تشغيل طريقة () إذا تم العثور على الكلمة الأساسية الخاصة به في بداية رسالة المستخدم. يتم تعريف الكلمات الأساسية للأوامر في متغير المشغلات لفئة الأمر (سلسلة أو قائمة سلاسل)

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

كما يتبين من توقيع طريقة () ، يتلقى كل أمر رابطًا لمثيل عضو في المجموعة ، ورسالته (بدون كلمة أساسية) ، والتطبيقات ، ورابطًا لمثيل المجموعة. أي أن كل تفاعل مع عضو المجموعة يقع على عاتق الفريق. أعتقد أن هذا هو الحل الأكثر صحة ، لأنه بهذه الطريقة من الممكن إنشاء غلاف (شل) لمزيد من التفاعل.
(في الحقيقة ، للقيام بذلك ، ستحتاج إما إلى إضافة غير متزامن ، لأن المعالجة متزامنة ، أو يجب معالجة كل رسالة مستلمة في سلسلة رسائل جديدة ، وهي ليست مربحة على الإطلاق)

أمثلة على تنفيذ الأوامر:

 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 

الجزء 4. "نحن فريق واحد كبير".

الآن يجب دمج وتكوين كل هذه الوحدات والمعالجات.
فئة أخرى من فضلك!
إنشاء واجهة من شأنها تخصيص الروبوت لدينا.

 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 

المرحلة الأخيرة هي الإطلاق. دائمًا ما يكون أكثر سوءًا ، لأن نوعًا ما من المفاجأة يمكن أن يخرج. ليس هذه المرة.

• يتم استيراد ConfigParser من core.settings.ConfigParser. في الواقع ، يقرأ التكوين فقط.

• يتم استيراد project_path من وحدة الإعدادات في جذر المشروع.

  
  

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

  
  

يبدو أن هذا كل شيء.

في الوقت الحالي ، أفاد هذا البرنامج ثلاث مجموعات على الأقل ، وآمل أن يجلب لك أيضًا.

يمكنك نشره مجانًا على Heroku ، لكن هذه قصة أخرى.

المراجع:

• مستودع على جيثب