Задача
Автоматически наполнять ленту предложки сообществ релевантными постами из других пабликов. На входе — список сообществ с поисковым запросом, словами-исключениями и периодом, на выходе — запланированные публикации. Токен один, с правами админа только своих групп.
Как делал
Главная сложность — newsfeed.search отдаёт не больше ~200 результатов на запрос. Период режется на шестичасовые окна, и внутри каждого окна идёт пагинация через offset со сдвигом по 100. Так поиск обходится всей лентой за сутки, а не первыми двумястами.
WINDOW = 6 * 3600
while pos < total_seconds:
chunk = min(WINDOW, total_seconds - pos)
w_end = end_time - pos
windows.append((w_end - chunk, w_end))
pos += chunk
for w_start, w_end in windows:
offset = 0
while offset < 200: # потолок пагинации VK
response = vk_request('newsfeed.search', {
'offset': offset, 'count': 100, 'start_time': w_start, 'end_time': w_end, ...
})Слова-исключения не фильтруются в коде, а уходят прямо в запрос как -слово. Отсев происходит на стороне поиска, до выборки, и лишние посты просто не приходят.
Дедупликация трёхуровневая. Первый уровень — пара owner_id_id: между страницами пагинации VK отдаёт пересечения. Второй — хеш содержимого внутри запуска: один и тот же текст от разных авторов. Третий — seen_hashes.json, история между запусками.
post_id = f"{post.get('owner_id', 0)}_{post.get('id', 0)}"
if post_id in seen_ids: # 1. повтор в рамках пагинации
continue
...
phash = get_post_hash(post)
if phash in skip_hashes: # 3. этот контент уходил в прошлые запуски
continue
if phash in seen_hashes: # 2. тот же контент в этом запуске
continueХеш считается не от ID, а от текста с нормализованными пробелами плюс набор типов вложений с количеством, отсортированный. ID вложений в хеш не входят намеренно: при перепечатке они всегда новые, и хеш по ID ломался бы на каждом репосте.
text = re.sub(r'\s+', ' ', post.get('text', '').strip())
att_counts = {}
for a in post.get('attachments', []):
t = a.get('type')
if t in VALID_TYPES:
att_counts[t] = att_counts.get(t, 0) + 1
att_str = '|'.join(f"{t}:{c}" for t, c in sorted(att_counts.items())) or 'no_atts'
return hashlib.md5(f"{text}|{att_str}".encode('utf-8')).hexdigest()Файл с хешами пишется атомарно — сначала .tmp, потом os.replace(). Плюс сохранение после каждого отправленного поста и регистрация через atexit. Прерывание посреди публикации не даёт ни битого JSON, ни повторного прогона уже ушедших постов.
def save_hashes(hashes):
tmp = HASHES_PATH + '.tmp'
with open(tmp, 'w', encoding='utf-8') as f:
json.dump(sorted(hashes), f, ensure_ascii=False)
os.replace(tmp, HASHES_PATH) # замена целиком, файл не может остаться битымПубликация в предложку планируется с запасом вперёд, и на каждый сдвиг есть три попытки: VK возвращает ошибку, если две публикации попали на одну минуту.
base_time = int(datetime.datetime.now().timestamp()) + 100000 * 60
for attempt in range(3):
publish_date = base_time + (post_index + attempt) * 60
...
if 'already scheduled' in err_msg and attempt < 2:
time.sleep(1)
continueРетраи в vk_request разделены по типу сбоя: rate limit с кодом 6 ждёт пять секунд, HTML вместо JSON или Too Large — десять, таймаут — три, обрыв соединения — линейно с ростом попытки. Невалидный ответ уходит в errors.log с превью первых трёхсот символов, потому что текст ошибки VK сам по себе бесполезен.
Отдельно проверяются лимиты VK: 10 000 символов на пост, 10 вложений и 4 096 символов для личных сообщений — у ЛС лимит ниже, поэтому текст режется с записью в лог. Сборка в один .exe делается PyInstaller под Wine, и путь к конфигу ищется и рядом с бинарником, и в рабочей папке — у sys.frozen домашний каталог всегда один.
Итог
Один файл на 480 строк, requests без обёрток, конфиг в JSONC с комментариями. test_mode переключает отправку из предложки в личку — отладка не засоряет ленту. Windows-сборка запускается без установленного Python.
