From d8251ea16ea6a11fecb522c6f22ac76a13525a7f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 10:17:20 +0000 Subject: [PATCH 1/3] feat(db): store channel HTTP validators and poll live notify targets Add channels.http_etag and http_last_modified, create them on connect if the hand-applied migration has not run yet, and skip channels that have nobody to notify (notify=1 on a live user, or an active tg-channel). Co-authored-by: Kovalenko K. --- .../00012_channel_http_validators.py | 45 ++++++++ db/sqliteAdapter.py | 104 ++++++++++++++++++ 2 files changed, 149 insertions(+) create mode 100644 db/migrations/00012_channel_http_validators.py diff --git a/db/migrations/00012_channel_http_validators.py b/db/migrations/00012_channel_http_validators.py new file mode 100644 index 0000000..90b3aca --- /dev/null +++ b/db/migrations/00012_channel_http_validators.py @@ -0,0 +1,45 @@ +# -*- coding: utf-8 -*- +import os +import sys +bot_path = os.getcwd().split('/db/migrations')[0] +sys.path.insert(1, bot_path) + +from db.connection import connect_sqlite + +from db.sqliteAdapter import SQLighter +from config import db_path + +class SQLighterLocal: + + def __init__(self, database): + self.connection = connect_sqlite(database) + self.cursor = self.connection.cursor() + + def close(self): + self.connection.close() + + def add_channel_http_validators(self): + with self.connection: + columns = [ + row[1] for row in + self.cursor.execute("PRAGMA table_info(channels)").fetchall()] + if 'http_etag' not in columns: + self.cursor.execute( + "ALTER TABLE channels ADD COLUMN http_etag TEXT;") + else: + print("http_etag already exists") + if 'http_last_modified' not in columns: + self.cursor.execute( + "ALTER TABLE channels ADD COLUMN http_last_modified TEXT;") + else: + print("http_last_modified already exists") + + +db = SQLighterLocal(db_path) +db.add_channel_http_validators() +db.close() + +# db_users = SQLighter(db_path) +# db_users.close() + +print("created") diff --git a/db/sqliteAdapter.py b/db/sqliteAdapter.py index 53f0533..1704502 100644 --- a/db/sqliteAdapter.py +++ b/db/sqliteAdapter.py @@ -18,6 +18,47 @@ _users_deleted_at_lock = threading.Lock() _users_deleted_at_checked = False +_channel_http_validators_lock = threading.Lock() +_channel_http_validators_checked = False + + +def _ensure_channel_http_validators_columns(connection: sqlite3.Connection) -> None: + # db/migrations/00012_channel_http_validators.py adds ETag/Last-Modified + # columns, but migrations are applied by hand while deploy only git-pulls + # and restarts. The updater reads these on every circle, so create them + # here as well, once per process, to survive a deploy ahead of the migration. + global _channel_http_validators_checked + if _channel_http_validators_checked: + return + + with _channel_http_validators_lock: + if _channel_http_validators_checked: + return + try: + columns = [ + row[1] for row in + connection.execute("PRAGMA table_info(channels)").fetchall()] + if len(columns) > 0: + if 'http_etag' not in columns: + connection.execute( + "ALTER TABLE channels ADD COLUMN http_etag TEXT") + connection.commit() + logger.warn( + "channels.http_etag was missing and has been created") + if 'http_last_modified' not in columns: + connection.execute( + "ALTER TABLE channels ADD COLUMN http_last_modified TEXT") + connection.commit() + logger.warn( + "channels.http_last_modified was missing " + "and has been created") + except Exception as e: + logger.err( + "Could not ensure channels HTTP validator columns:", e) + finally: + _channel_http_validators_checked = True + + def _ensure_users_deleted_at_column(connection: sqlite3.Connection) -> None: # db/migrations/00011_users_deleted_at.py adds users.deleted_at, but migrations are applied @@ -63,6 +104,7 @@ def __init__(self, database): self.connection.row_factory = sqlite3.Row self.cursor = self.connection.cursor() _ensure_users_deleted_at_column(self.connection) + _ensure_channel_http_validators_columns(self.connection) def __enter__(self): return self @@ -287,6 +329,59 @@ def get_channel_or_next(self, channel_id, channel_set=None): LIMIT 1", (str(channel_id),)).fetchone() + def _live_notify_recipient_exists_sql(self): + # Same joins as get_uccs_by_channel / getTgChannelSubConnectionsByPodcast: + # notify=1 on a live user (paid or not), or an active tg-channel + # connection whose owner has channel_control. No CAST on telegram ids. + return """ + ( + EXISTS ( + SELECT 1 + FROM user_channel_cs uc + WHERE uc.channel_id = c.id + AND uc.notify = 1 + AND NOT EXISTS ( + SELECT 1 FROM users du + WHERE du.telegramId = uc.user_telegram_id + AND du.deleted_at IS NOT NULL + ) + ) + OR EXISTS ( + SELECT 1 + FROM user_channel_cs AS uc + INNER JOIN subscription_to_tg_channel_cs AS sttcc + ON (sttcc.user_channel_cs_id = uc.id) + INNER JOIN tg_channels AS tc + ON (tc.id = sttcc.tg_channel_id) + LEFT JOIN user_tariff_cs AS ut ON ut.uid = (SELECT id + FROM users AS u + WHERE u.telegramId = uc.user_telegram_id) + LEFT JOIN tariffs AS t ON t.id = ut.tariff_id + WHERE uc.channel_id = c.id + AND tc.active = 1 + AND ut.notify_count != 0 + AND ut.time_left > 0 + AND t.channel_control = 1 + ) + ) + """ + + def get_next_channel_to_poll(self, channel_id): + return self._get_channel_to_poll(channel_id, inclusive=False) + + def get_channel_or_next_to_poll(self, channel_id): + return self._get_channel_to_poll(channel_id, inclusive=True) + + def _get_channel_to_poll(self, channel_id, inclusive=True): + with self.connection: + op = ">=" if inclusive else ">" + sql = ( + "SELECT c.* FROM channels c " + "WHERE c.id %s ? AND %s " + "ORDER BY c.id ASC LIMIT 1" + ) % (op, self._live_notify_recipient_exists_sql()) + return self.cursor.execute(sql, (str(channel_id),)).fetchone() + def get_last_channel_id(self): with self.connection: return self.cursor.execute( @@ -302,6 +397,15 @@ def update_channel_last_guid_date( str(lastGuid), str(lastDate), str(podcastId),)) self.connection.commit() + def update_channel_http_validators( + self, podcastId, etag, last_modified): + with self.connection: + self.cursor.execute( + 'UPDATE channels SET http_etag = ?, http_last_modified = ? \ + WHERE id = ?', + (etag, last_modified, str(podcastId),)) + self.connection.commit() + def get_uccs_by_channel( self, channel_id, notifications_enabled=None, have_subscription=None, From 9e97b92c5edb26a4a642943275031cc9fb1167de Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 10:17:23 +0000 Subject: [PATCH 2/3] feat(updater): fetch rss_link with ETag and skip unchanged feeds The scheduled circle uses our stored rss_link plus If-None-Match / If-Modified-Since. HTTP 304 is not parsed, not a feed failure, and does not sleep 6s. iTunes lookup stays the fallback when rss_link is empty and for the manual update button / search UI. Co-authored-by: Kovalenko K. --- app/jobs/podcastsUpdater.py | 106 +++++++++++++++++++++++--------- app/service/podcast/podcast.py | 100 +++++++++++++++++++++++++++--- app/service/podcast/rss.py | 107 +++++++++++++++++++++++++++------ 3 files changed, 258 insertions(+), 55 deletions(-) diff --git a/app/jobs/podcastsUpdater.py b/app/jobs/podcastsUpdater.py index d6a7d18..6c00a11 100644 --- a/app/jobs/podcastsUpdater.py +++ b/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,14 +127,15 @@ 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: @@ -132,7 +143,10 @@ def main(interval=120): 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( diff --git a/app/service/podcast/podcast.py b/app/service/podcast/podcast.py index b35d8ec..c64ded3 100644 --- a/app/service/podcast/podcast.py +++ b/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 diff --git a/app/service/podcast/rss.py b/app/service/podcast/rss.py index 1ca8c77..e04ffbd 100644 --- a/app/service/podcast/rss.py +++ b/app/service/podcast/rss.py @@ -5,7 +5,7 @@ from lxml import etree from lib.requests import requesterModule -from lib.tools.logger import logger +from lib.requests.requesterModule import STD_REQUEST_HEADERS requester = requesterModule.Requester() @@ -21,13 +21,15 @@ FEED_RETRY_PAUSE_SECONDS = 3 # Статусы выборки фида. -FeedStatus = typing.Literal['ok', 'gone', 'unavailable'] +FeedStatus = typing.Literal['ok', 'gone', 'unavailable', 'not_modified'] # фид получен и разобран FEED_STATUS_OK: FeedStatus = 'ok' # фид точно больше не существует: сервер ответил 404/410 FEED_STATUS_GONE: FeedStatus = 'gone' # сеть, таймаут, 5xx, 403/429, битый xml — ошибка может быть временной FEED_STATUS_UNAVAILABLE: FeedStatus = 'unavailable' +# HTTP 304: тело не менялось, парсить нечего +FEED_STATUS_NOT_MODIFIED: FeedStatus = 'not_modified' # коды, по которым считаем, что фида больше нет FEED_GONE_STATUS_CODES = (404, 410) @@ -35,22 +37,45 @@ HEADERS_PARAMS = ['Usual', 'Empty'] -def get_rss_root(feed_url): - root, _status = get_rss_root_with_status(feed_url) +class FeedValidators(typing.TypedDict): + etag: str | None + last_modified: str | None + + +def empty_validators( + etag: str | None = None, + last_modified: str | None = None) -> FeedValidators: + return {'etag': etag, 'last_modified': last_modified} + + +def feed_status_counts_as_failure(status: FeedStatus) -> bool: + # 304 — фид доступен и не изменился; в счётчик FEED_FAILURES не идёт + return status in (FEED_STATUS_GONE, FEED_STATUS_UNAVAILABLE) + + +def get_rss_root(feed_url, etag=None, last_modified=None): + root, _status, _validators = get_rss_root_with_status( + feed_url, etag=etag, last_modified=last_modified) return root -def get_rss_root_with_status(feed_url) -> typing.Tuple[typing.Any, FeedStatus]: - """Возвращает (root, status). +def get_rss_root_with_status( + feed_url, etag=None, last_modified=None +) -> typing.Tuple[typing.Any, FeedStatus, FeedValidators]: + """Возвращает (root, status, validators). root — разобранный фид либо False. - status — FEED_STATUS_OK / FEED_STATUS_GONE / FEED_STATUS_UNAVAILABLE. + status — ok / gone / unavailable / not_modified. + validators — ETag и Last-Modified с финального ответа (после редиректов). + + 304 не ретраим и не считаем ошибкой: тела нет, парсить нечего. Вызывающий код должен различать 'gone' (фида больше нет) и 'unavailable' (могло просто моргнуть), чтобы не наказывать пользователей за разовый сбой сети. """ errors: list[str] = [] statuses: list[FeedStatus] = [] + last_validators = empty_validators(etag, last_modified) for retry_round in range(FEED_RETRY_ROUNDS): if retry_round > 0: @@ -59,19 +84,21 @@ def get_rss_root_with_status(feed_url) -> typing.Tuple[typing.Any, FeedStatus]: round_statuses: list[FeedStatus] = [] for headers_code in HEADERS_PARAMS: - headers: None | dict - if headers_code == 'Empty': - headers = {} - else: - headers = None + headers = __headers_for_attempt(headers_code, etag, last_modified) - content_result = __load_rss_root(feed_url, headers) + content_result = __load_rss_root(feed_url, headers, etag, last_modified) + last_validators = content_result.get('validators') or last_validators content = content_result['content'] + load_status = content_result['status'] + + if load_status == FEED_STATUS_NOT_MODIFIED: + return False, FEED_STATUS_NOT_MODIFIED, last_validators + if content is not False: root_result = __parse_rss_root(content) root = root_result['root'] if root is not False: - return root, FEED_STATUS_OK + return root, FEED_STATUS_OK, last_validators errors.append(root_result['error']) # ответ получили, но это не rss: может быть страница-заглушка, @@ -79,7 +106,7 @@ def get_rss_root_with_status(feed_url) -> typing.Tuple[typing.Any, FeedStatus]: round_statuses.append(FEED_STATUS_UNAVAILABLE) else: errors.append(content_result['error']) - round_statuses.append(content_result['status']) + round_statuses.append(load_status) statuses += round_statuses @@ -90,12 +117,37 @@ def get_rss_root_with_status(feed_url) -> typing.Tuple[typing.Any, FeedStatus]: print('; '.join(errors), flush=True) if statuses and all(status == FEED_STATUS_GONE for status in statuses): - return False, FEED_STATUS_GONE + return False, FEED_STATUS_GONE, last_validators - return False, FEED_STATUS_UNAVAILABLE + return False, FEED_STATUS_UNAVAILABLE, last_validators -def __load_rss_root(feed_url, headers=None): +def __headers_for_attempt(headers_code, etag, last_modified) -> dict: + if headers_code == 'Empty': + headers = {} + else: + headers = dict(STD_REQUEST_HEADERS) + + if etag: + headers['If-None-Match'] = etag + if last_modified: + headers['If-Modified-Since'] = last_modified + return headers + + +def __validators_from_response(request, fallback_etag=None, fallback_last_modified=None, + keep_fallback=False) -> FeedValidators: + etag = request.headers.get('ETag') + last_modified = request.headers.get('Last-Modified') + if keep_fallback: + if not etag: + etag = fallback_etag + if not last_modified: + last_modified = fallback_last_modified + return empty_validators(etag, last_modified) + + +def __load_rss_root(feed_url, headers=None, etag=None, last_modified=None): # запрос try: request = feed_requester.get(feed_url, headers=headers, timeout=FEED_REQUEST_TIMEOUT) @@ -103,9 +155,20 @@ def __load_rss_root(feed_url, headers=None): return { 'content': False, 'status': FEED_STATUS_UNAVAILABLE, + 'validators': empty_validators(etag, last_modified), 'error': "mainf/parsing_error2: " + str(e) + "; feed_url: " + str(feed_url) } + # 304: requests считает ok (status < 400), но тела нет — не парсим. + if request.status_code == 304: + return { + 'content': False, + 'status': FEED_STATUS_NOT_MODIFIED, + 'validators': __validators_from_response( + request, etag, last_modified, keep_fallback=True), + 'error': None + } + # запрос не удался if not request.ok: if request.status_code in FEED_GONE_STATUS_CODES: @@ -117,11 +180,17 @@ def __load_rss_root(feed_url, headers=None): return { 'content': False, 'status': status, + 'validators': empty_validators(etag, last_modified), 'error': "mainf/parsing_error3, result is not ok: " + str(request.status_code) + ", feed_url: " + feed_url } - return {'content': request.content, 'status': FEED_STATUS_OK} + return { + 'content': request.content, + 'status': FEED_STATUS_OK, + 'validators': __validators_from_response(request, keep_fallback=False), + 'error': None + } def __parse_rss_root(content): From e0f5b77c080b153904df1584185388766568e25b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 3 Sep 2026 10:17:28 +0000 Subject: [PATCH 3/3] test: cover feed ETag 304 and channel poll iterator Temp-file DBs and mocked HTTP only. 304 is not treated as ok-content, unchanged feeds are not parsed, and channels without live notify recipients are skipped. Co-authored-by: Kovalenko K. --- app/service/podcast/test_feed_etag.py | 244 ++++++++++++++++++++++++++ db/test_channel_poll.py | 235 +++++++++++++++++++++++++ 2 files changed, 479 insertions(+) create mode 100644 app/service/podcast/test_feed_etag.py create mode 100644 db/test_channel_poll.py diff --git a/app/service/podcast/test_feed_etag.py b/app/service/podcast/test_feed_etag.py new file mode 100644 index 0000000..f67f492 --- /dev/null +++ b/app/service/podcast/test_feed_etag.py @@ -0,0 +1,244 @@ +# -*- coding: utf-8 -*- +"""HTTP ETag / Last-Modified feed fetch. + +Uses mocked HTTP only — never opens production databases or the network. +Run from the repo root: + python app/service/podcast/test_feed_etag.py +If cchardet/lxml are only in the app venv: + venv/bin/python app/service/podcast/test_feed_etag.py +""" +import os +import sys +from unittest import mock + +_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.dirname(os.path.abspath(__file__))))) +if _ROOT not in sys.path: + sys.path.insert(0, _ROOT) + +from app.service.podcast import rss # noqa: E402 +from app.service.podcast import podcast as podcast_mod # noqa: E402 + + +MINIMAL_RSS = ( + b'' + b'' + b'Test Feed' + b'Ep1g1' + b'' +) + + +class FakeResponse: + def __init__(self, status_code, content=b'', headers=None): + self.status_code = status_code + self.content = content + self.headers = headers or {} + self.ok = status_code < 400 + + +def _assert(cond, label): + if not cond: + raise AssertionError(label) + print("ok %s" % label) + + +def test_304_is_not_modified_and_does_not_parse(): + calls = [] + parse_calls = [] + + def fake_get(url, headers=None, timeout=None, **kwargs): + calls.append({'url': url, 'headers': dict(headers or {})}) + return FakeResponse(304, content=b'', headers={'ETag': 'W/"abc"'}) + + real_parse = rss.__parse_rss_root + + def spy_parse(content): + parse_calls.append(content) + return real_parse(content) + + with mock.patch.object(rss.feed_requester, 'get', fake_get), \ + mock.patch.object(rss, '__parse_rss_root', spy_parse): + root, status, validators = rss.get_rss_root_with_status( + 'http://feed.example/rss', etag='W/"abc"', + last_modified='Wed, 01 Jan 2020 00:00:00 GMT') + + _assert(root is False, "304 root is False") + _assert(status == rss.FEED_STATUS_NOT_MODIFIED, "304 status is not_modified") + _assert(not rss.feed_status_counts_as_failure(status), + "304 does not count as feed failure") + _assert(parse_calls == [], "304 does not parse body") + _assert(len(calls) == 1, "304 is not retried with Empty headers") + _assert(calls[0]['headers'].get('If-None-Match') == 'W/"abc"', + "304 request sent If-None-Match") + _assert( + calls[0]['headers'].get('If-Modified-Since') + == 'Wed, 01 Jan 2020 00:00:00 GMT', + "304 request sent If-Modified-Since") + _assert(validators.get('etag') == 'W/"abc"', "304 keeps weak ETag") + + +def test_304_must_not_be_treated_as_ok_content(): + # requests.Response.ok is True for 304 (status < 400). If we treated that + # as a body, we would try to parse empty XML and call it unavailable. + def fake_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse(304, content=b'', headers={}) + + with mock.patch.object(rss.feed_requester, 'get', fake_get): + root, status, _validators = rss.get_rss_root_with_status( + 'http://feed.example/rss', etag='"x"') + + _assert(status != rss.FEED_STATUS_OK, "304 is not ok-content") + _assert(status != rss.FEED_STATUS_UNAVAILABLE, "304 is not unavailable") + _assert(status != rss.FEED_STATUS_GONE, "304 is not gone") + _assert(status == rss.FEED_STATUS_NOT_MODIFIED, "304 is not_modified") + _assert(root is False, "304 has no parsed root") + + +def test_200_parses_and_returns_new_etag(): + def fake_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse( + 200, content=MINIMAL_RSS, + headers={ + 'ETag': '"new-etag"', + 'Last-Modified': 'Thu, 02 Jan 2020 00:00:00 GMT', + }) + + with mock.patch.object(rss.feed_requester, 'get', fake_get): + root, status, validators = rss.get_rss_root_with_status( + 'http://feed.example/rss', etag='"old"') + + _assert(status == rss.FEED_STATUS_OK, "200 status is ok") + _assert(root is not False, "200 body is parsed") + _assert(root.tag == 'channel', "200 root is channel") + _assert(validators.get('etag') == '"new-etag"', "200 returns new ETag") + _assert( + validators.get('last_modified') == 'Thu, 02 Jan 2020 00:00:00 GMT', + "200 returns Last-Modified") + + +def test_podcast_info_query_threads_validators_and_304(): + def fake_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse(304, content=b'', headers={'ETag': '"keep"'}) + + with mock.patch.object(rss.feed_requester, 'get', fake_get): + root, pc_info = podcast_mod.podcast_info_query( + {'rss_link': 'http://feed.example/rss'}, 'rss', + etag='"keep"', last_modified=None) + + _assert(root is False, "query 304 root is False") + _assert(pc_info.get('notModified') is True, "query 304 sets notModified") + _assert('failureReason' not in pc_info, + "query 304 is not a failureReason") + _assert(pc_info.get('http_etag') == '"keep"', "query 304 returns etag") + + +def test_podcast_info_query_200_returns_etag(): + def fake_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse( + 200, content=MINIMAL_RSS, headers={'ETag': '"fresh"'}) + + with mock.patch.object(rss.feed_requester, 'get', fake_get): + root, pc_info = podcast_mod.podcast_info_query( + {'rss_link': 'http://feed.example/rss'}, 'rss') + + _assert(root is not False, "query 200 has root") + _assert(pc_info.get('http_etag') == '"fresh"', "query 200 http_etag") + _assert(not pc_info.get('notModified'), "query 200 not notModified") + + +def test_scheduled_path_skips_itunes_when_rss_link_present(): + calls = [] + + def fake_query(payload, service_name='itunes', direct_link=False, + etag=None, last_modified=None): + calls.append({ + 'service_name': service_name, + 'payload': payload, + 'etag': etag, + 'last_modified': last_modified, + }) + return False, { + 'notModified': True, + 'http_etag': etag, + 'http_last_modified': last_modified, + } + + channel = { + 'id': 7, + 'itunes_id': 999, + 'rss_link': 'http://feed.example/rss', + 'http_etag': '"abc"', + 'http_last_modified': 'Wed, 01 Jan 2020 00:00:00 GMT', + 'name': 'Show', + } + with mock.patch.object(podcast_mod, 'podcast_info_query', fake_query): + root, pc_info, service_name, service_id = podcast_mod.fetch_channel_feed( + channel, manual=False) + + _assert(len(calls) == 1, "scheduled rss_link: one fetch") + _assert(calls[0]['service_name'] == 'rss', "scheduled uses rss, not itunes") + _assert(calls[0]['payload'] == {'rss_link': 'http://feed.example/rss'}, + "scheduled payload is rss_link") + _assert(calls[0]['etag'] == '"abc"', "scheduled sends stored etag") + _assert(service_name == 'rss', "scheduled service_name is rss") + _assert(pc_info.get('notModified') is True, "scheduled 304 bubbles up") + _assert(root is False, "scheduled 304 root is False") + + +def test_manual_path_still_tries_itunes_first(): + calls = [] + + def fake_query(payload, service_name='itunes', direct_link=False, + etag=None, last_modified=None): + calls.append(service_name) + if service_name == 'itunes': + return False, {'failureReason': rss.FEED_STATUS_UNAVAILABLE} + return False, {'notModified': True, 'http_etag': etag} + + channel = { + 'id': 7, + 'itunes_id': 999, + 'rss_link': 'http://feed.example/rss', + 'http_etag': '"abc"', + 'name': 'Show', + } + with mock.patch.object(podcast_mod, 'podcast_info_query', fake_query): + podcast_mod.fetch_channel_feed(channel, manual=True) + + _assert(calls == ['itunes', 'rss'], "manual keeps itunes-then-rss") + + +def test_404_is_gone_503_is_unavailable(): + def gone_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse(404, content=b'nope') + + with mock.patch.object(rss.feed_requester, 'get', gone_get), \ + mock.patch.object(rss, 'FEED_RETRY_PAUSE_SECONDS', 0): + root, status, _v = rss.get_rss_root_with_status('http://feed.example/rss') + _assert(status == rss.FEED_STATUS_GONE, "404 is gone") + _assert(rss.feed_status_counts_as_failure(status), "404 counts as failure") + + def fail_get(url, headers=None, timeout=None, **kwargs): + return FakeResponse(503, content=b'') + + with mock.patch.object(rss.feed_requester, 'get', fail_get), \ + mock.patch.object(rss, 'FEED_RETRY_PAUSE_SECONDS', 0): + root, status, _v = rss.get_rss_root_with_status('http://feed.example/rss') + _assert(status == rss.FEED_STATUS_UNAVAILABLE, "503 is unavailable") + + +def main(): + test_304_is_not_modified_and_does_not_parse() + test_304_must_not_be_treated_as_ok_content() + test_200_parses_and_returns_new_etag() + test_podcast_info_query_threads_validators_and_304() + test_podcast_info_query_200_returns_etag() + test_scheduled_path_skips_itunes_when_rss_link_present() + test_manual_path_still_tries_itunes_first() + test_404_is_gone_503_is_unavailable() + print("all feed etag checks passed") + + +if __name__ == "__main__": + main() diff --git a/db/test_channel_poll.py b/db/test_channel_poll.py new file mode 100644 index 0000000..e3fbbd2 --- /dev/null +++ b/db/test_channel_poll.py @@ -0,0 +1,235 @@ +# -*- coding: utf-8 -*- +"""Channel poll iterator + HTTP validator columns. + +Uses a temporary file DB only — never opens production databases. +Run from the repo root: python db/test_channel_poll.py +""" +import os +import sys +import tempfile + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _ROOT not in sys.path: + sys.path.insert(0, _ROOT) + +from db import sqliteAdapter # noqa: E402 +from db.sqliteAdapter import SQLighter # noqa: E402 + + +def _assert_eq(got, expected, label): + if got != expected: + raise AssertionError("%s: expected %r, got %r" % (label, expected, got)) + print("ok %s = %r" % (label, got)) + + +def _schema(conn): + conn.executescript(""" + CREATE TABLE users ( + id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE, + telegramId INTEGER NOT NULL UNIQUE, + lang char(15), + deleted_at TEXT + ); + CREATE TABLE channels ( + id INTEGER PRIMARY KEY, + itunes_id INTEGER, + name TEXT, + rss_link TEXT, + last_guid TEXT, + last_date TEXT, + http_etag TEXT, + http_last_modified TEXT + ); + CREATE TABLE user_channel_cs ( + id INTEGER PRIMARY KEY, + user_telegram_id INTEGER NOT NULL, + channel_id INTEGER NOT NULL, + last_guid TEXT, + last_date TEXT, + notify INTEGER + ); + CREATE TABLE user_tariff_cs ( + id INTEGER PRIMARY KEY, + uid INTEGER NOT NULL, + tariff_id INTEGER NOT NULL, + balance INTEGER, + notify_count INTEGER, + time_left INTEGER + ); + CREATE TABLE tariffs ( + id INTEGER PRIMARY KEY, + channel_control INTEGER + ); + CREATE TABLE tg_channels ( + id INTEGER PRIMARY KEY, + user_id INTEGER, + tg_id INTEGER, + active INTEGER + ); + CREATE TABLE subscription_to_tg_channel_cs ( + id INTEGER PRIMARY KEY, + user_channel_cs_id INTEGER, + tg_channel_id INTEGER + ); + """) + conn.commit() + + +def _add_user(db, telegram_id, deleted_at=None): + db.cursor.execute( + "INSERT INTO users (telegramId, deleted_at) VALUES (?, ?)", + (telegram_id, deleted_at)) + db.connection.commit() + return db.cursor.execute( + "SELECT id FROM users WHERE telegramId = ?", + (telegram_id,)).fetchone()['id'] + + +def _add_channel(db, channel_id, name="p", rss_link=None, etag=None): + db.cursor.execute( + "INSERT INTO channels " + "(id, itunes_id, name, rss_link, http_etag) VALUES (?, ?, ?, ?, ?)", + (channel_id, channel_id, name, rss_link, etag)) + db.connection.commit() + + +def _add_sub(db, telegram_id, channel_id, notify=1): + db.cursor.execute( + "INSERT INTO user_channel_cs " + "(user_telegram_id, channel_id, notify) VALUES (?, ?, ?)", + (telegram_id, channel_id, notify)) + db.connection.commit() + + +def test_iterator_skips_empty_and_includes_live_notify(db_path): + db = SQLighter(db_path) + try: + _schema(db.connection) + _add_user(db, 1001) + _add_user(db, 1002) + _add_user(db, 2001, deleted_at="2026-01-01") + + # 1: nobody listens + _add_channel(db, 1, name="empty") + # 2: notify off + _add_channel(db, 2, name="muted") + _add_sub(db, 1002, 2, notify=0) + # 3: only a blocked user + _add_channel(db, 3, name="blocked") + _add_sub(db, 2001, 3, notify=1) + # 4: live user with notify=1 (no tariff — nosubs still get episodes) + _add_channel(db, 4, name="live") + _add_sub(db, 1001, 4, notify=1) + # 5: another empty after the live one + _add_channel(db, 5, name="empty-after") + # 6: live notify again + _add_channel(db, 6, name="live-2") + _add_sub(db, 1001, 6, notify=1) + + first = db.get_channel_or_next_to_poll(1) + _assert_eq(int(first['id']), 4, "first pollable from id>=1") + + nxt = db.get_next_channel_to_poll(int(first['id'])) + _assert_eq(int(nxt['id']), 6, "skips empty channel 5") + + after = db.get_next_channel_to_poll(int(nxt['id'])) + _assert_eq(after, None, "no more pollable channels") + + none_before_live = db.get_next_channel_to_poll(4) + _assert_eq(int(none_before_live['id']), 6, "id>4 skips 5, hits 6") + finally: + db.close() + + +def test_iterator_includes_tg_channel_recipient(db_path): + db = SQLighter(db_path) + try: + _schema(db.connection) + owner_id = _add_user(db, 3001) + _add_channel(db, 10, name="tg-only") + db.cursor.execute( + "INSERT INTO user_channel_cs " + "(user_telegram_id, channel_id, notify) VALUES (?, ?, ?)", + (3001, 10, 0)) + db.connection.commit() + uc_id = db.cursor.execute( + "SELECT id FROM user_channel_cs WHERE channel_id = 10" + ).fetchone()['id'] + db.cursor.execute( + "INSERT INTO tariffs (id, channel_control) VALUES (2, 1)") + db.cursor.execute( + "INSERT INTO user_tariff_cs " + "(uid, tariff_id, balance, notify_count, time_left) " + "VALUES (?, 2, 0, -1, 100)", + (owner_id,)) + db.cursor.execute( + "INSERT INTO tg_channels (id, user_id, tg_id, active) " + "VALUES (1, 3001, -100, 1)") + db.cursor.execute( + "INSERT INTO subscription_to_tg_channel_cs " + "(user_channel_cs_id, tg_channel_id) VALUES (?, 1)", + (uc_id,)) + db.connection.commit() + + row = db.get_channel_or_next_to_poll(1) + _assert_eq(int(row['id']), 10, "tg-channel notify recipient is pollable") + finally: + db.close() + + +def test_update_and_read_http_validators(db_path): + db = SQLighter(db_path) + try: + _schema(db.connection) + _add_channel(db, 8, name="etag", rss_link="http://feed.example/rss") + db.update_channel_http_validators(8, 'W/"x"', 'Wed, 01 Jan 2020 00:00:00 GMT') + row = db.get_channel(8) + _assert_eq(row['http_etag'], 'W/"x"', "stored weak etag") + _assert_eq( + row['http_last_modified'], 'Wed, 01 Jan 2020 00:00:00 GMT', + "stored last-modified") + db.update_channel_http_validators(8, '"fresh"', None) + row = db.get_channel(8) + _assert_eq(row['http_etag'], '"fresh"', "etag updated on 200") + _assert_eq(row['http_last_modified'], None, "last-modified cleared") + finally: + db.close() + + +def test_ensure_columns_on_connect(db_path): + import sqlite3 + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE channels (id INTEGER PRIMARY KEY, name TEXT)") + conn.commit() + conn.close() + + sqliteAdapter._channel_http_validators_checked = False + db = SQLighter(db_path) + try: + columns = [ + row[1] for row in + db.connection.execute("PRAGMA table_info(channels)").fetchall()] + _assert_eq('http_etag' in columns, True, "ensure created http_etag") + _assert_eq( + 'http_last_modified' in columns, True, + "ensure created http_last_modified") + finally: + db.close() + sqliteAdapter._channel_http_validators_checked = False + + +def main(): + tmpdir = tempfile.mkdtemp(prefix="yourcast_channel_poll_") + test_iterator_skips_empty_and_includes_live_notify( + os.path.join(tmpdir, "iter.db")) + test_iterator_includes_tg_channel_recipient( + os.path.join(tmpdir, "tg.db")) + test_update_and_read_http_validators( + os.path.join(tmpdir, "etag.db")) + test_ensure_columns_on_connect( + os.path.join(tmpdir, "ensure.db")) + print("all channel poll checks passed") + + +if __name__ == "__main__": + main()