feat: updater circle uses rss_link + HTTP ETag, skips empty channels

app/jobs/podcastsUpdater.py

@@ -47,6 +47,15 @@
 # Фид ответил 404/410 — его точно больше нет, порог ниже.
 FEED_GONE_FAILURES_BEFORE_NOTIFY_OFF = 3
 
+# skipped — канал без получателей, not_modified — HTTP 304, fetched — качали/парсили
+class ChannelUpdateResult(typing.NamedTuple):
+    new_recs: bool = False
+    outcome: str = 'fetched'
+
+    def __bool__(self):
+        return self.new_recs
+
+
 logger = Logger(file="updater")
 
 
@@ -91,8 +100,9 @@ def main(interval=120):
         # channel = db_users.get_channel_or_next(
         # 	last_updated_channel_id, channels_to_check_ids)
         # ---------
-        # получение всех, так как шлём пользователям без подписки уведомления
-        channel = db_users.get_channel_or_next(last_updated_channel_id)
+        # только каналы, по которым есть кому слать (notify=1, живой юзер
+        # или активный tg-канал). Пустые больше не крутим и не sleep(6).
+        channel = db_users.get_channel_or_next_to_poll(last_updated_channel_id)
         db_users.close()
 
         while channel is not None:
@@ -117,22 +127,26 @@ def main(interval=120):
 
             storage.set_last_channel_id(channel['id'])
 
+            update_result = ChannelUpdateResult(False, 'skipped')
             if connections is not None:
-                send_new_records_by_channel(
+                update_result = send_new_records_by_channel(
                     channel, connections, thonbot=thonbot,
                     nosubs_connections=nosubs_connections,
                     tg_channel_connections=tg_channel_connections)
 
             db_users = SQLighter(db_path)
-            channel = db_users.get_next_channel(channel['id'])
+            channel = db_users.get_next_channel_to_poll(channel['id'])
             db_users.close()
 
             if channel is None:
                 logger.log("Next channel is None!")
             else:
                 logger.log("Next channel is:", channel['id'])
 
-            time.sleep(6)
+            # 6s только после реальной загрузки фида, чтобы не молотить хосты.
+            # 304 и каналы без получателей — сразу к следующему.
+            if update_result.outcome == 'fetched':
+                time.sleep(6)
             # time.sleep(60 * 60)
 
         storage.set_last_channel_id(1)
@@ -166,22 +180,30 @@ def send_new_records_by_channel(
             and (tg_channel_connections is None or len(tg_channel_connections) == 0):
         # # обновление инф. о последнем обновлении канала
         # updatePodcastLastGuidDate(channel)
-        return new_recs_flag
-
-    root: app.service.podcast.podcast.RootAdapter | typing.Literal[False] = False
-    pc_info: app.service.podcast.podcast.PodcastInfoType = {}
-    service_name: str | None = None
-
-    if channel["itunes_id"] is not None and channel["itunes_id"]:
-        payload = {'entity': 'podcast', 'id': channel['itunes_id']}
-        root, pc_info = app.service.podcast.podcast.podcast_info_query(payload)
-        service_name = 'itunes'
-        service_id = channel["itunes_id"]
-    if root is False and channel["rss_link"] is not None and channel["rss_link"]:
-        payload = {'rss_link': channel["rss_link"]}
-        root, pc_info = app.service.podcast.podcast.podcast_info_query(payload, 'rss')
-        service_name = 'rss'
-        service_id = channel["rss_link"]
+        return ChannelUpdateResult(new_recs_flag, 'skipped')
+
+    root, pc_info, service_name, service_id = \
+        app.service.podcast.podcast.fetch_channel_feed(channel, manual=manual)
+
+    if pc_info.get('notModified'):
+        # 304: фид живой и не менялся. Не парсим, не шлём, счётчик сбоев не трогаем
+        # (сбрасываем — это успех «фид доступен»).
+        storage.reset_channel_feed_failures(channel['id'])
+        _persist_channel_http_validators(channel, pc_info)
+        if nosubs_connections:
+            nosub_connections_to_pgd = {}
+            for connection in nosubs_connections:
+                nosub_connections_to_pgd[connection['user_telegram_id']] = \
+                    connection['last_guid']
+            flag_nosubs_for_digest(
+                nosub_connections_to_pgd,
+                latest_episode_id(channel, connections),
+                channel['last_date'] if 'last_date' in channel.keys() else None,
+                channel['id'])
+        logger.log(
+            "Feed not modified for channel", channel['id'],
+            "; etag:", pc_info.get('http_etag'))
+        return ChannelUpdateResult(False, 'not_modified')
 
     if root is False:
         failure_reason = pc_info.get(
@@ -222,7 +244,7 @@ def send_new_records_by_channel(
                 except Exception as e:
                     logger.err("podcastsUpdater/manualFeedUnavailableNotice: ", e)
 
-            return new_recs_flag
+            return ChannelUpdateResult(new_recs_flag, 'fetched')
 
         failures = storage.increase_channel_feed_failures(channel['id'])
         if failure_reason == app.service.podcast.rss.FEED_STATUS_GONE:
@@ -238,7 +260,7 @@ def send_new_records_by_channel(
                 "; reason:", failure_reason,
                 "; consecutive failures:", failures, "of", failures_threshold,
                 "; notifications are left enabled")
-            return new_recs_flag
+            return ChannelUpdateResult(new_recs_flag, 'fetched')
 
         logger.warn(
             "Feed is stably unavailable for channel", channel['id'],
@@ -290,10 +312,11 @@ def send_new_records_by_channel(
                 logger.err(
                     "PARSING ERROR! In podcastUpdater2. ",
                     "Notifications disabled for podcast ", e)
-        return new_recs_flag
+        return ChannelUpdateResult(new_recs_flag, 'fetched')
 
     # фид получен — серия неудач прервана
     storage.reset_channel_feed_failures(channel['id'])
+    _persist_channel_http_validators(channel, pc_info)
 
     if service_name == 'itunes':
         # # flag = True
@@ -308,7 +331,7 @@ def send_new_records_by_channel(
         itunes_link = pc_info["itunesLink"]  # None
         feed_url = pc_info["feedUrl"]
     else:
-        return
+        return ChannelUpdateResult(False, 'fetched')
 
     # объединение связей с пользователями и каналами
     all_target_connections = connections
@@ -425,7 +448,7 @@ def send_new_records_by_channel(
                 flag_nosubs_for_digest(
                     nosub_connections_to_pgd, latest_pgd, last_date,
                     channel['id'])
-                return new_recs_flag
+                return ChannelUpdateResult(new_recs_flag, 'fetched')
 
         elif channelDescr.tag == "item":
             if flag_have_users == 1 or i > MAX_EPISODES_PER_PODCAST:  # ограничение на кол-во подкастов
@@ -507,7 +530,7 @@ def send_new_records_by_channel(
             i += 1
 
     if len(links) < 1:
-        return new_recs_flag
+        return ChannelUpdateResult(new_recs_flag, 'fetched')
 
     last_date = app.service.podcast.podcast.set_last_date(last_date, pub_dates_strped[0])
 
@@ -676,11 +699,36 @@ def send_new_records_by_channel(
                         user_tg_id, channel['id'], pgd, last_date)
                 except Exception as e:
                     logger.err("podacstUpdater/db_after_ops2: ", e)
-                    return new_recs_flag
+                    return ChannelUpdateResult(new_recs_flag, 'fetched')
 
     db_users.close()
 
-    return new_recs_flag
+    return ChannelUpdateResult(new_recs_flag, 'fetched')
+
+
+def _persist_channel_http_validators(channel, pc_info):
+    if not pc_info:
+        return
+    new_etag = pc_info.get('http_etag')
+    new_lm = pc_info.get('http_last_modified')
+    try:
+        old_etag = channel['http_etag']
+    except (KeyError, IndexError, TypeError):
+        old_etag = None
+    try:
+        old_lm = channel['http_last_modified']
+    except (KeyError, IndexError, TypeError):
+        old_lm = None
+    if new_etag == old_etag and new_lm == old_lm:
+        return
+    db_users = SQLighter(db_path)
+    try:
+        db_users.update_channel_http_validators(
+            channel['id'], new_etag, new_lm)
+    except Exception as e:
+        logger.err("podcastsUpdater/persistHttpValidators: ", e)
+    finally:
+        db_users.close()
 
 
 def flag_nosubs_for_digest(

app/service/podcast/podcast.py

@@ -4,7 +4,8 @@
 
 from app.service.podcast.rss import (
     get_rss_root_with_status,
-    FeedStatus, FEED_STATUS_GONE, FEED_STATUS_UNAVAILABLE)
+    FeedStatus, FEED_STATUS_GONE, FEED_STATUS_UNAVAILABLE,
+    FEED_STATUS_NOT_MODIFIED)
 from lib.requests import requesterModule
 from lib.tools.logger import logger
 from lib.tools.time_tools.general import format_rss_last_date, prepare_date_time_from_formatted
@@ -30,9 +31,68 @@ class PodcastInfoType(TypedDict, total=False):
     itunesData: Dict | None
     # почему выборка не удалась: FEED_STATUS_GONE | FEED_STATUS_UNAVAILABLE
     failureReason: FeedStatus
+    http_etag: str | None
+    http_last_modified: str | None
+    notModified: bool
 
 
-def podcast_info_query(payload, service_name='itunes', direct_link=False) \
+def _channel_field(channel, key, default=None):
+    try:
+        value = channel[key]
+    except (KeyError, IndexError, TypeError):
+        return default
+    if value is None or value == '':
+        return default
+    return value
+
+
+def fetch_channel_feed(channel, manual=False):
+    """Выборка фида канала для апдейтера.
+
+    Scheduled (manual=False): наш rss_link + сохранённые HTTP-валидаторы.
+    iTunes — только если rss_link пустой.
+    Ручная кнопка «обновить» оставляет itunes-then-rss, но валидаторы
+    всё равно уходят на GET RSS.
+    """
+    etag = _channel_field(channel, 'http_etag')
+    last_modified = _channel_field(channel, 'http_last_modified')
+    rss_link = _channel_field(channel, 'rss_link')
+    itunes_id = _channel_field(channel, 'itunes_id')
+
+    if not manual and rss_link:
+        root, pc_info = podcast_info_query(
+            {'rss_link': rss_link}, 'rss',
+            etag=etag, last_modified=last_modified)
+        return root, pc_info, 'rss', rss_link
+
+    root: RootAdapter | Literal[False] = False
+    pc_info: PodcastInfoType = {}
+    service_name: str | None = None
+    service_id = None
+
+    if itunes_id:
+        payload = {'entity': 'podcast', 'id': itunes_id}
+        root, pc_info = podcast_info_query(
+            payload, etag=etag, last_modified=last_modified)
+        service_name = 'itunes'
+        service_id = itunes_id
+    if (
+            root is False
+            and not pc_info.get('notModified')
+            and rss_link
+    ):
+        root, pc_info = podcast_info_query(
+            {'rss_link': rss_link}, 'rss',
+            etag=etag, last_modified=last_modified)
+        service_name = 'rss'
+        service_id = rss_link
+
+    return root, pc_info, service_name, service_id
+
+
+def podcast_info_query(
+        payload, service_name='itunes', direct_link=False,
+        etag=None, last_modified=None) \
         -> Tuple[RootAdapter | Literal[False], PodcastInfoType]:
     if service_name == 'itunes':
         api_url_base = 'https://itunes.apple.com/lookup'
@@ -100,22 +160,48 @@ def podcast_info_query(payload, service_name='itunes', direct_link=False) \
 
     feed_url = str(feed_url)
 
-    root, feed_status = get_rss_root_with_status(feed_url)
-    if root is False and "www." in feed_url and not direct_link:
+    root, feed_status, validators = get_rss_root_with_status(
+        feed_url, etag=etag, last_modified=last_modified)
+    if (
+            root is False
+            and feed_status != FEED_STATUS_NOT_MODIFIED
+            and "www." in feed_url
+            and not direct_link
+    ):
         feed_url = feed_url.replace("www.", "")
-        root, feed_status = get_rss_root_with_status(feed_url)
+        root, feed_status, validators = get_rss_root_with_status(
+            feed_url, etag=etag, last_modified=last_modified)
+
+    http_etag = validators.get('etag') if validators else None
+    http_last_modified = validators.get('last_modified') if validators else None
+
+    if feed_status == FEED_STATUS_NOT_MODIFIED:
+        return False, {
+            'lastDate': last_date,
+            'itunesLink': itunes_link,
+            'feedUrl': feed_url,
+            'collectionName': collection_name,
+            'itunesData': itunes_podcast_data,
+            'http_etag': http_etag,
+            'http_last_modified': http_last_modified,
+            'notModified': True}
+
     if root is False:
         logger.warn("Error payload info: ", payload, "; reason: ", feed_status)
         return False, {
             'collectionName': collection_name,
             'failureReason': (
                 FEED_STATUS_GONE if feed_status == FEED_STATUS_GONE
-                else FEED_STATUS_UNAVAILABLE)}
+                else FEED_STATUS_UNAVAILABLE),
+            'http_etag': http_etag,
+            'http_last_modified': http_last_modified}
 
     return root, {
         'lastDate': last_date, 'itunesLink': itunes_link,
         'feedUrl': feed_url, 'collectionName': collection_name,
-        'itunesData': itunes_podcast_data}
+        'itunesData': itunes_podcast_data,
+        'http_etag': http_etag,
+        'http_last_modified': http_last_modified}
 
 
 def set_last_date(last_date, last_pub_date) -> str:  # strings: itunes, rss; not formatted