implement RSS cache update mechanism

api_server.py

@@ -36,6 +36,7 @@ from config import get_settings, setup_logging
 from rss_generator import generate_channel_rss, generate_channel_html
 from post_parser import PostParser
 from url_signer import verify_media_digest, generate_media_digest
+from rss_cache import update_channel_access_time, start_cache_updater
 
 # Define custom exception for zero-size files
 class ZeroSizeFileError(Exception):
@@ -72,10 +73,13 @@ async def lifespan(_: FastAPI):
     
     await client.start()
     background_task = asyncio.create_task(cache_media_files()) # Start background task
+    cache_updater_task = asyncio.create_task(start_cache_updater(client.client)) # Start cache updater task
     yield
     background_task.cancel() # Cleanup
+    cache_updater_task.cancel()
     try:
         await background_task
+        await cache_updater_task
     except asyncio.CancelledError:
         pass
     await client.stop()
@@ -785,7 +789,10 @@ async def get_rss_feed(channel: str,
             logger.info(f"Valid token for RSS endpoint: {token}")
     elif Config["token"] and is_local_request(request):
         logger.info(f"Local request, skipping token check for RSS endpoint.")
-        
+    
+    # Обновляем время доступа к каналу в cache_access.json
+    update_channel_access_time(channel)
+    
     while True:
         try:
             

config.py

@@ -60,4 +60,7 @@ def get_settings() -> dict[str, Any]:
         "time_based_merge": os.getenv("TIME_BASED_MERGE", "False").strip() in ["True", "true"],
         "show_bridge_link": os.getenv("SHOW_BRIDGE_LINK", "False").strip() in ["True", "true"],
         "show_post_flags": os.getenv("SHOW_POST_FLAGS", "False").strip() in ["True", "true"],
+        "rss_cache_update_interval": int(os.getenv("RSS_CACHE_UPDATE_INTERVAL", "3600")),
+        "rss_cache_update_delay": int(os.getenv("RSS_CACHE_UPDATE_DELAY", "60")),
+        "rss_cache_max_age_days": int(os.getenv("RSS_CACHE_MAX_AGE_DAYS", "30")),
     } 

rss_cache.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# flake8: noqa
+# pylint: disable=broad-exception-raised, raise-missing-from, too-many-arguments, redefined-outer-name
+# pylint: disable=multiple-statements, logging-fstring-interpolation, trailing-whitespace, line-too-long
+# pylint: disable=broad-exception-caught, missing-function-docstring, missing-class-docstring
+# pylint: disable=f-string-without-interpolation
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+
+import logging
+import os
+import json
+import time
+import asyncio
+import hashlib
+from pathlib import Path
+from datetime import datetime, timedelta
+from typing import Union, Optional
+from config import get_settings
+
+logger = logging.getLogger(__name__)
+Config = get_settings()
+
+def get_cache_path(channel: Union[str, int], content_type: str = 'rss') -> Path:
+    """
+    Get the path to the cache file for a given channel
+    Args:
+        channel: Telegram channel name or id
+        content_type: Type of content (rss or html)
+    Returns:
+        Path to the cache file
+    """
+    if content_type == 'html':
+        base_dir = './data/html-cache'
+        file_ext = 'html'
+    else:
+        base_dir = './data/rss-cache'
+        file_ext = 'xml'
+        
+    cache_dir = Path(Config.get('cache_dir', base_dir))
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    
+    # Create a unique filename based on the channel identifier
+    if isinstance(channel, int):
+        filename = f"channel_{channel}.{file_ext}"
+    else:
+        # Use hash for channel names with special characters
+        if any(c in channel for c in '/<>:"|?*\\'):
+            channel_hash = hashlib.md5(channel.encode()).hexdigest()
+            filename = f"channel_{channel_hash}.{file_ext}"
+        else:
+            filename = f"channel_{channel}.{file_ext}"
+    
+    return cache_dir / filename
+
+def is_cache_valid(cache_path: Path, max_age_hours: int = 2) -> bool:
+    """
+    Check if the cache file exists and is recent enough
+    Args:
+        cache_path: Path to the cache file
+        max_age_hours: Maximum age of the cache in hours
+    Returns:
+        True if cache exists and is recent enough, False otherwise
+    """
+    if not cache_path.exists():
+        return False
+    
+    file_mod_time = datetime.fromtimestamp(cache_path.stat().st_mtime)
+    max_age = timedelta(hours=max_age_hours)
+    
+    return datetime.now() - file_mod_time < max_age
+
+def read_cache(channel: Union[str, int], 
+            content_type: str = 'rss', 
+            max_age_hours: int = 2) -> Union[str, None]:
+    """
+    Read content from cache if it exists and is valid
+    Args:
+        channel: Channel name or ID
+        content_type: Type of content (rss or html)
+        max_age_hours: Maximum age of the cache in hours
+    Returns:
+        Cached content as string or None if cache is invalid/missing
+    """
+    cache_path = get_cache_path(channel, content_type)
+    
+    if not is_cache_valid(cache_path, max_age_hours):
+        return None
+    
+    try:
+        logger.debug(f"using_cached_{content_type}: channel {channel}, cache_file {cache_path}")
+        with open(cache_path, 'r', encoding='utf-8') as f:
+            return f.read()
+    except Exception as e:
+        logger.error(f"cache_read_error: channel {channel}, error {str(e)}")
+        return None
+
+def write_cache(channel: Union[str, int], 
+             content: str, 
+             content_type: str = 'rss') -> bool:
+    """
+    Write content to cache
+    Args:
+        channel: Channel name or ID
+        content: Content to cache
+        content_type: Type of content (rss or html)
+    Returns:
+        True if cache was written successfully, False otherwise
+    """
+    cache_path = get_cache_path(channel, content_type)
+    
+    try:
+        with open(cache_path, 'w', encoding='utf-8') as f:
+            f.write(content)
+        logger.debug(f"{content_type}_cache_saved: channel {channel}, cache_file {cache_path}")
+        return True
+    except Exception as e:
+        logger.error(f"cache_write_error: channel {channel}, error {str(e)}")
+        return False
+
+def get_access_file_path() -> str:
+    """
+    Get path to the cache access file
+    Returns:
+        Path to the cache access file
+    """
+    return os.path.join(os.path.abspath("./data"), 'cache_access.json')
+
+def update_channel_access_time(channel: Union[str, int]) -> bool:
+    """
+    Update access time for a channel in cache_access.json
+    Args:
+        channel: Channel name or ID
+    Returns:
+        True if successful, False otherwise
+    """
+    try:
+        access_file_path = get_access_file_path()
+        cache_access_data = {}
+        
+        # Load existing data if file exists
+        if os.path.exists(access_file_path):
+            try:
+                with open(access_file_path, 'r', encoding='utf-8') as f:
+                    cache_access_data = json.load(f)
+            except json.JSONDecodeError:
+                logger.error(f"Error decoding cache_access.json, creating new file")
+                cache_access_data = {}
+        
+        # Update channel access time
+        channel_key = str(channel)
+        cache_access_data[channel_key] = time.time()
+        
+        # Save updated data
+        with open(access_file_path, 'w', encoding='utf-8') as f:
+            json.dump(cache_access_data, f, indent=2)
+        
+        logger.debug(f"Updated access time for channel {channel} in cache_access.json")
+        return True
+    except Exception as e:
+        logger.error(f"Failed to update cache_access.json: {str(e)}")
+        return False
+
+async def clean_old_channels(max_age_days: int = 30) -> int:
+    """
+    Remove channels that haven't been accessed for more than specified days
+    
+    Args:
+        max_age_days: Maximum age of channels in days
+    
+    Returns:
+        Number of removed channels
+    """
+    access_file_path = get_access_file_path()
+    
+    if not os.path.exists(access_file_path):
+        logger.info("access_file_not_found: Creating new file")
+        with open(access_file_path, 'w', encoding='utf-8') as f:
+            json.dump({}, f)
+        return 0
+    
+    try:
+        with open(access_file_path, 'r', encoding='utf-8') as f:
+            cache_access = json.load(f)
+        
+        now = time.time()
+        max_age_seconds = max_age_days * 86400  # days to seconds
+        
+        old_channels = []
+        for channel, timestamp in list(cache_access.items()):
+            age = now - timestamp
+            if age > max_age_seconds:
+                old_channels.append(channel)
+                del cache_access[channel]
+        
+        # Save updated data
+        if old_channels:
+            with open(access_file_path, 'w', encoding='utf-8') as f:
+                json.dump(cache_access, f, indent=2)
+            logger.info(f"removed_old_channels: {len(old_channels)} channels removed (older than {max_age_days} days)")
+        
+        return len(old_channels)
+    
+    except Exception as e:
+        logger.error(f"clean_old_channels_error: {str(e)}")
+        return 0
+
+async def update_rss_cache(client, channel: str, limit: int = 50) -> bool:
+    """
+    Update RSS cache for specified channel
+    
+    Args:
+        client: Telegram client
+        channel: Channel identifier
+        limit: Maximum number of posts
+    
+    Returns:
+        True if cache was successfully updated, False otherwise
+    """
+    try:
+        logger.info(f"updating_rss_cache: channel {channel}")
+        
+        # Import here to avoid circular imports
+        from rss_generator import generate_channel_rss
+        
+        # Force generate new RSS without using cache
+        await generate_channel_rss(
+            channel=channel,
+            client=client,
+            use_cache=False,  # Force generate new RSS
+            limit=limit
+        )
+        
+        logger.info(f"updated_rss_cache: channel {channel}")
+        return True
+    except Exception as e:
+        logger.error(f"rss_cache_update_error: channel {channel}, error {str(e)}")
+        return False
+
+async def start_cache_updater(client) -> None:
+    """
+    Background task for updating RSS cache periodically
+    
+    Args:
+        client: Telegram client instance
+    """
+    # Settings from configuration
+    update_interval = Config.get("rss_cache_update_interval", 3600)  # seconds
+    update_delay = Config.get("rss_cache_update_delay", 60)  # seconds
+    max_age_days = Config.get("rss_cache_max_age_days", 30)
+    
+    logger.info(f"starting_rss_cache_updater: interval={update_interval}s, delay={update_delay}s, max_age={max_age_days}d")
+    
+    try:
+        while True:
+            # Clean old channels
+            cleaned = await clean_old_channels(max_age_days)
+            
+            # Load channels for update
+            access_file_path = get_access_file_path()
+            try:
+                if os.path.exists(access_file_path):
+                    with open(access_file_path, 'r', encoding='utf-8') as f:
+                        cache_access = json.load(f)
+                    channels = list(cache_access.keys())
+                    
+                    if channels:
+                        logger.info(f"starting_cache_update: updating {len(channels)} channels")
+                        
+                        # Sort channels by last access time (newest first)
+                        channels_sorted = sorted(channels, key=lambda c: cache_access[c], reverse=True)
+                        
+                        # Update cache for each channel with delay
+                        for channel in channels_sorted:
+                            success = await update_rss_cache(client, channel)
+                            # Sleep between updates to avoid rate limits
+                            await asyncio.sleep(update_delay)
+                        
+                        logger.info(f"finished_cache_update: updated {len(channels)} channels")
+                    else:
+                        logger.info("no_channels_to_update: Waiting for next interval")
+                else:
+                    logger.info("no_access_file: Waiting for next interval")
+            
+            except Exception as e:
+                logger.error(f"rss_cache_update_cycle_error: {str(e)}")
+            
+            # Wait until next update cycle
+            logger.info(f"waiting_for_next_update: next update in {update_interval} seconds")
+            await asyncio.sleep(update_interval)
+    
+    except asyncio.CancelledError:
+        logger.info("rss_cache_updater_cancelled")
+    except Exception as e:
+        logger.error(f"rss_cache_updater_error: {str(e)}") 

rss_generator.py

@@ -15,7 +15,7 @@ import time
 import os
 import hashlib
 from pathlib import Path
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 from types import SimpleNamespace
 from typing import Optional
 from feedgen.feed import FeedGenerator
@@ -23,6 +23,7 @@ from pyrogram import errors, Client
 from pyrogram.types import Message
 from post_parser import PostParser
 from config import get_settings
+from rss_cache import read_cache, write_cache
 
 Config = get_settings()
 
@@ -301,47 +302,6 @@ async def _render_messages_groups(messages_groups: list[list[Message]],
     rendered_posts.sort(key=lambda x: x['date'], reverse=True)
     return rendered_posts
 
-def _get_cache_path(channel: str | int) -> Path:
-    """
-    Get the path to the cache file for a given channel
-    Args:
-        channel: Telegram channel name or id
-    Returns:
-        Path to the cache file
-    """
-    cache_dir = Path(Config.get('cache_dir', './data/rss-cache'))
-    cache_dir.mkdir(parents=True, exist_ok=True)
-    
-    # Create a unique filename based on the channel identifier
-    if isinstance(channel, int):
-        filename = f"channel_{channel}.xml"
-    else:
-        # Use hash for channel names with special characters
-        if any(c in channel for c in '/<>:"|?*\\'):
-            channel_hash = hashlib.md5(channel.encode()).hexdigest()
-            filename = f"channel_{channel_hash}.xml"
-        else:
-            filename = f"channel_{channel}.xml"
-    
-    return cache_dir / filename
-
-def _is_cache_valid(cache_path: Path, max_age_hours: int = 2) -> bool:
-    """
-    Check if the cache file exists and is recent enough
-    Args:
-        cache_path: Path to the cache file
-        max_age_hours: Maximum age of the cache in hours
-    Returns:
-        True if cache exists and is recent enough, False otherwise
-    """
-    if not cache_path.exists():
-        return False
-    
-    file_mod_time = datetime.fromtimestamp(cache_path.stat().st_mtime)
-    max_age = timedelta(hours=max_age_hours)
-    
-    return datetime.now() - file_mod_time < max_age
-
 async def generate_channel_rss(channel: str | int, 
                                 client: Client, 
                                 limit: int = 20, 
@@ -365,17 +325,11 @@ async def generate_channel_rss(channel: str | int,
     Returns:
         RSS feed as string in XML format
     """
-    cache_path = _get_cache_path(channel)
-    
     # Check if we can use cached version
-    if use_cache and _is_cache_valid(cache_path, cache_max_age_hours):
-        try:
-            logger.debug(f"using_cached_rss: channel {channel}, cache_file {cache_path}")
-            with open(cache_path, 'r', encoding='utf-8') as f:
-                return f.read()
-        except Exception as e:
-            logger.error(f"cache_read_error: channel {channel}, error {str(e)}")
-            # If reading cache fails, continue to generate new content
+    if use_cache:
+        cached_content = read_cache(channel, 'rss', cache_max_age_hours)
+        if cached_content:
+            return cached_content
     
     total_start_time = time.time()
     
@@ -473,12 +427,7 @@ async def generate_channel_rss(channel: str | int,
         
         # Save to cache
         if use_cache:
-            try:
-                with open(cache_path, 'w', encoding='utf-8') as f:
-                    f.write(rss_feed)
-                logger.debug(f"rss_cache_saved: channel {channel}, cache_file {cache_path}")
-            except Exception as e:
-                logger.error(f"cache_write_error: channel {channel}, error {str(e)}")
+            write_cache(channel, rss_feed, 'rss')
         
         total_elapsed = time.time() - total_start_time
         logger.info(f"rss_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")
@@ -488,7 +437,7 @@ async def generate_channel_rss(channel: str | int,
         logger.error(f"generate_channel_rss: channel {channel}, error {str(e)}")
         raise
 
-async def _reply_enrichment(client: Client, messages: list[Message]) -> list[Message]:
+async def _reply_enrichment(client: Client, messages: list[Message]) -> list[Message]: #TODO: почему это тут?
     """
     Enrich messages with reply to messages
     """
@@ -527,32 +476,6 @@ async def generate_channel_html(channel: str | int,
     Returns:
         HTML feed as string
     """
-    # Use different directory for html cache
-    cache_dir = Path(Config.get('cache_dir', './data/html-cache'))
-    cache_dir.mkdir(parents=True, exist_ok=True)
-    
-    # Create a unique filename for the html cache
-    if isinstance(channel, int):
-        filename = f"channel_{channel}.html"
-    else:
-        # Use hash for channel names with special characters
-        if any(c in channel for c in '/<>:"|?*\\'):
-            channel_hash = hashlib.md5(channel.encode()).hexdigest()
-            filename = f"channel_{channel_hash}.html"
-        else:
-            filename = f"channel_{channel}.html"
-    
-    cache_path = cache_dir / filename
-    
-    # Check if we can use cached version
-    if use_cache and _is_cache_valid(cache_path, cache_max_age_hours):
-        try:
-            logger.debug(f"using_cached_html: channel {channel}, cache_file {cache_path}")
-            with open(cache_path, 'r', encoding='utf-8') as f:
-                return f.read()
-        except Exception as e:
-            logger.error(f"html_cache_read_error: channel {channel}, error {str(e)}")
-            # If reading cache fails, continue to generate new content
     
     total_start_time = time.time()
     
@@ -627,14 +550,6 @@ async def generate_channel_html(channel: str | int,
         html_gen_elapsed = time.time() - html_gen_start_time
         logger.info(f"html_generation_timing: channel {channel}, HTML generated in {html_gen_elapsed:.3f} seconds")
         
-        # Save to cache
-        if use_cache:
-            try:
-                with open(cache_path, 'w', encoding='utf-8') as f:
-                    f.write(html_content)
-                logger.debug(f"html_cache_saved: channel {channel}, cache_file {cache_path}")
-            except Exception as e:
-                logger.error(f"html_cache_write_error: channel {channel}, error {str(e)}")
         
         total_elapsed = time.time() - total_start_time
         logger.info(f"html_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")