Python vk meme bot example.

vk_bot.py

# -*- encoding: utf-8 -*-
import os
import pathlib
from threading import Thread

import requests
import random
import secrets
import uuid
from multiprocessing import Process
import re

import vk_api
from PIL import Image, ImageDraw, ImageFont
from aiohttp import web
from settings import API_KEY
from chatterbot import ChatBot

BASE_DIR = os.path.dirname(os.path.realpath(__file__))

session = vk_api.VkApi(token=API_KEY, api_version='5.38')

upload = vk_api.VkUpload(session)
api = session.get_api()

FONT_PATH = os.path.join(os.path.join(BASE_DIR, 'fonts'), 'impact.ttf')
MESSAGE_TYPES = (
    'message_new'
)

RE_HOOKS = (
    r'^((-ш) (\w+)){,1}(.*)$',
)


BOTTOM_OFFSET = 5
FONT_SIZE = 72


def text_wrap(text, font, max_width):
    lines = []
    # If the width of the text is smaller than image width
    # we don't need to split it, just add it to the lines array
    # and return
    if font.getsize(text)[0] <= max_width:
        lines.append((text, 0))
    else:
        # split the line by spaces to get words
        words = text.split(' ')
        i = 0
        # append every word to a line while its width is shorter than image width
        while i < len(words):
            line = ''
            size = 0
            while i < len(words) and font.getsize(line + words[i])[0] <= max_width:
                line = line + words[i] + " "
                i += 1

            if not line:
                line = words[i]
                i += 1
            # when the line gets longer than the max width do not append the word,
            # add the line to the lines array
            line_width = font.getsize(line)[0]
            lines.append((line, line_width))
    return lines


def image_is_smaller(img, max_x, max_y):
    return img.size[0] < max_x and img.size[1] < max_y


def set_font(font):
    font_path = os.path.exists(os.path.join(os.path.join(BASE_DIR, 'fonts'), font.lower() + '.ttf'))
    if os.path.exists(font_path):
        FONT_PATH = os.path.join(os.path.join(BASE_DIR, 'fonts'), font + '.ttf')



def open_random_image(images_folder_path='terrach_images'):
    """
    Функция, которая возвращает случайно выбранное изобаржение из пути.
    :param images_folder_path: str
    :return: Image
    """
    images_folder_path = os.path.join(BASE_DIR, images_folder_path)
    images_directory_files = os.listdir(images_folder_path)
    random_image = secrets.choice(images_directory_files)
    image = Image.open(pathlib.Path(os.path.join(images_folder_path, random_image)))

    max_x, max_y = 960, 960

    if image_is_smaller(image, max_x, max_y):
        image = image.resize((max_x, max_y), Image.ANTIALIAS)
    else:
        image.thumbnail((max_x, max_y), Image.ANTIALIAS)

    return image


def draw_stroked_text(img, x, y, text, font, offset=4, inline_color='white', outline_color='black'):
    """
    Функция, которая рисует обведенный текст на изображении
    :param img: Image
    :param x: int
    :param y: int
    :param text: str
    :param font: ImageFont
    :param offset: int
    :param inline_color: str or tuple
    :param outline_color: str  or tuple
    :return: None
    """
    # Внутренний цвет текста
    font.color = inline_color
    inline_font = font

    # Цвет обводки
    font.color = outline_color
    #font.size = FONT_SIZE + 3
    outline_font = font
    draw = ImageDraw.Draw(img)
    # Рисуем текст 8 раз в разных направлениях с одинаковым смещением, цвет внешней обводки.
    for min_offset in range(offset):
        draw.text((x + min_offset, y), text, fill=outline_color, font=outline_font)
        draw.text((x + min_offset, y + min_offset), text, fill=outline_color, font=outline_font)
        draw.text((x, y + min_offset), text, fill=outline_color, font=outline_font)
        draw.text((x - min_offset, y), text, fill=outline_color, font=outline_font)
        draw.text((x - min_offset, y - min_offset), text, fill=outline_color, font=outline_font)
        draw.text((x, y - min_offset), text, fill=outline_color, font=outline_font)
        draw.text((x + min_offset, y - min_offset), text, fill=outline_color, font=outline_font)
        draw.text((x - min_offset, y + min_offset), text, fill=outline_color, font=outline_font)

    draw.text((x, y), text, fill=inline_color, font=inline_font)


def mark_image(img, text):
    """
    Непосредственно функция рисования текста на картинке.
    В качестве аргумента принимает изображение Pillow и возвращает изображение с текстом.
    :param img: Image
    :param text: str
    :return: Image
    """
    img_width, img_height = img.size
    draw = ImageDraw.Draw(img)
    font = ImageFont.truetype(FONT_PATH, FONT_SIZE)
    line_img_height = font.getsize('hg')[1]

    w, h = draw.textsize(text, font=font)

    text = text.upper()
    text_lines = text_wrap(text, font, img_width)

    x, y = (img_width - int(text_lines[0][1])) / 2, (img_height - h * len(text_lines))
    y -= BOTTOM_OFFSET
    for text_line in text_lines:
        text = text_line[0]
        line_img_width, _ = draw.textsize(text, font=font)

        x = (img_width - line_img_width) / 2
        draw_stroked_text(img, x=x, y=y, text=text, font=font)

        y = y + line_img_height

    return img


commands = {
    '-ш': globals()["set_font"]
}


def send_reply(user_id, user_message, args=None):
    global api
    global upload

    if args:
        key = args[0]
        value = args[1]
        if key in commands:
            commands[value](value)

    random_image = open_random_image()
    marked_image = mark_image(random_image, user_message)
    marked_image_ext = '.png'
    marked_image_name = str(uuid.uuid1()) + marked_image_ext
    marked_image.save(marked_image_name, )

    photo = upload.photo_messages(photos=marked_image_name)
    print(photo)
    photo = photo[0]
    # photo = bytes(photo)
    print(f'peer_id: {user_id}')
    attachments = 'photo{}_{}'.format(photo['owner_id'], photo['id'])
    api.messages.send(peer_id=user_id, message='', attachment=attachments)
    os.remove(marked_image_name)


async def webhook(request):
    data = await request.json()
    msg_type = data.get('type')
    print(data)
    if msg_type == 'message_new':
        vk_object = data.get('object')
        user_id = vk_object.get('peer_id')
        print(vk_object)
        message = vk_object['text']
        message = re.sub(r'\[[^\]]*\]', '', message).strip()
        #message = message.split(' ')[1:]
        #message = ' '.join(message)
        for re_hook in RE_HOOKS:
            matching = re.match(re_hook, message)
            if matching:
                message = matching.group(4).strip()
                key = matching.group(2)
                value = matching.group(3)
                command_args = None
                if key and value:
                    command_args = (key, value)
                p = Thread(target=send_reply, args=(user_id, message, command_args))
                p.start()
                p.join()
                break

    return web.Response(body="OK")


app = web.Application()
app.add_routes([web.post('/', webhook)])

if __name__ == '__main__':
    web.run_app(app, port=80)