В этой статье разберем создание 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/
Добавить комментарий