Вводная

Привет. У меня есть несколько групп в ВК, в которые нужно периодически публиковать посты. В целом мое желание можно описать фразой «А когда мне это делать, если я все время не хочу». Мотивация в моем случае это лень.

Я не являюсь разработчиком, но решил набросать скрипт, который будет публиковать посты. Дополнительный бонус для меня получение опыта в области разработки.

Требования

Контент, который я публикую это тематические фото и изображения с короткими комментариями. На данный момент все разложено по папкам. Альбомы и картинки имеют свои порядковые номера.

Что важно отразить в алгоритме:

• публикацию постов через временной интервал;

• разместить недельный план публикаций с комментариями к постам;

• возможность опубликовать пост без очереди;

• публиковать на выбор одну или несколько изображений;

• хранить историю опубликованных фото и комментариев;

• автопостинг для нескольких групп.

Логика

Все пункты требований я постарался вместить в три основных этапа:

1. Проверка временного интервала публикаций. Проверяем текущую дату и время с временем последней опубликованной записью. Получаем интервал и сравниваем его с целевым.

2. Определить очередь публикации. Сравниваем количество публикаций по плану в день и по факту из истории публикаций.

3. Определить неопубликованное ранее фото и комментарий. Берем из плана номер альбома и выгружаем из истории опубликованные фото, определяем картинку и комментарий, которые не публиковались.

Работая над таблицами задался вопросом: «А что должен делать скрипт, когда все фото будут опубликованы?». Решил, что нужно предусмотреть отдельную таблицу в БД, которая будет отображать количества фото в альбоме.

Предусмотреть триггер, где после публикаций всех изображений, запускается пересчет картинок в альбоме и обновлять дату. Скрипт будет смотреть на новую дату и выгружать из истории фото, которые опубликованы позже этой даты, так мы начинаем новый круг публикаций изображений.

Схема работы приложения

Если смотреть на схему получилось довольно простое приложение, в котором скрипт взаимодействует с базой данной, социальной сетью и приложением для логирования.

Итоговая логика работы скрипта у меня получилась следующая:

Наполнение и таблицы

Очередь для публикаций будет выглядеть следующим образом:

- день недели;

- номер альбома с картинками (отсортированы по нарастающей);

- количество фото для публикации;

- признак: фото одиночное или тематически объединены в несколько по умолчанию.

Иерархия размещения альбомов с картинками:

Исходя из поставленной задачи и иерархии контента у меня получилось 5 таблиц для БД:

• history_post -  история опубликованных постов

• line_post – план недельной публикации

• out_of_line_post – поставить публикацию вне плана

• total_photos – количество фото в альбоме

• comment_photo – комментарии к картинкам из альбомов

Более детальное описание столбцов оставлю ниже.

Разработка

Файлы и их назначение.

сonfig.yaml – файл со всеми настройками. БД, Grafana, VK.
config_table.json – справочник с названиями таблиц и столбцов в БД
main.py – запуск приложения
utils.py – основная логика
sql_query.py – запросы к базе
manager_DB.py – подключение к базе и запись строк в БД
GetPic.py – определение картинок для публикации
loadVkConnent.py – публикация контента в VK
logi.py – тексты с логами

Не буду подробно разбирать каждый файл, остановлюсь только на utils.py, все файлы будут на GitLab

Создаем класс ManagerPost, который наследует все остальные классы, а именно:

• manager_DB – подключение к базе и запись строк в БД
• GP - определение картинок для публикации
• SqlMan - запросы к базе
• VkMan - публикация контента в VK
• Log – логи

Создаем функции:

CheckTime.

По действиям:

• Выгружаем кол-во постов, которые необходимо опубликовать;
• Определяем текущие время и время публикации последнего поста;
• Через If определяем время с учетом количества постов в день и какой интервал между нами. Если True продолжаем скрипт, если нет выходим из программы.

Если необходимо справочник TimeIntervalDict можно изменить под ваши потребности.

class ManagerPost(manager_DB,GP,SqlMan,VkMan,Log):
    "класс для управленеия постановкой постов"
    
    def CheckTime (self):
        """определяем сколько разница по времени должна быть
        исходя из количества планируемых постов
        возвращает True если необходимое время прошло"""
        
        CountPost = str(pd.read_sql(SqlMan.CountPostDay(self), self.conn)['count'][0])
        TimeIntervalDict = {'2': 13, 
                    '3': 6,
                    '4': 5,
                    }
        
        Today = datetime.now()
       
        TimePost = pd.read_sql(SqlMan.GetTimePostFromLine(self), self.conn)['time_push'][0]
       
        if (Today - TimePost) > timedelta(hours=TimeIntervalDict[CountPost]):
            Log.TimeCorrect(self)
            return 
        else: 
            Log.TimeLess(self,Today,TimePost,timedelta(hours=TimeIntervalDict[CountPost]))
            sys.exit()

GetInfoPost

• Получаем пост, который должен быть опубликован. Если DataFrama пустой, то отправляем лог, о том, что все посты уже опубликованы и завершаем скрипт
• По номеру альбома получаем все ранее опубликованные фото и общие количество картинок в альбоме
• Складываем количество опубликованных фото + кол-во которое нужно опубликовать в текущей сессии. Если значение получается больше (True), чем фото, то в альбоме запускается ветка переопределения количества фото в альбоме.
На выход выплевывается информация о посте который нужно опубликовать и список опубликованных картинок.

def GetInfoPost(self):
      """Проверка количество уже опубликованных фото.
      Если все фото почти опубликованы, обнуляем в БД запись с новым 
      количеством фото"""

      InfoPost = pd.read_sql(SqlMan.GetInfoPostSQL(self), self.conn)

      if InfoPost.shape[0] == 0:
          Log.AllAlbomPush(self)
          sys.exit()
      IdAlbom = InfoPost['id_albom'][0]
      IdPhotoPush = pd.read_sql(SqlMan.GetIdPic(self,  IdAlbom), self.conn)
      ListPhotoPush = IdPhotoPush['id_pic'].to_list()
      
      BoolCheck = (len(IdPhotoPush) + InfoPost['count_photo'][0]) >= InfoPost['total_photo'][0]
      
      # переопределяем количество фото в таблице
      if  BoolCheck == True:
          Date = datetime.now(timezone.utc)
 
          if InfoPost['union_photos'][0] == False:
              AlbomDir = Path(self.WithoutUnion, str(IdAlbom))
              CountPhotoFromAlbom = len([name for name in os.listdir(AlbomDir) if os.path.isfile(os.path.join(AlbomDir, name))])
          else:
              AlbomDir = Path(self.WithUnion, str(IdAlbom))
              CountPhotoFromAlbom = len(set([filename.split('-')[0] for filename in os.listdir(AlbomDir)]))

              
          LoadConnent = [[IdAlbom, CountPhotoFromAlbom, 'TIMESTAMP WITH TIME ZONE ', self.IdGroupQuery ,InfoPost['union_photos'][0]]]

          mananger_DB.UploadRowDB(self,self.ConfigTable['total_photos'],LoadConnent, Date) 

          Log.NewCountPhoto(self,IdAlbom,CountPhotoFromAlbom)
          ListPhotoPush = []

      return InfoPost, ListPhotoPush

GetComment

• По номеру альбома и картинки получаем таблицу с комментариями из БД, если запись отсутствует оставляем NULL;
• Получаем историю ранее опубликованных комментариев к фото;
• Далее объединяем два df и выбираем комментарий, который публиковался реже всех.

def GetCommment(self, InfoPost, BoxNewPic):
    "определяем комменатрий для публикации"

    try:
        CommentDf = pd.DataFrame(\
                pd.read_sql(SqlMan.GetIdCom(self,InfoPost, BoxNewPic, True), self.conn)['comment'][0].split(';'),\
                columns=['comment'])
    except IndexError:
        comment = 'NULL'
        return  comment
    
    # вытаскиваем все комментарии
    CommentHistory = pd.read_sql(SqlMan.GetIdCom(self,InfoPost, BoxNewPic, False), self.conn)


    # считаем комментарии
    CommentBox = CommentDf.merge(CommentHistory, on='comment', how='left').fillna(0).sort_values(by='count').reset_index(drop=True)
    Log.CommentPhoto(self, CommentBox)

    return  CommentBox['comment'][0]

CommitPost

Эта сборка алгоритма. Получилась избыточная функция и на мой взгляд ее надо переписать и разбить на 2-3 более мелких. Но раз работает, решил не трогать. Прокомментирую ее по частям.
• проверяем время публикации;
• определяем если ли у нас публикации вне плана;
• если True, проверяем совпадения дат. Вытаскиваем коммент, определяем фото и отправляем на публикацию;
• если False, определяем фото и комментарий из плана публикаций;

def CommitPost(self):
        
    # проверяем новые посты и время
    ManagerPost.CheckTime(self)

    InfoPost = pd.read_sql(SqlMan.GetNewLinePost(self), self.conn)
    today = datetime.now()
    
    # сценарий если новая фото
    if InfoPost[InfoPost['date']==today.date()].shape[0] != 0:

        Line = False
        if InfoPost['comment'][0] != None:
            Comment = f"{InfoPost['comment'][0]}"
        else:
            Comment = 'NULL'
        
        if InfoPost['union_photos'][0] == True:
            NewIdPic, BoxNewPic, DirAlbom = GP.GetPicUnionTrue(self,InfoPost,[], True)
        else:
            BoxNewPic, DirAlbom = GP.GetPicUnionFalse(self,InfoPost,[],True)
        
        self.cursor.execute(SqlMan.ChahgeStatusNewPost(self))
        self.conn.commit()

    else:
        # сценарий если в рамках линии
        Log.NewPostNotExist(self)
        InfoPost, IdPhotoPush = ManagerPost.GetInfoPost(self)
        Line = True

        if InfoPost['union_photos'][0] == True:
            NewIdPic, BoxNewPic, DirAlbom = GP.GetPicUnionTrue(self,InfoPost,IdPhotoPush)
        else:
            BoxNewPic, DirAlbom = GP.GetPicUnionFalse(self,InfoPost,IdPhotoPush)
        
        # получаем номер фото и коммент
        Log.CommentPhoto(self, BoxNewPic)
        Comment = ManagerPost.GetCommment(self,InfoPost, BoxNewPic[0])

• определяем время для отложенного постинга в vk;
• далее грузим данные в БД. Развилка по алгоритму в поле InfoPost['union_photos'] (фото одно или несколько тематически связанных);
• закрываем соединения с базой;

# определяем время
minute = randint(5, 59)
Date = datetime(today.year, today.month, today.day, today.hour, minute)

BoxWithPath = []

# отображаем время публикации в БД (history_post)
if InfoPost['union_photos'][0] == False:
    for i in range(len(BoxNewPic)):
        LoadConnent = [
        [InfoPost['id_albom'][0], 
        BoxNewPic[i],
        Comment.strip(),
        Line, 
        'TIMESTAMP WITH TIME ZONE ',
        self.GroupIdForDB,
        InfoPost['union_photos'][0]]
        ]
        manager_DB.UploadRowDB(self, self.ConfigTable['history_post'],LoadConnent, Date)
        Log.PostUploudDB(self)
        BoxWithPath.append(DirAlbom + str('\\') + str(BoxNewPic[i]) + str('.jpg'))
else:
        LoadConnent = [
        [InfoPost['id_albom'][0], 
        NewIdPic,
        Comment.strip(),
        Line, 
        'TIMESTAMP WITH TIME ZONE ',
        self.GroupIdForDB,
        InfoPost['union_photos'][0]]
        ]
    
        manager_DB.UploadRowDB(self, self.ConfigTable['history_post'],LoadConnent, Date)
        Log.PostUploudDB(self)

        for i in range(len(BoxNewPic)):
            BoxWithPath.append(DirAlbom + str('\\') + str(BoxNewPic[i]) + str('.jpg'))

manager_DB.CoonClose(self)
Log.InfoPostLog(self,InfoPost['id_albom'][0],BoxNewPic,Comment,Date)
        

• переводим дату для формата публикации в VK;
• загружаем фото в VK оставляем комментарий и ставим публикацию на таймер;
• отправляем лог с деталями публикации.

unixtime = time.mktime(Date.timetuple())
VkMan.LoadVkСontent(self,BoxWithPath,unixtime,Comment)
Log.PostUploudVK(self)

return 

Развернуть

Чтобы не нагружать читателя еще текстом, о том, как я настроил БД, Grafana Loki, загрузил картинки и запустил код на сервере я напишу в отдельной статье.

Результат

Цель достигнута посты опубликовываются.

История сохраняется.

Что можно доработать?

«Умные мысли часто преследовали его, но он был быстрее». Когда все было написано и протестировано, я подумал, что можно было сделать:

• контент с изображениями хранить в облаке;

В текущем решение, изображение нужно хранить локально. Если вам необходимо добавить/удалить картинки, то нужно обновлять их локально. Как оказалось, на практике это неудобно.

• функция для загрузки и выгрузки из БД плана публикаций, комментарий, количество фото;

На этапе тестирования я немного поменял скрипт загрузил туда необходимые данные и в процессе понял, что у меня нет отдельной функции для этого. Посмотреть и произвести точечные изменения можно через pgAdmin подключившись к базе postgres. Когда нужно загрузить несколько десятков или сотни строк, функция бы пригодилась.

• уведомление в случае ошибки;

Все логи отправляются на Grafana Loki, при тестировании появилась потребность настроить уведомления на почту в случае ошибки.

• обработать ошибки при загрузках в VK;

За все несколько недель тестирования 1-2 раза отлетал api vk или происходил сбой при загрузке картинок. В таком случае в БД отображается строка, что публикация прошла успешно, а в реальности ошибка. Возможно имеет смысл обработать эту ошибку и удалить последнею строку в БД.

• предусмотреть несколько расширений для изображений.

На данный момент приложение работает только с расширением jpg для остальных форматов картинок скрипт выдаст ошибку. Можно переписать часть кода, чтобы сделать его универсальным. У меня такой потребности нет, но думаю она не будет лишний.

Заключение

Скрипт работает и цель автопостинга достигнута. Возможно мое решение поможет кому-то высвободить свое время. Наверное скрипт можно использовать для любой социальной сети у которой есть api адаптировав код под свои нужны. Если у вас есть идеи, предложения как можно улучшить скрипт или есть свое решение пишите в комментариях.

Код на GitHub