from flask import Flask, request, json
from units.methods import Method
from units.vk_utils import *
import vk_api
app = Flask(__name__)
vk = vk_api.VkApi(token=VK_TOKEN)
class Bot:
@app.route("/", methods=['POST', 'GET'])
def event_handler(self) -> str:
""" Function, which handles POST requests
to servers and then processes it.
If request does not contains field "method"
:returns: Error -> str
:return: ok -> str
"""
data = json.loads(request.data)
if 'method' in data and data["secret"] == SECRET:
if data['method'] == Method.ADD_USER:
self.add_user(data['object']['user_id'])
if data['method'] == Method.BAN_EXPIRED:
self.ban_expired(data['object']['user_id'], data['object'].get('comment', 'Неизвестна.'))
if data['method'] == Method.DELETE_MESSAGES_FROM_USER:
self.delete_messages(data['object']['user_id'], data['object']['is_spam'])
if data['method'] == Method.FORBIDDEN_LINKS:
self.delete_links(*data['object']['local_ids'])
else:
return "Error"
return "ok"
@staticmethod
def add_user(user_id: int) -> None:
""" Signal ADD_USER
Uses to adding user after command usage
:param user_id: int
:return None:
"""
if check_friends(vk, user_id):
add_to_chat(vk, user_id)
else:
send_msg(vk, "Пользователь отсутствует в друзьях.")
@staticmethod
def ban_expired(user_id: int, reason: str) -> None:
""" Signal BAN_EXPIRED
Uses to mention and add user after his ban expiring
:param user_id: int
:param reason: str
:return None:
"""
if check_friends(vk, user_id):
add_to_chat(vk, user_id)
else:
users_to_add.append(user_id)
send_msg(
vk,
f'У пользователя {get_user_name(vk, user_id, with_id=True)} истек срок бана.' +
f'\nПричина бана: {reason}'
f'\nОн не добавил меня в друзья, поэтому я не смогу пригласить его в беседу.'
)
@staticmethod
def delete_messages(user_id: int, spam: bool = False) -> None:
""" Signal DELETE_MESSAGES_FROM_USER
Uses to delete messages from user
after command usage
:param user_id: int
:param spam: bool
:return None:
"""
with vk_api.VkRequestsPool(vk) as pool:
for i in vk.method("messages.getHistory", {"peer_id": peer_id, "count": 200})['items']:
if i['from_id'] == user_id:
pool.method("messages.delete", {"message_ids": i['id'], "delete_for_all": 1, "spam": 1 if spam else 0})
@staticmethod
def delete_links(*message_ids: list) -> None:
""" Signal forbiddenLinks
Uses to delete found links by local_ids
:param message_ids: list
:return None:
"""
with vk_api.VkRequestsPool(vk) as pool:
for i in vk.method("messages.getHistory", {"peer_id": peer_id, "count": 200})['items']:
if i["conversation_message_id"] in message_ids:
pool.method("messages.delete", {"message_ids": i['id'], "delete_for_all": 1})
@staticmethod
def start() -> None:
""" Static method which
starts "Flask" app
:return None:
"""
app.run(host="0.0.0.0", port=8080)
if __name__ == '__main__':
bot = Bot()
bot.start()
