Парсим и управляем постами VK: Разбор Python-скрипта для работы с API ВКонтакте

от автора

В этой статье разберем создание Python-скрипта для работы со стеной VK. Научимся автоматизировать сбор постов и рассмотрим осторожное удаление контента через API. Началось все с того, что я решил почистить свою стенку в ВК. Жизнь, знаете ли, непредсказуемая нынче 🙂 Но, в целом, мне видится, что код может быть полезен как для общего развития, так и в качестве основы для бэкапа данных, модерации контента или анализа своей активности в соцсети.

Содержание

  • Введение в VK API

  • Архитектура скрипта

  • Аутентификация и работа с токенами

  • Получение постов с пагинацией

  • Безопасное удаление контента

  • Соломка

  • Вместо заключения

Введение в VK API

VK предоставляет мощный API для разработчиков. Для работы с ним нужен access token — ключ доступа, который определяет права приложения. Наш скрипт использует следующие методы API:

  • users.get — информация о пользователе

  • wall.get — получение постов со стены

  • wall.delete — удаление постов

Архитектура скрипта

Скрипт построен по ООП принципам. Основной класс VKParser инкапсулирует всю логику работы с API:

class VKParser:     def init(self, login, password):     self.login = login     self.password = password     self.session = requests.Session()     self.access_token = None     self.user_id = None

Используем requests.Session() для сохранения cookies и повторного использования соединения, что ускоряет множественные запросы.

Аутентификация и работа с токенами

Получение access token

Прямой логин-пароль в VK API давно уже не используется, посему, вместо этого получаем token через OAuth:

def login_vk(self):     auth_url = "https://oauth.vk.com/authorize"     auth_params = {         'client_id': '6121396',  # Standalone app ID         'redirect_uri': 'https://oauth.vk.com/blank.html',         'display': 'page',         'scope': 'wall,offline',  # Requested permissions         'response_type': 'token',         'v': '5.131'  # API version     }          # Generate auth URL     auth_link = f"{auth_url}?{'&'.join([f'{k}={v}' for k, v in auth_params.items()])}"     print(f"Авторизуйтесь по ссылке: {auth_link}")

Права доступа (scope)

  • wall — доступ к стене (чтение и запись)

  • offline — бессрочный доступ без повторной авторизации

  • groups — доступ к группам (если нужно)

Получение постов с пагинацией

VK API возвращает посты порциями (обычно до 100 за запрос), с чем пришлось позаниматься интимишвили. Реализована она следующим образом:

def get_all_posts(self):     all_posts = []     offset = 0     count = 100  # Max per request          while True:         response = self._make_api_request('wall.get', {             'owner_id': self.user_id,             'count': count,             'offset': offset         })                  posts = response['response']['items']         if not posts:             break                      all_posts.extend(posts)         offset += count                  time.sleep(0.5)  # Rate limiting                  if len(posts) < count:             break          return all_posts

Обработка rate limits

Важно соблюдать лимиты VK API (1000 сообщений со стены удалялись весьма не быстро):

  • 3 запроса в секунду

  • Задержка time.sleep(0.5) между запросами

  • Обработка ошибок и таймаутов

Безопасное удаление контента

Многоуровневое подтверждение

Удаление постов — необратимая операция, поэтому тут тройное внимание если будете играть с кодом. Реализуем защиту от случайного запуска:

def delete_posts_with_confirmation(self):     print("⚠️  ВНИМАНИЕ! ЭТА ОПЕРАЦИЯ НЕОБРАТИМА! ⚠️")          confirmation1 = input("Введите 'DELETE MY POSTS' для подтверждения: ")     if confirmation1 != "DELETE MY POSTS":         return False          confirmation2 = input("Введите 'YES I AM SURE' для окончательного подтверждения: ")     if confirmation2 != "YES I AM SURE":         return False          return self._delete_posts(posts)

Постепенное удаление с отчетностью

def _delete_posts(self, posts):     deleted_count = 0     error_count = 0          for i, post in enumerate(posts, 1):         response = self._make_api_request('wall.delete', {             'owner_id': self.user_id,             'post_id': post['id']         })                  if response and response['response'] == 1:             deleted_count += 1         else:             error_count += 1                  time.sleep(1)  # Important delay          print(f"Успешно: {deleted_count}, Ошибок: {error_count}")     return error_count == 0

Пример работы

parser = VKParser("your_login", "your_password")  # Авторизация if parser.login_vk():     # ПОлучить посты     posts = parser.get_all_posts()     print(f"Найдено {len(posts)} постов")          # Сохранить в файл для бэкапа     parser.save_posts_to_file(posts)          # Удаляем (с подтверждением)     parser.delete_posts_with_confirmation()

Немного соломки на случай факапа

1. Бэкап перед удалением

Всегда сохраняйте данные перед любыми destructive-операциями:

def save_posts_to_file(self, texts, filename='vk_posts.txt'):     with open(filename, 'w', encoding='utf-8') as f:         f.write(f"Всего постов: {len(texts)}\n")         for text in texts:             f.write(text + "\n\n")

2. Обработка ошибок API

def _make_api_request(self, method, params):     try:         response = self.session.get(             f'https://api.vk.com/method/{method}',             params=params,             timeout=30         )         data = response.json()                  if 'error' in data:             print(f"API Error: {data['error']['error_msg']}")             return None                      return data     except Exception as e:         print(f"Request failed: {e}")         return None

3. Безопасное хранение токенов

В коде, разумеется, никаких секретов/токенов/паролей! Используйте переменные окружения:

import os  token = os.getenv('VK_ACCESS_TOKEN') if not token:     token = input("Введите access token: ")

Вместо заключения

Скрипт со своей основной задачей справился, но, потенциально, конечно, есть с чем поиграться при наличии фантазии. Тут нам и выгрузка в JSON, XML и тд, и фильтрации сообщений, и все такое прочее.

Статья написана в познавательных целях. Автор не несет ответственности за использование приведенного кода.

Готовый к запуску код доступен на GitHub и ниже целиком под спойлером

Скрытый текст
import requests import json import time   class VKParser:     def __init__(self, login, password):         self.login = login         self.password = password         self.session = requests.Session()         self.access_token = None         self.user_id = None      def login_vk(self):         """Авторизация в VK"""         print("Выполняю авторизацию...")          # Получить данные для авторизации         auth_url = "https://oauth.vk.com/authorize"         auth_params = {             'client_id': '6121396',             'redirect_uri': 'https://oauth.vk.com/blank.html',             'display': 'page',             'scope': 'wall,offline,groups',  # Добавили права для удаления             'response_type': 'token',             'v': '5.131'         }          print("Для работы скрипта необходим access_token")         print("Получите его по ссылке:")         print(f"{auth_url}?{'&'.join([f'{k}={v}' for k, v in auth_params.items()])}")         print("\nПосле авторизации скопируйте access_token из адресной строки")          self.access_token = input("Введите access_token: ").strip()          if self.access_token:             user_info = self._make_api_request('users.get', {})             if user_info and 'response' in user_info:                 self.user_id = user_info['response'][0]['id']                 print(f"Успешная авторизация! User ID: {self.user_id}")                 return True             else:                 print("Ошибка авторизации. Проверьте токен.")                 return False         return False      def _make_api_request(self, method, params):         """Выполнение запроса к VK API"""         params['access_token'] = self.access_token         params['v'] = '5.131'          try:             response = self.session.get(                 f'https://api.vk.com/method/{method}',                 params=params,                 timeout=30             )             return response.json()         except Exception as e:             print(f"Ошибка при запросе к API: {e}")             return None      def get_all_posts(self):         """Получение всех постов со стены"""         if not self.access_token:             print("Сначала выполните авторизацию!")             return []          print("Начинаю сбор постов...")         all_posts = []         offset = 0         count = 100          while True:             print(f"Загружаю посты, offset: {offset}")              response = self._make_api_request('wall.get', {                 'owner_id': self.user_id,                 'count': count,                 'offset': offset             })              if not response or 'response' not in response:                 print("Ошибка при получении постов")                 break              posts = response['response']['items']             if not posts:                 break              all_posts.extend(posts)             offset += count              time.sleep(0.5)              if len(posts) < count:                 break          return all_posts      def delete_all_posts(self):         """Удаление всех постов со стены"""         if not self.access_token:             print("Сначала выполните авторизацию!")             return False          print("⚠️  ВНИМАНИЕ! ЭТА ОПЕРАЦИЯ НЕОБРАТИМА! ⚠️")         print("Все посты будут удалены без возможности восстановления!")          confirmation = input("Вы уверены? (введите 'DELETE ALL' для подтверждения): ")         if confirmation != "DELETE ALL":             print("Операция отменена.")             return False          posts = self.get_all_posts()         if not posts:             print("Постов для удаления нет.")             return True          print(f"Найдено постов для удаления: {len(posts)}")          deleted_count = 0         error_count = 0          for i, post in enumerate(posts, 1):             print(f"Удаляю пост {i}/{len(posts)}...")              # Что бы удалить пост нужен его ID и owner_id             response = self._make_api_request('wall.delete', {                 'owner_id': self.user_id,                 'post_id': post['id']             })              if response and 'response' in response and response['response'] == 1:                 deleted_count += 1                 print(f"✓ Пост {post['id']} удален")             else:                 error_count += 1                 print(f"✗ Ошибка удаления поста {post['id']}")                 if 'error' in response:                     print(f"   Код ошибки: {response['error']['error_code']}")                     print(f"   Сообщение: {response['error']['error_msg']}")              # Задержка что б ВК не блочил             time.sleep(1)          print(f"\nРезультат удаления:")         print(f"Успешно удалено: {deleted_count}")         print(f"Ошибок: {error_count}")          return error_count == 0      def delete_posts_with_confirmation(self):         """Удаление постов с подробным подтверждением"""         if not self.access_token:             print("Сначала выполните авторизацию!")             return False          posts = self.get_all_posts()         if not posts:             print("Постов для удаления нет.")             return True          print(f"\nНайдено постов: {len(posts)}")         print("Первые 5 постов для примера:")          # Показываем примеры постов         for i, post in enumerate(posts[:5], 1):             post_date = time.strftime('%Y-%m-%d', time.localtime(post['date']))             text_preview = post['text'][:100] + "..." if len(post['text']) > 100 else post['text']             print(f"{i}. [{post_date}] {text_preview}")          print(f"\n⚠️  ВНИМАНИЕ! БУДУТ УДАЛЕНЫ ВСЕ {len(posts)} ПОСТОВ! ⚠️")         print("Это действие нельзя отменить!")          # Двойное подтверждение         confirmation1 = input("\nВведите 'DELETE MY POSTS' для подтверждения: ")         if confirmation1 != "DELETE MY POSTS":             print("Операция отменена.")             return False          confirmation2 = input("Введите 'YES I AM SURE' для окончательного подтверждения: ")         if confirmation2 != "YES I AM SURE":             print("Операция отменена.")             return False          return self._delete_posts(posts)      def _delete_posts(self, posts):         """Внутренняя функция удаления постов"""         deleted_count = 0         error_count = 0          for i, post in enumerate(posts, 1):             print(f"Удаляю пост {i}/{len(posts)} (ID: {post['id']})...")              response = self._make_api_request('wall.delete', {                 'owner_id': self.user_id,                 'post_id': post['id']             })              if response and 'response' in response and response['response'] == 1:                 deleted_count += 1             else:                 error_count += 1                 print(f"   Ошибка при удалении поста {post['id']}")              time.sleep(1)          print(f"\nУдаление завершено:")         print(f"Успешно: {deleted_count}")         print(f"Ошибок: {error_count}")          return error_count == 0      def extract_text_from_posts(self, posts):         """Извлечение текста из постов"""         texts = []          for post in posts:             if 'text' in post and post['text'].strip():                 post_date = time.strftime('%Y-%m-%d %H:%M:%S',                                           time.localtime(post['date']))                 texts.append(f"Дата: {post_date}\nТекст: {post['text']}\n{'-' * 50}")          return texts      def save_posts_to_file(self, texts, filename='vk_posts.txt'):         """Сохранение постов в файл"""         with open(filename, 'w', encoding='utf-8') as f:             f.write(f"Всего постов: {len(texts)}\n")             f.write("=" * 60 + "\n\n")             for text in texts:                 f.write(text + "\n\n")          print(f"Посты сохранены в файл: {filename}")   def main():     print("VK Parser - Управление постами")     print("=" * 40)      login = input("Введите логин VK: ").strip()     password = input("Введите пароль VK: ").strip()      parser = VKParser(login, password)      if parser.login_vk():         while True:             print("\nВыберите действие:")             print("1 - Получить все посты и сохранить в файл")             print("2 - Просмотреть количество постов")             print("3 - Удалить все посты (БЫСТРОЕ ПОДТВЕРЖДЕНИЕ)")             print("4 - Удалить все посты (ПОДРОБНОЕ ПОДТВЕРЖДЕНИЕ)")             print("0 - Выход")              choice = input("Ваш выбор: ").strip()              if choice == "1":                 posts = parser.get_all_posts()                 if posts:                     post_texts = parser.extract_text_from_posts(posts)                     parser.save_posts_to_file(post_texts)                     print(f"Сохранено {len(post_texts)} постов")                 else:                     print("Постов нет или ошибка получения")              elif choice == "2":                 posts = parser.get_all_posts()                 print(f"Всего постов на стене: {len(posts)}")              elif choice == "3":                 if parser.delete_all_posts():                     print("Удаление завершено успешно")                 else:                     print("Удаление завершено с ошибками")              elif choice == "4":                 if parser.delete_posts_with_confirmation():                     print("Удаление завершено успешно")                 else:                     print("Удаление завершено с ошибками")              elif choice == "0":                 print("Выход...")                 break              else:                 print("Неверный выбор")   if __name__ == "__main__":     main()


ссылка на оригинал статьи https://habr.com/ru/articles/939472/


Комментарии

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *