छात्र जीवन में थोड़ी सुविधा

लापता होने की जानकारी के लिए सहपाठियों से लगातार पूछने में शर्म महसूस करने के बारे में एक सरल कहानी और मैंने अपने जीवन को थोड़ा आसान बनाने का फैसला किया।

मेरा मानना ​​है कि मेरे कई साथी उस स्थिति से परिचित हैं जब एक सामान्य चैट रूम में, जहां महत्वपूर्ण जानकारी काफी बार चमकती है, लगभग 30 सक्रिय वार्ताकार हैं जो अपने संदेशों के साथ VKontakte डेटाबेस को लगातार लोड करते हैं। ऐसी शर्तों के तहत, यह संभावना नहीं है कि हर कोई इस महत्वपूर्ण जानकारी को देखेगा। मेरे साथ भी ऐसा होता है। एक साल पहले इस गलतफहमी को ठीक करने का फैसला किया गया था।

जो लोग बॉट के बारे में अगले लेख के बारे में बताने के लिए तैयार नहीं हैं, मैं बिल्ली के नीचे पूछता हूं।

चूंकि मैं प्रथम श्रेणी का छात्र हूं, इसलिए उदाहरण इस विषय से संबंधित होंगे।
इसलिए, एक कार्य है: मुखिया से छात्रों के लिए जानकारी के हस्तांतरण को मुखिया और छात्रों दोनों के लिए सुविधाजनक बनाना। Vkontakte की अपेक्षाकृत नई विशेषताओं के लिए धन्यवाद (अर्थात्, समुदायों से व्यक्तिगत संदेश), निर्णय ने तुरंत मेरी आंख को पकड़ लिया। एक समूह में बैठे एक बॉट को बड़े (बड़ों, यदि धारा पर कई समूह हैं) से संदेश प्राप्त करना चाहिए और उन्हें इच्छुक पार्टियों (छात्रों) को भेजना चाहिए।

कार्य निर्धारित है, आगे बढ़ें।

हमें आवश्यकता होगी:

1. vk_api vk api का उपयोग करने के लिए लाइब्रेरी
2. डेटाबेस के साथ काम करने के लिए peewee orm
3. और अजगर में निर्मित मॉड्यूल

इसके अलावा, पढ़ने से पहले, मैं "ऑब्जर्वर" ( हैबर , विकी ) और " फेसडे " ( हैबर , विकी ) पैटर्न की याद को ताज़ा करने का प्रस्ताव करता हूं।

भाग 1. "आपसे मिलकर अच्छा लगा, कॉमरेड बॉट।"

पहले आपको खुद को समुदाय के रूप में समझने के लिए हमारे बॉट को सिखाना होगा। ग्रुप नामक एक क्लास बनाएं। तर्क के रूप में, इसे सत्र ऑब्जेक्ट और डेटाबेस प्रतिनिधि ऑब्जेक्ट (प्रॉक्सी) को स्वीकार करने दें।

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) 

समुदाय के सदस्यों को अपडेट करने के लिए एक विधि बनाएं। तुरंत उन्हें प्रशासकों और प्रतिभागियों में विभाजित करें और उन्हें डेटाबेस में सहेजें।

• समूह आधार वर्ग (BaseCommunicateVK) बनाते समय self.api कॉन्फ़िगर किया गया है

 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 ... स्वागत .."

बॉट को हमारे समुदाय के सदस्यों के सभी संदेश सुनने दें।
ऐसा करने के लिए, एक वर्ग चैटहैंडलर बनाएं, जो यह करेगा
मापदंडों में:

• 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. "आपने मेरे बारे में क्या लिखा है ..?"

कमांड मान्यता को ऑब्जर्वर पैटर्न के माध्यम से कार्यान्वित एक अलग सबसिस्टम द्वारा नियंत्रित किया जाता है।
ध्यान दें:

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

  
  

वह सब मालूम होता है।

फिलहाल, इस कार्यक्रम से कम से कम तीन समूहों को फायदा हुआ है और मुझे उम्मीद है कि आप भी इसे लाएंगे।

आप इसे हरोकू पर मुफ्त में तैनात कर सकते हैं, लेकिन यह एक और कहानी है।

संदर्भ:

• गितुब पर भंडार