最近、友人から、ウェブサイトのRSSフィードからニュースをTelegramチャンネルにインポートするボットを書くように頼まれました。 この通知方法の最大の利点は、自分のデバイスで購読しているすべてのユーザーにプッシュ通知が届くことです。 長い間、私は似たようなことをしたかった。 考え直すことなく、サンプルとしてHabraチャネルtelegram.me/habr_ruを選択しました。 Pythonがプログラミング言語として選択されました。
その結果、次の問題を解決する必要がありました。
- RSSの解析。
- 条件の1つは、メッセージの投稿の延期でした（ニュースが投稿された後、n時間以内に非表示/削除/名前変更された場合は、正しいニュースに関する通知が送信される代わりに公開されません）
- 電報でメッセージを投稿する。
- bit.lyサービスによるターゲットリンクの短縮
私は自分から追加しました：
- ライブラリを使用したロギング（ロギング）。
- 構成処理（configparser）。
1.メッセージの延期された投稿
この問題を解決するために、SQLiteデータベースを使用することが決定されました。 データベースを操作するために、 SQLalchemyライブラリが使用されました 。
平凡な構造は単純です-ただ1つのテーブル。 オブジェクトコードを以下に示します。
class News(Base): __tablename__ = 'news' id = Column(Integer, primary_key=True) # text = Column(String) # (), link = Column(String) # . date = Column(Integer) # . . UNIX_TIME. publish = Column(Integer) # . . UNIX_TIME. chat_id = Column(Integer) # . , message_id = Column(Integer) # . . def __init__(self, text, link, date, publish=0,chat_id=0,message_id=0): self.link = link self.text = text self.date = date self.publish = publish self.chat_id = chat_id self.message_id = message_id def _keys(self): return (self.text, self.link) def __eq__(self, other): return self._keys() == other._keys() def __hash__(self): return hash(self._keys()) def __repr__(self): return "<News ('%s','%s', %s)>" % (base64.b64decode(self.text).decode(),\ base64.b64decode(self.link).decode(),\ datetime.fromtimestamp(self.publish)) #
Base64はテキスト情報とリンクの保存に使用され 、Unixタイムスタンプが日時形式として選択されました。
セッションデータは別のクラスで処理されます。
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
メソッドが呼び出されると、rssフィードの最後の30の投稿からNewsクラスのオブジェクトのリストが形成されます。
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=%s&longUrl=%s&format=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のみがintメソッドに渡されます。 短縮リンクが失敗した場合、 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) & set(posts_from_db)) for_publishing.reverse() # for post in for_publishing: text = '%s %s' % (base64.b64decode(post.text).decode('utf8'),\ self.bit_ly.short_link(base64.b64decode(post.link).decode('utf-8'))) a = self.bot.sendMessage(chat_id=self.chat_id, text=text, parse_mode=telegram.ParseMode.HTML) message_id = a.message_id chat_id = a['chat']['id'] self.db.update(post.link, chat_id, message_id) logging.info( u'Public: %s;%s;' % (post, message_id)) time.sleep(self.delay_between_messages)
configparserライブラリを使用して初期化するとき、設定ファイルを読み取り、ロギングを設定します。
ニュースを検出するには、 detect
メソッドを使用しdetect
。 最新の30の投稿を取得し、データベース内のリンクの可用性を1つずつ確認します。
公開する前に、rssチャネルのデータベースからアップロードされた投稿を確認する必要があります。 多くの人がこれを手伝ってくれます。 その後、 電報ライブラリを使用してニュースをすでに公開しています。 その機能は非常に広く、ボットの作成に重点を置いています。 公開後、 message_id
とchat_id
を更新する必要があります。
その結果、以下が得られます。
rssクラスを書き換えると、他のソース（VK、facebookなど）からニュースをインポートすることもできます。
ソースはGithubにあります： https : //github.com/Vispiano/rss2telegram
UPD：はい、誤って「印刷」されたように見えるのはひどく、CamelCaseにないクラス名の方が優れています。