简易信息聚合(也叫聚合内容)是一种RSS基于XML标准,在互联网上被广泛采用的内容包装和投递协议。RSS(Really Simple Syndication)是一种描述和同步网站内容的格式,是使用最广泛的XML应用。RSS搭建了信息迅速传播的一个技术平台,使得每个人都成为潜在的信息提供者。

电报群:https://t.me/joinchat/FZtb9xYKHROjqsIlzovisA
# -*- coding: utf-8 -*-
# telegram_bot_feed.py

import logging
import json
import feedparser
import html2text

g_feed_chat_id = -369761555


def file_reader(path, mode):
    try:
        with open(path, mode) as target_file:
            return json.load(target_file)
    except IOError as io_ex:
        logging.error("Failed to open %s error : " % (path, io_ex))


def modify_date_title(file_name, object_name, p_date_title):
    try:
        with open(file_name, 'r+') as data_file:
            items = json.load(data_file)
            for name, data in items.items():
                if name == object_name:
                    data['date_title'] = p_date_title
                    data_file.seek(0)
                    json.dump(items, data_file, indent=2, ensure_ascii=False)
                    data_file.truncate()
    except IOError as io_ex:
        logging.error("Failed to file_name %s error : " % (file_name, io_ex))


def get_feed_item(feed_data):
    try:
        logging.debug("feed_to_md feed_data : %s" % feed_data)
        feed_content = feedparser.parse(feed_data['url'])
        if feed_content and len(feed_content['entries']) > 0:
            first_post = feed_content['entries'][0]
            title = first_post['title']
            summary = first_post['summary']
            post_date = first_post['published']
            link = first_post['link']
            h = html2text.HTML2Text()
            h.ignore_images = True
            h.ignore_links = True
            summary = h.handle(summary)
            return {'title': title, 'summary': summary, 'url': link, 'post_date': post_date}
    except BaseException as ex:
        logging.error("feed_to_md ex : %s" % ex)


def check_feeds(bot, job):
    try:
        logging.debug('Checking Feeds...')
        feeds = file_reader('feeds.json', 'r')
        for name, feed_data in feeds.items():
            result = get_feed_item(feed_data)
            logging.debug('result title : %s' % result['title'])
            if feed_data['date_title'] != result['title']:
                modify_date_title('feeds.json', name, result['title'])
                rss_msg = '[%s](%s)' % (result['title'], result['url'])
                logging.info('rss_msg : %s' % rss_msg)
                bot.send_message(chat_id=g_feed_chat_id, text=rss_msg, parse_mode='Markdown')
        logging.debug('job %s Sleeping for 30 mins...' % job)
    except BaseException as ex:
        logging.error("check_feeds : %s" % ex)


def start_feed(job):
    logging.info('start_feed')
    job.run_repeating(check_feeds, interval=30 * 60, first=0)
{
"人民网国内新闻": {
"url": "http://www.people.com.cn/rss/politics.xml",
"date_title": "习近平出席金正恩举行的欢送仪式"
},
"kai_x": {
"url": "https://kaix.in/feed",
"date_title": "第一章:初识咖啡"
}
}
# -*- coding: utf-8 -*-
# telegram_bot_jobs.py

import logging
from telegram.ext import Updater
from datetime import datetime

import telegram_bot_feed


def error(bot, update, err):
    logging.error('bot : %s update : %s error : %s' % (bot, update, err))


def echo(bot, update):
    logging.info("bot : %s chat_id:%d text:%s" % (bot, update.message.chat_id, update.message.text))


def start_bot():
    try:
        start_time = datetime.now()  # 获取当前datetime
        logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)

        # 温顺的莱昂
        updater = Updater(token='739344882:AAF_BMyjc7S45nado1dK5E6sMt-0jYH5VMA')
        dp = updater.dispatcher
        job = updater.job_queue
        dp.add_error_handler(error)

        # #2 电报机器人-RSS订阅
        telegram_bot_feed.start_feed(job)

        updater.start_polling()
        logging.info('telegram bot updater start_polling...')
        updater.idle()

        end_time = datetime.now()  # 获取当前datetime
        logging.info('花费时间:%f秒' % (end_time - start_time).total_seconds())
    except BaseException as start_bot_ex:
        logging.error("telegram_bot_jobs ex : %s" % start_bot_ex)

发表评论

电子邮件地址不会被公开。 必填项已用*标注