Add DB-first channel support and live send tests

README.md

@@ -57,3 +57,12 @@ But if you are geek enough then install mongodb, ffmpeg, python and setup cron:
 46 * * * * ~/reddit2telegram/auto_update.sh
 * * * * * ~/reddit2telegram/reddit2telegram/cron_job.sh
 ```
+
+Tests
+-----
+
+Live integration tests send real messages to `@r_channels_test` and use Reddit API.
+
+```bash
+R2T_LIVE_TESTS=1 /root/reddit2telegram/.venv/bin/python -m unittest tests/test_live_send.py
+```

reddit2telegram/channels/reddit2telegram/app.py

@@ -2,7 +2,6 @@
 
 import random
 from datetime import datetime
-import os
 import hashlib
 
 import pymongo
@@ -99,12 +98,10 @@ def what_channel(submodule_name_to_promte):
 
 
 def get_tags(submodule_name_to_promte):
-    tags_filename = os.path.join('channels', submodule_name_to_promte, 'tags.txt')
-    if not os.path.exists(tags_filename):
+    tags_string = utils.channels_stuff.get_tags_for_submodule(submodule_name_to_promte)
+    if not tags_string:
         return None
-    with open(tags_filename, 'r') as tags_file:
-        tags = tags_file.read()
-        return tags.split()
+    return tags_string.split()
 
 
 def make_nice_submission(submission, r2t, submodule_name_to_promte, extra_ending=None, **kwargs):

reddit2telegram/channels/tech_receiver/app.py

@@ -35,7 +35,7 @@ def send_post(submission, r2t):
             'last_update': 0
         })
 
-    updates = r2t.telegram_bot.get_updates(offset=last_update_doc['last_update'])
+    updates = r2t.get_updates(offset=last_update_doc['last_update'])
 
     last_update = 0
     for update in updates:
@@ -59,7 +59,7 @@ def send_post(submission, r2t):
             continue
 
         message_id = update['message']['message_id']
-        r2t.telegram_bot.forward_message(chat_id=get_dev_channel(), from_chat_id=user_id, message_id=message_id)
+        r2t.forward_message(chat_id=get_dev_channel(), from_chat_id=user_id, message_id=message_id)
         if int(update['message']['chat']['id']) == int(config['telegram']['papa']):
             # print('>>>>>>>>>>>>>>>>>^^^^^^^^^^^^^^')
             text = update['message']['text']

reddit2telegram/channels/tech_store_stat/app.py

@@ -177,7 +177,7 @@ def send_post(submission, r2t):
             logging.error(err_to_send)
         short_sleep(sleep_coef)
         try:
-            current_members_cnt = r2t.telegram_bot.get_chat_members_count(chat_id=channel_name)
+            current_members_cnt = r2t.get_chat_members_count(chat_id=channel_name)
             stat_to_store['members_cnt'] = current_members_cnt
             total['members'] += current_members_cnt
             prev_members_cnt = get_last_members_cnt(r2t, channel_name)

reddit2telegram/reporting_stuff.py

@@ -11,7 +11,8 @@
 import utils
 
 
-with open(os.path.join('configs', 'prod.yml')) as config_file:
+CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'configs', 'prod.yml')
+with open(CONFIG_PATH) as config_file:
     config = yaml.safe_load(config_file.read())
 
 

reddit2telegram/utils/__init__.py

@@ -13,14 +13,16 @@
 import logging
 import enum
 import subprocess
+import asyncio
 
 from imgurpython import ImgurClient
 import yaml
 import pymongo
 from pymongo.collection import ReturnDocument
-import telegram
 from telegram.error import TelegramError, BadRequest
-from telegram import ParseMode
+from telegram import Bot, InputMediaPhoto, InputMediaVideo
+from telegram.constants import ParseMode
+from telegram.request import HTTPXRequest
 
 from utils.tech import short_sleep, long_sleep
 
@@ -298,13 +300,26 @@ def __init__(self, t_channel=None, config=None):
             with open(os.path.join('configs', 'prod.yml')) as f:
                 config = yaml.safe_load(f.read())
         self.config = config
-        self.telegram_bot = telegram.Bot(self.config['telegram']['token'])
+        request = HTTPXRequest(connection_pool_size=8, pool_timeout=30)
+        self.telegram_bot = Bot(self.config['telegram']['token'], request=request)
+        self._loop = asyncio.new_event_loop()
         if t_channel is None:
             t_channel = '@r_channels_test'
         self.t_channel = t_channel
         self._make_mongo_connections()
         short_sleep()
 
+    def _run_async(self, coro):
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        if loop and loop.is_running():
+            return asyncio.run_coroutine_threadsafe(coro, loop).result()
+        if self._loop.is_closed():
+            self._loop = asyncio.new_event_loop()
+        return self._loop.run_until_complete(coro)
+
     def _make_mongo_connections(self):
         self.stats = pymongo.MongoClient(host=self.config['db']['host'])[self.config['db']['name']]['stats']
         self.urls = pymongo.MongoClient(host=self.config['db']['host'])[self.config['db']['name']]['urls']
@@ -434,11 +449,11 @@ def send_gif(self, url, text, parse_mode=None):
         if len(text) > TELEGRAM_CAPTION_LIMIT:
             text, next_text = self._split_1024(text)
         try:
-            self.telegram_bot.send_document(chat_id=self.t_channel,
+            self._run_async(self.telegram_bot.send_document(chat_id=self.t_channel,
                 document=url,
                 caption=text,
                 parse_mode=parse_mode
-            )
+            ))
         except BadRequest as e:
             logging.info('Unknown error.')
             return SupplyResult.SKIP_FOR_NOW
@@ -479,7 +494,7 @@ def send_video(self, url, text, parse_mode=None):
         if len(text) > TELEGRAM_CAPTION_LIMIT:
             text, next_text = self._split_1024(text)
         f = open(video_with_audio_filename, 'rb')
-        self.telegram_bot.send_video(chat_id=self.t_channel, video=f, caption=text, parse_mode=parse_mode)
+        self._run_async(self.telegram_bot.send_video(chat_id=self.t_channel, video=f, caption=text, parse_mode=parse_mode))
         f.close()
         if len(next_text) > 1:
             short_sleep()
@@ -497,11 +512,11 @@ def send_img(self, url, text, parse_mode=None):
             logging.info(f'Long pic in {self.t_channel}.')
             return self._send_img_as_link(url, text)
         try:
-            self.telegram_bot.send_photo(chat_id=self.t_channel,
+            self._run_async(self.telegram_bot.send_photo(chat_id=self.t_channel,
                 photo=url,
                 caption=text,
                 parse_mode=parse_mode
-            )
+            ))
             return SupplyResult.SUCCESSFULLY
         except TelegramError as e:
             logging.info(f'TelegramError prevented at {self.t_channel}.')
@@ -510,30 +525,30 @@ def send_img(self, url, text, parse_mode=None):
 
     def send_text(self, text, disable_web_page_preview=False, parse_mode=None):
         if len(text) < 4096:
-            self.telegram_bot.send_message(chat_id=self.t_channel,
+            self._run_async(self.telegram_bot.send_message(chat_id=self.t_channel,
                                             text=text,
                                             disable_web_page_preview=disable_web_page_preview,
-                                            parse_mode=parse_mode)
+                                            parse_mode=parse_mode))
             return SupplyResult.SUCCESSFULLY
         # If text is longer than 4096 symbols.
         next_text = text
         while len(next_text) > 0:
             list_of_words = next_text.split(' ')
             if len(list_of_words[0]) > 4096:
                 new_text, next_text = self._split_4096(next_text)
-                self.telegram_bot.send_message(chat_id=self.t_channel,
+                self._run_async(self.telegram_bot.send_message(chat_id=self.t_channel,
                                                 text=new_text,
                                                 disable_web_page_preview=disable_web_page_preview,
-                                                parse_mode=parse_mode)
+                                                parse_mode=parse_mode))
             elif len(list_of_words[0]) <= 4096:
                 # If first word is less than 4096.
                 words_to_send = list()
                 while (len(list_of_words) > 0) and (sum([len(x) for x in words_to_send]) + len(words_to_send) + len(list_of_words[0]) <= 4096):
                     words_to_send.append(list_of_words.pop(0))
-                self.telegram_bot.send_message(chat_id=self.t_channel,
+                self._run_async(self.telegram_bot.send_message(chat_id=self.t_channel,
                                                 text=' '.join(words_to_send),
                                                 disable_web_page_preview=disable_web_page_preview,
-                                                parse_mode=parse_mode)
+                                                parse_mode=parse_mode))
                 next_text = ' '.join(list_of_words)
             short_sleep()
         return SupplyResult.SUCCESSFULLY
@@ -568,17 +583,17 @@ def send_gallery(self, dict_of_dicts_of_pics, text):
 
             for item in sorted(dict_of_pics.items(), key=lambda item: item[0]):
                 if item[1]['type'] == 'pic':
-                    list_of_items_in_one_group.append(telegram.InputMediaPhoto(item[1]['url']))
+                    list_of_items_in_one_group.append(InputMediaPhoto(item[1]['url']))
                 elif item[1]['type'] == 'video':
-                    list_of_items_in_one_group.append(telegram.InputMediaVideo(item[1]['url']))
+                    list_of_items_in_one_group.append(InputMediaVideo(item[1]['url']))
                 else:
                     logging.error('Unkown item in gallery.')
                     return SupplyResult.SKIP_FOR_NOW 
 
             try:
-                self.telegram_bot.send_media_group(chat_id=self.t_channel,
+                self._run_async(self.telegram_bot.send_media_group(chat_id=self.t_channel,
                                                 media=list_of_items_in_one_group,
-                                                timeout=66)
+                                                timeout=66))
                 logging.info('Successful gallery sent.')
             except Exception as e:
                 logging.error('Gallery sent failed.')
@@ -589,6 +604,22 @@ def send_gallery(self, dict_of_dicts_of_pics, text):
     def forward_last_message_from_the_channel(self, from_channel_name):
         pass
 
+    def get_chat_administrators(self, chat_id):
+        return self._run_async(self.telegram_bot.get_chat_administrators(chat_id=chat_id))
+
+    def get_updates(self, **kwargs):
+        return self._run_async(self.telegram_bot.get_updates(**kwargs))
+
+    def forward_message(self, chat_id, from_chat_id, message_id):
+        return self._run_async(self.telegram_bot.forward_message(
+            chat_id=chat_id,
+            from_chat_id=from_chat_id,
+            message_id=message_id
+        ))
+
+    def get_chat_members_count(self, chat_id):
+        return self._run_async(self.telegram_bot.get_chat_members_count(chat_id=chat_id))
+
     def send_simple(self, submission, **kwargs):
         '''
         Universal send method for most of the channels.

reddit2telegram/utils/channels_stuff.py

@@ -2,12 +2,14 @@
 
 import os
 import importlib
+import re
 
 import pymongo
 import yaml
 
 
 CHANNELS_COLLECTION = 'channels'
+_SIMPLE_SEND_RE = re.compile(r'^\s*return\s+r2t\.send_simple\(submission\)\s*$')
 
 
 def get_config(config_filename=None):
@@ -17,19 +19,69 @@ def get_config(config_filename=None):
         return yaml.safe_load(config_file.read())
 
 
+def get_db(config_filename=None):
+    config = get_config(config_filename=config_filename)
+    return pymongo.MongoClient(host=config['db']['host'])[config['db']['name']]
+
+
+def _get_channels_collection(config_filename=None):
+    db = get_db(config_filename=config_filename)
+    return db[CHANNELS_COLLECTION]
+
+
+def get_channel_doc(submodule_name, config_filename=None):
+    channels = _get_channels_collection(config_filename=config_filename)
+    return channels.find_one({'submodule': submodule_name.lower()})
+
+
+def _file_based_overrides(config):
+    channels_config = config.get('channels', {})
+    file_based = channels_config.get('file_based', [])
+    return set(name.lower() for name in file_based)
+
+
+def is_simple_channel_module(submodule_name):
+    app_path = os.path.join('channels', submodule_name, 'app.py')
+    if not os.path.isfile(app_path):
+        return False
+    with open(app_path, 'r') as app_file:
+        code = app_file.read()
+    lines = [
+        line.strip()
+        for line in code.splitlines()
+        if line.strip() and not line.strip().startswith('#')
+    ]
+    has_simple_send = any(_SIMPLE_SEND_RE.match(line) for line in lines)
+    if not has_simple_send:
+        return False
+    for line in lines:
+        if 'r2t.send_simple' in line and not _SIMPLE_SEND_RE.match(line):
+            return False
+    if sum(1 for line in lines if line.startswith('def ')) > 1:
+        return False
+    return True
+
+
 def import_submodule(submodule_name):
-    if os.path.isdir(os.path.join('channels', submodule_name)):
-        submodule = importlib.import_module(f'channels.{submodule_name}.app')
-    else:
-        submodule = DefaultChannel(submodule_name)
-    return submodule
+    config = get_config()
+    submodule_name = submodule_name.lower()
+    channel_dir = os.path.join('channels', submodule_name)
+    has_module = os.path.isdir(channel_dir)
+    has_db = get_channel_doc(submodule_name) is not None
+    force_file = submodule_name in _file_based_overrides(config)
+
+    if force_file and has_module:
+        return importlib.import_module(f'channels.{submodule_name}.app')
+    if has_db and (not has_module or is_simple_channel_module(submodule_name)):
+        return DefaultChannel(submodule_name)
+    if has_module:
+        return importlib.import_module(f'channels.{submodule_name}.app')
+    return DefaultChannel(submodule_name)
 
 
 def set_new_channel(channel, **kwargs):
     channel = channel.replace('@', '')
-    config = get_config()
-    db = pymongo.MongoClient(host=config['db']['host'])[config['db']['name']]
-    channels = db[CHANNELS_COLLECTION]
+    channels = _get_channels_collection()
     is_any = channels.find_one({'submodule': channel.lower()})
     if is_any is not None:
         return
@@ -49,7 +101,7 @@ class DefaultChannel(object):
     '''docstring for DefaultChannel'''
     def __init__(self, submodule):
         super(DefaultChannel, self).__init__()
-        self.submodule = submodule
+        self.submodule = submodule.lower()
         self.get_settings_from_db()
         if self.content is None:
             self.content = dict(
@@ -71,13 +123,10 @@ def __init__(self, submodule):
             self.content['other'] = self.content.get('other', False)
 
     def get_settings_from_db(self):
-        config = get_config()
-        db = pymongo.MongoClient(host=config['db']['host'])[config['db']['name']]
-        channels = db[CHANNELS_COLLECTION]
-        channel_details = channels.find_one({'submodule': self.submodule})
+        channel_details = get_channel_doc(self.submodule)
         if channel_details is None:
             self.t_channel = 'NO CHANNEL FOUND FOR: self.submodule'
-            raise
+            raise ValueError('No channel found in DB for submodule: {}'.format(self.submodule))
         self.t_channel = channel_details.get('channel', None)
         self.submissions_ranking = channel_details.get('submissions_ranking', None)
         self.submissions_limit = channel_details.get('submissions_limit', None)
@@ -97,3 +146,15 @@ def send_post(self, submission, r2t):
             gallery=self.content['gallery'],
             other=self.content['other']
         )
+
+
+def get_tags_for_submodule(submodule_name):
+    submodule_name = submodule_name.lower()
+    channel_doc = get_channel_doc(submodule_name)
+    if channel_doc and channel_doc.get('tags'):
+        return channel_doc.get('tags')
+    tags_filename = os.path.join('channels', submodule_name, 'tags.txt')
+    if os.path.exists(tags_filename):
+        with open(tags_filename, 'r') as tags_file:
+            return tags_file.read()
+    return None