Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import asyncpraw
- import datetime
- import logging
- from aiogram import Bot, Dispatcher, executor, types, exceptions
- import timeago
- API_TOKEN = 'Bot API token'
- logging.basicConfig(level=logging.DEBUG, format="%(asctime)s:%(levelname)s:%(message)s",
- handlers=[logging.StreamHandler()])
- bot = Bot(token=API_TOKEN)
- dp = Dispatcher(bot)
- reddit = asyncpraw.Reddit(client_id=client_id, client_secret=client_secret, user_agent=user_agent)
- def time_ago(timestamp):
- return timeago.format(datetime.datetime.fromtimestamp(timestamp), datetime.datetime.now())
- def flair_check(post, flair_name):
- try:
- post_flair = post.link_flair_text.lower()
- except AttributeError:
- post_flair = None
- if post_flair == flair_name:
- return True
- else:
- return False
- async def check_title(message, search_terms, post):
- for search_term in search_terms:
- if search_term in post.title.lower():
- await message.reply(time_ago(post.created_utc) +
- "\n" + post.title + "\n" +
- "https://reddit.com" + post.permalink)
- async def reddit_func(message):
- search_query = message.text.split(',')[1].strip().lower()
- search_terms = search_query.split()
- subred_name = message.text.split(',')[2].strip().lower()
- try:
- flair_name = message.text.split(',')[3].strip().lower()
- except IndexError:
- flair_name = ''
- print(search_query, search_terms, subred_name, flair_name)
- search_flair = None
- if flair_name != '':
- search_flair = flair_name
- await message.reply('You have asked for the terms: ' + search_query + '\n' +
- 'on the subreddit: ' + subred_name + '\n' +
- 'with flair: ' + ('None' if flair_name == '' else flair_name))
- subreddit = await reddit.subreddit(subred_name)
- while True:
- try:
- async for post in subreddit.stream.submissions(skip_existing=False):
- if flair_name == '':
- await check_title(message, search_terms, post)
- else:
- if flair_check(post, flair_name):
- await check_title(message, search_terms, post)
- except exceptions as e:
- await message.reply(str(e))
- continue
- @dp.message_handler(commands=['start', 'help'])
- async def send_welcome(message: types.Message):
- await message.reply('Hello, this will pr)ompt you in case new post matching search terms on a given subreddit.\n'
- 'Send:\n'
- 'Check<comma>search terms (space separated)<comma>subreddit name<comma>flair filter(optional)\n'
- '\nFor example:\ncheck, RAM,hardwareswap,selling')
- @dp.message_handler(text_startswith=['check', 'Check', 'CHECK'])
- async def check_msg(message: types.Message):
- await reddit_func(message)
- if __name__ == '__main__':
- executor.start_polling(dp, skip_updates=True)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement