Sedikit kenyamanan dalam kehidupan siswa

Sebuah cerita sederhana tentang bagaimana saya merasa malu untuk terus-menerus meminta informasi yang hilang kepada teman sekelas dan saya memutuskan untuk membuat hidup kita sedikit lebih mudah.

Saya percaya bahwa banyak rekan saya yang akrab dengan situasi ketika di ruang obrolan umum, di mana informasi penting sering muncul, ada sekitar 30 lawan bicara aktif yang terus-menerus memuat basis data Vkontakte dengan pesan mereka. Dalam kondisi seperti itu, tidak mungkin semua orang akan melihat informasi penting ini. Itu terjadi pada saya juga. Setahun yang lalu, diputuskan untuk memperbaiki kesalahpahaman ini.

Mereka yang siap untuk tidak marah tentang artikel selanjutnya tentang bot, saya bertanya di bawah kucing.

Karena saya adalah siswa tingkat pertama, contoh akan terkait dengan topik ini.
Jadi, ada tugas: untuk membuat transfer informasi dari kepala sekolah ke siswa nyaman bagi kepala sekolah dan siswa. Berkat fitur-fitur Vkontakte yang relatif baru (yaitu, pesan pribadi dari komunitas), keputusan itu langsung menarik perhatian saya. Bot yang duduk dalam kelompok harus menerima pesan dari penatua (penatua, jika ada banyak kelompok di sungai) dan mengirimkannya ke pihak yang berkepentingan (siswa).

Tugas sudah diatur, lanjutkan.

Kami akan membutuhkan:

1. pustaka vk_api untuk menggunakan vk api
2. peewee orm untuk bekerja dengan database
3. dan modul built-in python

Juga, sebelum membaca, saya sarankan menyegarkan pola Pengamat ( Habr , wiki ) dan Fasad ( Habr , wiki )

Bagian 1. "Senang bertemu denganmu, Kamerad Bot."

Pertama, Anda perlu mengajarkan bot kami untuk memahami dirinya sebagai komunitas. Buat kelas yang disebut Grup. Sebagai argumen, biarkan ia menerima objek sesi dan objek perwakilan basis data (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) 

Buat metode untuk memperbarui anggota komunitas. Segera bagi mereka menjadi administrator dan peserta dan simpan dalam database.

• self.api dikonfigurasikan saat membuat kelas dasar Grup (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 

Kelas ini juga harus dapat mengirim pesan ke penerima, jadi metode selanjutnya di studio. Dalam parameter: milis, teks pesan dan aplikasi. Semua ini dimulai di utas terpisah sehingga bot dapat menerima pesan dari peserta lain.
Pesan diterima dalam mode sinkron, jadi dengan peningkatan jumlah klien aktif, kecepatan respons jelas akan menurun.

  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 

Mengenai hal ini, tampaknya, abstraksi kelompok sudah berakhir. Kami bertemu dengan semua anggota komunitas, sekarang kami perlu belajar bagaimana memahami mereka.

Bagian 2. "Psh ... selamat datang .."

Biarkan bot mendengarkan semua pesan dari anggota komunitas kami.
Untuk melakukan ini, buat kelas ChatHandler, yang akan melakukan ini
Dalam parameter:

• group_manager adalah turunan dari kelas komunitas yang baru saja kita tulis
• command_observer mengenali perintah yang terhubung (tetapi lebih dari itu di bagian ketiga)

 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 

Lebih jauh, pada kenyataannya, kami mendengarkan pesan dari pengguna dan mengenali perintah.

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

Bagian 3. "Apa yang kamu tulis tentang milikku ..?"

Pengenalan perintah ditangani oleh subsistem terpisah yang diimplementasikan melalui pola Observer.
Attention 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 

Tapi apa yang akan dilihat pengamat ini?
Jadi kami sampai pada bagian yang paling menarik - tim.
Setiap tim adalah kelas independen, keturunan dari kelas Command dasar.
Semua yang diperlukan dari perintah ini adalah menjalankan metode continue () jika kata kuncinya ditemukan di awal pesan pengguna. Kata kunci perintah didefinisikan dalam variabel pemicu dari kelas perintah (string atau daftar string)

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

Seperti yang dapat dilihat dari tanda tangan dari metode continue (), setiap perintah menerima tautan ke turunan anggota grup, pesannya (tanpa kata kunci), aplikasi, dan tautan ke turunan grup. Artinya, semua interaksi dengan anggota grup ada di tangan tim. Saya pikir ini adalah solusi yang paling benar, karena dengan cara ini dimungkinkan untuk membuat shell (Shell) untuk interaktivitas yang lebih besar.
(Sebenarnya, untuk melakukan ini, Anda perlu menambahkan asinkron, karena pemrosesan sinkron, atau setiap pesan yang diterima harus diproses dalam utas baru, yang sama sekali tidak menguntungkan)

Contoh implementasi perintah:

 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 

Bagian 4. "Kami adalah satu tim besar."

Sekarang semua modul dan penangan ini perlu digabungkan dan dikonfigurasi.
Kelas lain, tolong!
Buat fasad yang akan menyesuaikan bot kami.

 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 

Tahap terakhir adalah peluncuran. Selalu yang paling jahat, karena semacam kejutan bisa keluar. Tidak kali ini.

• ConfigParser diimpor dari core.settings.ConfigParser. Bahkan, itu hanya membaca konfigurasi.

• project_path diimpor dari modul pengaturan di root proyek.

  
  

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

  
  

Sepertinya hanya itu saja.

Saat ini, program ini telah menguntungkan setidaknya tiga kelompok dan, saya harap, akan membawa Anda juga.

Anda dapat menggunakannya secara gratis di Heroku, tapi itu cerita lain.

Referensi:

• repositori di github