Как сделать автопостинг в Телеграмме python

Вы используете устаревший браузер. Этот и другие сайты могут отображаться в нем неправильно.
Необходимо обновить браузер или попробовать использовать другой.

r1dddy4sv

Пользователь
Ноя 26, 2020 3 0 1

import time
import eventlet
import requests
import logging
import telebot
from time import sleep
import os
os.chdir(r’C:UsersuserDesktopPython_scriptsbotpost’)

# Каждый раз получаем по 10 последних записей со стены
URL_VK = ‘https://api.vk.com/method/wall.get. 2389318:AAHV92lau4IMj2QeySdZJQ5lsCGOpTE62I0VK’
FILENAME_VK = r’C:UsersuserDesktopPython_scriptsbotpostlast_known_id.txt’
BASE_POST_URL = ‘https://vk.com/4svtest?w=wall-195312104_’
SINGLE_RUN=True

def get_data():
timeout = eventlet.Timeout(10)
try:
feed = requests.get(URL_VK)
return feed.json()
except eventlet.timeout.Timeout:
logging.warning(‘Got Timeout while retrieving VK JSON data. Cancelling. ‘)

Сделать автопостер в телеграмм канал? Телеграм-бот на python для автопостинга

return None
finally:
timeout.cancel()
def send_new_posts(items, last_id):
for item in items:
if item[‘id’] break
link = ».format(BASE_POST_URL, item[‘id’])
bot.send_message(CHANNEL_NAME, link)
# Спим секунду, чтобы избежать разного рода ошибок и ограничений (на всякий случай!)
time.sleep(1)
return

def check_new_posts_vk():
# Пишем текущее время начала
logging.info(‘[VK] Started scanning for new posts’)
with open(FILENAME_VK, ‘rt’) as file:
last_id = int(file.read())
if last_id is None:
logging.error(‘Could not read from storage. Skipped iteration.’)
return
logging.info(‘Last ID (VK) = ‘.format(last_id))
try:
feed = get_data()
# Если ранее случился таймаут, пропускаем итерацию. Если всё нормально — парсим посты.
if feed is not None:
entries = feed[‘response’][1:]
try:
# Если пост был закреплен, пропускаем его
tmp = entries[0][‘is_pinned’]
# И запускаем отправку сообщений
send_new_posts(entries[1:], last_id)
except KeyError:
send_new_posts(entries, last_id)
# Записываем новый last_id в файл.
with open(FILENAME_VK, ‘wt’) as file:
try:
tmp = entries[0][‘is_pinned’]
# Если первый пост — закрепленный, то сохраняем ID второго
file.write(str(entries[1][‘id’]))
logging.info(‘New last_id (VK) is ‘.format((entries[1][‘id’])))
except KeyError:
file.write(str(entries[0][‘id’]))
logging.info(‘New last_id (VK) is ‘.format((entries[0][‘id’])))
except Exception as ex:
logging.error(‘Exception of type in check_new_post(): ‘.format(type(ex).__name__, str(ex)))
pass
logging.info(‘[VK] Finished scanning’)
return

if __name__ == ‘__main__’:
# Избавляемся от спама в логах от библиотеки requests
logging.getLogger(‘requests’).setLevel(logging.CRITICAL)

Как создать телеграм бота с автопостингом в каналы ⚡️ Telegram bot на языке Python ⚡️

# Настраиваем наш логгер
logging.basicConfig(format='[%(asctime)s] %(filename)s:%(lineno)d %(levelname)s — %(message)s’, level=logging.INFO,
filename=’bot_log.log’, datefmt=’%d.%m.%Y %H:%M:%S’)
if not SINGLE_RUN:
while True:
check_new_posts_vk()
# Пауза в 4 минуты перед повторной проверкой
logging.info(‘[App] Script went to sleep.’)
time.sleep(60 * 4)
else:
check_new_posts_vk()
logging.info(‘[App] Script exited.n’)

ошибка: Exception of type KeyError in check_new_post(): ‘response’

не совсем понимаю что нужно исправить и как.
подскажите пожалуйста.
Спасибо.

Источник: itfy.org

TG_AutoPoster

Бот, пересылающий записи из групп ВК в канал/чат/ЛС в Telegram.

Установка (обновление)

pip3 install -U TG-AutoPoster

Настройка

1. Создайте файл config.yaml , скопируйте в него содержимое файла config.yaml.example и выполните настройку ключа vk

1. Получите ваши api_id и api_hash на https://my.telegram.org/apps и настройте ключ telegram (подробнее об Telegram API Keys здесь)

1. Если необходимо, настройте использование SOCKS5 прокси, добавив ключ proxy со следующим содержимым:

Запуск

1. Для запуска используйте TG_AutoPoster.sh или команду python3 -m TG_AutoPoster
2. Активируйте бота командой /start

Автопостинг рекомендуется настраивать через чат с ботом. Подробнее можно узнать, отправив боту команду /help

Для доступных параметров командой строки используйте bash TG_AutoPoster.sh —help или python3 -m TG_AutoPoster —help

По умолчанию бот проверяет группы на наличие новых постов раз в час. Если необходимо изменить период проверки постов запустите бота с параметром командной строки —sleep N , где N — значение в секундах. Не рекомендуется устанавливать малое значение, так как это может привести к заморозке страницы ВК (подробнее).

Описание настроек группы

Параметр Описание

channel	Список ID каналов/чатов в Telegram, разделенных пробелом, в которые отправлять посты из групп ВК
last_id (необязательно)	ID последнего отправленного поста. Если параметр отсутствует, он будет добавлен автоматически со значением 0
pinned_id (необязательно)	ID закреплённого поста.
last_story_id (необязательно)	ID последней отправленной истории.
use_long_poll (необязательно)	Использовать Long Poll API для получения постов из своей группы (или в которой вы являетесь администратором) в режиме реального времени. Чтобы использовать Long Poll API, откройте раздел «Управление сообществом», на вкладке «Работа с API»→«Long Poll API» выберите «Включено», «Версия API»: 5.131. Также необходимо включить тип события «Записи на стене»: «Добавление» на вкладке «Типы событий». Значение параметра по умолчанию: false

Описание настроек автопостинга (ключ settings )

Параметр Возможные значения Описание

sign_posts	true, false	Указывать ли автора поста (если это возможно) и ссылку на оригинальный пост. По умолчанию: true
send_reposts	false, post_only, true	Отправлять ли репосты? Подробнее в config.yaml.example. По умолчанию отправка репостов отключена.
send_stories	false, true	Отправлять ли истории? По умолчанию: false
what_to_send	all, text, link, photo, doc, video, music, polls	Какие типы вложений отправлять. Подробнее в config.yaml.example. По умолчанию отправляются все вложения.
stop_list	Абсолютный путь к файлу, содержащий стоп-слова (в файле должно быть по одному слову на каждой строке). Если вы не хотите использовать стоп-слова удалите этот параметр из файла конфигурации
blacklist	Абсолютный путь к файлу, содержащий слова, которые будут удалены из текста отправляемого поста. Поддерживаются регулярные выражения.
disable_notification	true, false	Отправляет сообщения молча. Пользователи получат уведомление без звука. По умолчанию: false
disable_web_page_preview	true, false	Отключить предпросмотр ссылок в сообщениях. По умолчанию: true
posts_count	Количество отправляемых ботом новых постов за раз. По умолчанию 11.
header	Текст с форматированием Markdown	Текст, который будет добавлен в начало сообщения
footer	Текст с форматированием Markdown	Текст, который будет добавлен в конец сообщения

Все параметры ключа settings могут быть заданы индивидуально для каждой группы

Отчеты об ошибках и предложения отправлять в:

Для пожертвований на развитие проекта:

1. Qiwi
2. Bitcoin: 1H1UVnXgvcLo3RWmxuYmi7b16ADo6XBWw5
3. TON: EQD42Z5d8d1gT1uSpKTAaLYHlQ95vdMXrlNlYMpSFpQawwuY

Источник: pypi.org

Пишем свой канал-бот для Telegram как у Хабра на Python

Недавно ко мне обратился друг с просьбой написать бота, импортирующего новости из RSS-канала на сайте в Telegram-канал. Огромнейшим плюсом данного способа оповещения являются push-уведомления, которые приходят каждому подписанному пользователю на его устройство. Уже давно хотелось заняться чем-то подобным. Недолго думая, в качестве образца я выбрал канал Хабра telegram.me/habr_ru. В качестве языка программирования был выбран Python.

В итоге, мне надо было решить следующие проблемы:

1. Парсинг RSS.
2. Одним из условий был отложенный постинг сообщений (если после того, как новость была выложена, в течение n часов её скрыли/удалили/переименовали, то она не должна быть опубликована, вместо нее отправляется оповещение о корректной новости)
3. Постинг сообщений в телеграм.
4. Сокращение целевой ссылки с помощью сервиса bit.ly

От себя добавил еще:

1. Ведение логов с помощью библиотеки (logging).
2. Обработка конфига (configparser).

1. Отложенный постинг сообщений

Для решения данной проблемы было принято решение использовать SQLite базу данных. Для работы с БД использовалась библиотека SQLalchemy.

Структура до банального проста — всего одна таблица. Код объекта представлен ниже:

Для хранения текстовой информации и ссылок использется base64, форматом хранения даты-времени был выбран Unix Timestamp.

Обработка данных сессии осуществляется отдельным классом.

Base = declarative_base() class Database: «»» Класс для обработки сессии SQLAlchemy. Также включает в себя минимальный набор методов, вызываемых в управляющем классе. Названия методов говорящие. «»» def __init__(self, obj): engine = create_engine(obj, echo=False) Session = sessionmaker(bind=engine) self.session = Session() def add_news(self, news): self.session.add(news) self.session.commit() def get_post_without_message_id(self): return self.session.query(News).filter(and_(News.message_id == 0, News.publish<=int(time.mktime(time.localtime())))).all() def update(self, link, chat, msg_id): self.session.query(News).filter_by(link = link).update(<«chat_id»:chat, «message_id»:msg_id>) self.session.commit() def find_link(self,link): if self.session.query(News).filter_by(link = link).first(): return True else: return False

При обнаружении новости, она добавляется в базу. Сразу же задается время публикации.

Для обнаружения новостей готовых к публикации используется метод get_post_withwithout_message_id . Фактически, мы выбираем из базы все посты, у которых message_id=0 и дата публикации меньше текущего времени.

Для проверки на новизну отправляем запрос базе данных на факт содержания ссылки на новость (метод find_link ).

Метод update служит для обновления данных, после публикации новости в канале.

2. Парсинг RSS

Стоит признаться, что писать свой RSS парсер совсем не хотелось, поэтому в бой вступила библиотека feedparser.

import feedparser class Source(object): def __init__(self, link): self.link = link self.news = [] self.refresh() def refresh(self): data = feedparser.parse(self.link) self.news = [News(binascii.b2a_base64(i[‘title’].encode()).decode(), binascii.b2a_base64(i[‘link’].encode()).decode(), int(time.mktime(i[‘published_parsed’]))) for i in data[‘entries’]]

Код до смешного прост. При вызове метода refresh с помощью генератора формируется список объектов класса News из последних 30 размещенных постов в rss ленте.

3. Сокращение ссылок

Как упоминалось выше, в качестве сервиса был выбран bit.ly. API не вызвает лишних вопросов.

class Bitly: def __init__(self,access_token): self.access_token = access_token def short_link(self, long_link): url = ‘https://api-ssl.bitly.com/v3/shorten?access_token=%sformat=json’ % (self.access_token, long_link) try: return json.loads(urllib.request.urlopen(url).read().decode(‘utf8’))[‘data’][‘url’] except: return long_link

В инит метод передается только наш access_token. В случае неудачного получения сокращенной ссылки, метод short_link возвращает переданную ему изначальную ссылку.

4. Управляющий класс

class ExportBot: def __init__(self): config = configparser.ConfigParser() config.read(‘./config’) log_file = config[‘Export_params’][‘log_file’] self.pub_pause = int(config[‘Export_params’][‘pub_pause’]) self.delay_between_messages = int(config[‘Export_params’][‘delay_between_messages’]) logging.basicConfig(format = u’%(filename)s[LINE:%(lineno)d]# %(levelname)-8s [%(asctime)s] %(message)s’,level = logging.INFO, filename = u’%s’%log_file) self.db = database(config[‘Database’][‘Path’]) self.src = source(config[‘RSS’][‘link’]) self.chat_id = config[‘Telegram’][‘chat’] bot_access_token = config[‘Telegram’][‘access_token’] self.bot = telegram.Bot(token=bot_access_token) self.bit_ly = bitly(config[‘Bitly’][‘access_token’]) def detect(self): #получаем 30 последних постов из rss-канала self.src.refresh() news = self.src.news news.reverse() #Проверяем на наличие в базе ссылки на новость. Если нет, то добавляем в базу данных с #отложенной публикацией for i in news: if not self.db.find_link(i.link): now = int(time.mktime(time.localtime())) i.publish = now + self.pub_pause logging.info( u’Detect news: %s’ % i) self.db.add_news(i) def public_posts(self): #Получаем 30 последних записей из rss канала и новости из БД, у которых message_id=0 posts_from_db = self.db.get_post_without_message_id() self.src.refresh() line = [i for i in self.src.news] #Выбор пересечний этих списков for_publishing = list(set(line) %s;’ % (post, message_id)) time.sleep(self.delay_between_messages)

При инциализации с помощью библиотеки configparser считываем наш конфиг-файл и настраиваем логгирование.

Чтобы детектировать новости, используем метод detect . Получаем последние 30 опубликованных постов, поочередно проверяем наличие ссылки в базе данных.

Перед публикацией, необходимо проверить наличие постов, выгруженных из базы данных в rss-канале. В этом нам помогут множества. И после этого уже публикуем новость с помощью библиотеки telegram. Её функционал довольно широк и ориентирован на написание ботов. После публикации необходимо обновить message_id и chat_id .

В итоге получаем:

Стоит отметить то, что если переписать класс rss, то так же можно будет импортировать новости из других источников (VK, facebook и т.д.).

UPD: Да, случайно забравшийся «print» выглядит ужасно и названия классов не в CamelCase не лучше.

Источник: habr.com