From 080dae3e742c3d98d6d53df0bc40fe700ed41220 Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Fri, 5 Jun 2026 20:45:08 +0300 Subject: [PATCH] feat(cache): add channel info cache with TTL and throttling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce `cached_get_chat` in `tg_cache` to store channel metadata on disk with a configurable TTL (default 12 hours). Use the `tg_rpc` throttling context for RPC calls. Update `rss_generator` to use the cached version and document related environment variables in `dockercompose.yml`. --- dockercompose.yml | 3 ++ rss_generator.py | 6 ++-- tg_cache.py | 91 +++++++++++++++++++++++++++++++++++++++++++++-- tg_throttle.py | 69 +++++++++++++++++++++++++++++++++++ 4 files changed, 165 insertions(+), 4 deletions(-) create mode 100644 tg_throttle.py diff --git a/dockercompose.yml b/dockercompose.yml index 5ae229a..a9156fa 100644 --- a/dockercompose.yml +++ b/dockercompose.yml @@ -21,6 +21,9 @@ services: # TG_WATCHDOG_HEARTBEAT_EVERY: 30 # Emit an INFO watchdog heartbeat every N successful probes (default: 30) # TG_DISCONNECT_FLAP_LIMIT: 3 # Disconnect events within the flap window before an in-process restart (default: 3) # TG_DISCONNECT_FLAP_WINDOW: 120 # Flap detection window in seconds (default: 120) + # TG_CHAT_CACHE_TTL_HOURS: 12 # TTL for cached channel info (title/username/id); removes GetFullChannel from the poll hot path (default: 12) + # TG_RPC_CONCURRENCY: 1 # Max concurrent live Telegram RPC calls — global throttle (default: 1) + # TG_RPC_MIN_INTERVAL_MS: 500 # Minimum gap between live Telegram RPC starts, ms (default: 500) PYROGRAM_BRIDGE_URL: https://pgbridge.example.com API_PORT: 80 TOKEN: ХХХ diff --git a/rss_generator.py b/rss_generator.py index 6fe5777..0545b0b 100644 --- a/rss_generator.py +++ b/rss_generator.py @@ -344,7 +344,8 @@ async def generate_channel_rss(channel: str | int, channel_info_start_time = time.time() try: channel = post_parser.channel_name_prepare(channel) - channel_info = await post_parser.client.get_chat(channel) + from tg_cache import cached_get_chat + channel_info = await cached_get_chat(post_parser.client, channel) channel_title = channel_info.title or f"Telegram: {channel}" channel_username = channel_info.username or (str(channel_info.id) if channel_info.id and str(channel_info.id).startswith('-100') else None) if not channel_username: @@ -556,7 +557,8 @@ async def generate_channel_html(channel: str | int, try: channel = post_parser.channel_name_prepare(channel) logger.debug(f"Prepared channel identifier for HTML: {channel} (type: {type(channel)})") # Log prepared channel - channel_info = await post_parser.client.get_chat(channel) + from tg_cache import cached_get_chat + channel_info = await cached_get_chat(post_parser.client, channel) channel_username = channel_info.username or (str(channel_info.id) if channel_info.id and str(channel_info.id).startswith('-100') else None) if not channel_username: logger.warning(f"Could not get username for channel {channel} in HTML generation, returning error feed structure (as string). NOTE: This should ideally return HTML error page.") diff --git a/tg_cache.py b/tg_cache.py index 435f7cf..af83019 100644 --- a/tg_cache.py +++ b/tg_cache.py @@ -17,15 +17,26 @@ import random import asyncio import time from datetime import datetime, timedelta +from types import SimpleNamespace from typing import Any, Optional, Union, List from pyrogram import Client from pyrogram.types import Chat, Message +from tg_throttle import tg_rpc logger = logging.getLogger(__name__) # Path to cache directory CACHE_DIR = os.path.join('data', 'tgcache') +# TTL (hours) for the channel-info cache. Channel title/username/id change rarely, +# so a long TTL removes the rate-limited GetFullChannel from the feed-poll hot path. +try: + CHAT_CACHE_TTL_HOURS = int(os.getenv("TG_CHAT_CACHE_TTL_HOURS", "12")) + if CHAT_CACHE_TTL_HOURS < 1: + CHAT_CACHE_TTL_HOURS = 12 +except ValueError: + CHAT_CACHE_TTL_HOURS = 12 + def _get_history_cache_file_path(channel_id: Union[str, int]) -> str: """Returns path to the message history cache file for the channel""" if not os.path.exists(CACHE_DIR): @@ -128,11 +139,87 @@ async def cached_get_chat_history(client: Client, channel_id: Union[str, int], l try: logger.info(f"history_cache_request: fetching fresh history for channel {channel_id}, limit {limit}") messages = [] - async for message in client.get_chat_history(channel_id, limit=limit): - messages.append(message) + async with tg_rpc(): + async for message in client.get_chat_history(channel_id, limit=limit): + messages.append(message) await asyncio.to_thread(_save_history_to_cache, channel_id, messages, limit) return messages except Exception as e: logger.error(f"history_cache_request_error: channel {channel_id}, limit {limit}, error {str(e)}") raise + + +def _get_chat_cache_file_path(channel_id: Union[str, int]) -> str: + """Returns path to the channel-info cache file for the channel.""" + if not os.path.exists(CACHE_DIR): + os.makedirs(CACHE_DIR, exist_ok=True) + logger.info(f"cache_dir_created: path {CACHE_DIR}") + channel_id_str = str(channel_id) + safe_filename = channel_id_str.replace('/', '_').replace('\\', '_') + # Distinct suffix so channel-info files never collide with history (.cache) files. + return os.path.join(CACHE_DIR, f"{safe_filename}.chatinfo") + + +def _save_chat_to_cache(channel_id: Union[str, int], data: dict) -> None: + """Saves channel-info (id/title/username) to cache.""" + try: + cache_file = _get_chat_cache_file_path(channel_id) + cache_data = {'timestamp': time.time(), 'data': data} + with open(cache_file, 'wb') as f: + pickle.dump(cache_data, f) + logger.info(f"chatinfo_cache_saved: channel {channel_id}, file {cache_file}") + except Exception as e: + logger.error(f"chatinfo_cache_save_error: channel {channel_id}, error {str(e)}") + + +def _get_chat_from_cache(channel_id: Union[str, int], max_age_hours: int = CHAT_CACHE_TTL_HOURS) -> Optional[dict]: + """Retrieves channel-info from cache if not older than max_age_hours (with up-to-20% randomization).""" + try: + cache_file = _get_chat_cache_file_path(channel_id) + if not os.path.exists(cache_file): + logger.info(f"chatinfo_cache_miss: channel {channel_id}, cache file not found") + return None + with open(cache_file, 'rb') as f: + cache_data = pickle.load(f) + cache_age = time.time() - cache_data['timestamp'] + # Add randomness up to 20% of max age, same approach as the history cache. + random_factor = 1 - random.uniform(0, 0.2) + adjusted_max_age = max_age_hours * 3600 * random_factor + if cache_age > adjusted_max_age: + logger.info(f"chatinfo_cache_expired: channel {channel_id}, age {cache_age:.1f}s > adjusted max {adjusted_max_age:.1f}s (random factor: {random_factor:.2f})") + return None + data = cache_data.get('data') + if not isinstance(data, dict): + logger.warning(f"chatinfo_cache_invalid: channel {channel_id}, unexpected payload type {type(data).__name__}") + return None + logger.info(f"chatinfo_cache_hit: channel {channel_id}, age {cache_age:.1f}s") + return data + except Exception as e: + logger.error(f"chatinfo_cache_read_error: channel {channel_id}, error {str(e)}") + return None + + +async def cached_get_chat(client: Client, channel_id: Union[str, int]) -> SimpleNamespace: + """Gets channel info (id/title/username) with disk TTL caching. + + Avoids the rate-limited channels.GetFullChannel on every feed poll. Mirrors + cached_get_chat_history. On a cache miss the live get_chat is throttled and its + exceptions (FloodWait, UsernameInvalid, ...) propagate to the caller unchanged; + only successful lookups are cached. The returned object exposes .id/.title/.username. + """ + cached = await asyncio.to_thread(_get_chat_from_cache, channel_id) + if cached is not None: + return SimpleNamespace(**cached) + + logger.info(f"chatinfo_cache_request: fetching fresh chat info for channel {channel_id}") + async with tg_rpc(): + chat = await client.get_chat(channel_id) + + data = { + 'id': getattr(chat, 'id', None), + 'title': getattr(chat, 'title', None), + 'username': getattr(chat, 'username', None), + } + await asyncio.to_thread(_save_chat_to_cache, channel_id, data) + return SimpleNamespace(**data) diff --git a/tg_throttle.py b/tg_throttle.py new file mode 100644 index 0000000..b96075c --- /dev/null +++ b/tg_throttle.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +# flake8: noqa +# pylint: disable=broad-exception-caught, global-statement, missing-function-docstring, missing-class-docstring +# pylint: disable=logging-fstring-interpolation, line-too-long + +import os +import time +import asyncio +import logging + +logger = logging.getLogger(__name__) + + +def _parse_int_env(name: str, default: int, minimum: int) -> int: + """Parse an int env var, falling back to default on missing/invalid/out-of-range value.""" + raw = os.getenv(name) + if raw is None or raw.strip() == "": + return default + try: + value = int(raw) + except ValueError: + logger.warning(f"tg_throttle: {name} is not a valid integer ({raw!r}); using default {default}") + return default + if value < minimum: + logger.warning(f"tg_throttle: {name}={value} below minimum {minimum}; using default {default}") + return default + return value + + +# Global throttle for live Telegram MTProto RPC calls. Serializes bursts (e.g. miniflux +# batching ~47 feeds at once) so they do not trip Telegram's FLOOD_WAIT (420). +_CONCURRENCY = _parse_int_env("TG_RPC_CONCURRENCY", 1, 1) # max concurrent Telegram RPCs +_MIN_INTERVAL = _parse_int_env("TG_RPC_MIN_INTERVAL_MS", 500, 0) / 1000.0 # min gap between RPC starts (seconds) + +_sem = asyncio.Semaphore(_CONCURRENCY) +_lock = asyncio.Lock() +_last_start = 0.0 + +logger.info(f"tg_throttle: initialized (concurrency={_CONCURRENCY}, min_interval={_MIN_INTERVAL*1000:.0f}ms)") + + +class _TgRpcGate: + """Async context manager that caps concurrency and enforces a minimum spacing between RPC starts.""" + + async def __aenter__(self): + await _sem.acquire() + global _last_start + try: + async with _lock: + wait = _last_start + _MIN_INTERVAL - time.monotonic() + if wait > 0: + await asyncio.sleep(wait) + _last_start = time.monotonic() + except BaseException: + # Do not leak a semaphore permit if cancelled while waiting for the spacing. + _sem.release() + raise + return self + + async def __aexit__(self, exc_type, exc, tb): + _sem.release() + return False + + +def tg_rpc(): + """Return an async context manager that throttles a single live Telegram RPC call.""" + return _TgRpcGate()