17 Commits

Author SHA1 Message Date
vvzvlad a2d67340a9 skip story messages 2025-05-11 15:30:45 +03:00
vvzvlad b5c52ddd70 refactor PostParser: improve story media handling by using original story sender channel for file paths and adding error logging for channel extraction 2025-05-10 21:57:51 +03:00
vvzvlad e0e35cd879 enhance PostParser: add support for story media handling, including video and photo differentiation 2025-05-10 21:48:01 +03:00
vvzvlad c27c3814b9 implement story media file saving logic in PostParser 2025-05-10 21:37:40 +03:00
vvzvlad 79105b713f add support for handling stories 2025-05-10 21:20:41 +03:00
vvzvlad f4774f9253 add semaphore for limiting concurrent Telegram API requests in get_chat_history 2025-04-30 14:47:28 +04:00
vvzvlad 740104bb7b refactor TelegramClient: remove disconnection handler and related exit logic 2025-04-30 14:39:09 +04:00
vvzvlad 1ec87cba14 refactor TelegramClient: simplify disconnection handling 2025-04-29 19:48:11 +04:00
vvzvlad f009c413e6 enhance TelegramClient: implement auto-reconnect logic with disconnect tracking and delay, improving resilience against connection issues 2025-04-29 15:22:36 +04:00
vvzvlad 7e6aa1f7ba refactor TelegramClient: remove signal handling and restart logic, replace app restart with process exit on disconnect limit 2025-04-29 15:10:13 +04:00
vvzvlad 5ac093c0bc implement RSS cache update mechanism 2025-04-29 14:34:32 +04:00
vvzvlad 1eaa99c3d4 add signal handling to TelegramClient for graceful shutdown on termination signals 2025-04-28 23:51:16 +04:00
vvzvlad a96c018a19 update cache directory paths 2025-04-28 13:33:07 +04:00
vvzvlad 4ddb6a7545 remove timing logs from get_rss_feed in api_server 2025-04-28 13:29:46 +04:00
vvzvlad 169578051e change debug logs to info level for timing metrics in rss_generator and html generation functions 2025-04-28 13:08:15 +04:00
vvzvlad d766220025 implement caching for RSS 2025-04-27 20:32:31 +04:00
vvzvlad ef945ff767 refactor TelegramClient: change _on_disconnect and _restart_app methods to async for better handling of asynchronous operations 2025-04-27 18:45:35 +04:00
5 changed files with 363 additions and 63 deletions
+7 -5
View File
@@ -36,6 +36,7 @@ from config import get_settings, setup_logging
from rss_generator import generate_channel_rss, generate_channel_html from rss_generator import generate_channel_rss, generate_channel_html
from post_parser import PostParser from post_parser import PostParser
from url_signer import verify_media_digest, generate_media_digest 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 # Define custom exception for zero-size files
class ZeroSizeFileError(Exception): class ZeroSizeFileError(Exception):
@@ -72,10 +73,13 @@ async def lifespan(_: FastAPI):
await client.start() await client.start()
background_task = asyncio.create_task(cache_media_files()) # Start background task 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 yield
background_task.cancel() # Cleanup background_task.cancel() # Cleanup
cache_updater_task.cancel()
try: try:
await background_task await background_task
await cache_updater_task
except asyncio.CancelledError: except asyncio.CancelledError:
pass pass
await client.stop() await client.stop()
@@ -786,9 +790,11 @@ async def get_rss_feed(channel: str,
elif Config["token"] and is_local_request(request): elif Config["token"] and is_local_request(request):
logger.info(f"Local request, skipping token check for RSS endpoint.") logger.info(f"Local request, skipping token check for RSS endpoint.")
# Обновляем время доступа к каналу в cache_access.json
update_channel_access_time(channel)
while True: while True:
try: try:
start_time = time.time()
if output_type == 'rss': if output_type == 'rss':
rss_content = await generate_channel_rss(channel, rss_content = await generate_channel_rss(channel,
@@ -797,8 +803,6 @@ async def get_rss_feed(channel: str,
exclude_flags=exclude_flags, exclude_flags=exclude_flags,
exclude_text=exclude_text, exclude_text=exclude_text,
merge_seconds=merge_seconds) merge_seconds=merge_seconds)
elapsed_time = time.time() - start_time
logger.info(f"rss_generation_timing: channel {channel}, generated in {elapsed_time:.3f} seconds")
return Response(content=rss_content, media_type="application/xml") return Response(content=rss_content, media_type="application/xml")
elif output_type == 'html': elif output_type == 'html':
rss_content = await generate_channel_html(channel, rss_content = await generate_channel_html(channel,
@@ -807,8 +811,6 @@ async def get_rss_feed(channel: str,
exclude_flags=exclude_flags, exclude_flags=exclude_flags,
exclude_text=exclude_text, exclude_text=exclude_text,
merge_seconds=merge_seconds) merge_seconds=merge_seconds)
elapsed_time = time.time() - start_time
logger.info(f"html_generation_timing: channel {channel}, generated in {elapsed_time:.3f} seconds")
return Response(content=rss_content, media_type="text/html") return Response(content=rss_content, media_type="text/html")
except ValueError as e: except ValueError as e:
error_message = f"invalid_parameters_error: {str(e)}" error_message = f"invalid_parameters_error: {str(e)}"
+3
View File
@@ -60,4 +60,7 @@ def get_settings() -> dict[str, Any]:
"time_based_merge": os.getenv("TIME_BASED_MERGE", "False").strip() in ["True", "true"], "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_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"], "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")),
} }
+296
View File
@@ -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)}")
+51 -18
View File
@@ -12,6 +12,10 @@
import logging import logging
import re import re
import time import time
import os
import hashlib
import asyncio
from pathlib import Path
from datetime import datetime, timezone from datetime import datetime, timezone
from types import SimpleNamespace from types import SimpleNamespace
from typing import Optional from typing import Optional
@@ -20,11 +24,15 @@ from pyrogram import errors, Client
from pyrogram.types import Message from pyrogram.types import Message
from post_parser import PostParser from post_parser import PostParser
from config import get_settings from config import get_settings
from rss_cache import read_cache, write_cache
Config = get_settings() Config = get_settings()
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Семафор для ограничения параллельных запросов к Telegram API при получении постов
POSTS_REQUEST_SEMAPHORE = asyncio.Semaphore(3)
async def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]: async def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]:
""" """
Create media groups based on time difference between messages Create media groups based on time difference between messages
@@ -87,6 +95,10 @@ async def _create_messages_groups(messages: list[Message]) -> list[list[Message]
# First pass - collect messages and organize into processing groups # First pass - collect messages and organize into processing groups
for message in messages: for message in messages:
try: try:
# Skip story messages
if message.media == "MessageMediaType.STORY" or hasattr(message, "story"):
continue
# Skip service messages about pinned posts and new chat photos # Skip service messages about pinned posts and new chat photos
if message.service: if message.service:
if 'PINNED_MESSAGE' in str(message.service): continue if 'PINNED_MESSAGE' in str(message.service): continue
@@ -303,7 +315,9 @@ async def generate_channel_rss(channel: str | int,
limit: int = 20, limit: int = 20,
exclude_flags: str | None = None, exclude_flags: str | None = None,
exclude_text: str | None = None, exclude_text: str | None = None,
merge_seconds: int = 5 merge_seconds: int = 5,
use_cache: bool = True,
cache_max_age_hours: int = 2
) -> str: ) -> str:
""" """
Generate RSS feed for channel using actual messages Generate RSS feed for channel using actual messages
@@ -314,9 +328,17 @@ async def generate_channel_rss(channel: str | int,
limit: Maximum number of posts to include in the RSS feed limit: Maximum number of posts to include in the RSS feed
exclude_flags: Flags to exclude from the RSS feed exclude_flags: Flags to exclude from the RSS feed
exclude_text: Text to exclude from posts exclude_text: Text to exclude from posts
use_cache: Whether to use cache
cache_max_age_hours: Maximum age of the cache in hours
Returns: Returns:
RSS feed as string in XML format RSS feed as string in XML format
""" """
# Check if we can use cached version
if use_cache:
cached_content = read_cache(channel, 'rss', cache_max_age_hours)
if cached_content:
return cached_content
total_start_time = time.time() total_start_time = time.time()
if limit < 1: if limit < 1:
@@ -349,9 +371,9 @@ async def generate_channel_rss(channel: str | int,
# Re-raise the original exception to be caught by the outer handler if needed, # Re-raise the original exception to be caught by the outer handler if needed,
# but add specific logging here. # but add specific logging here.
raise ValueError(f"Failed to get chat info for {channel}: {str(e)}") from e # Raise a more specific error perhaps raise ValueError(f"Failed to get chat info for {channel}: {str(e)}") from e # Raise a more specific error perhaps
finally:
channel_info_elapsed = time.time() - channel_info_start_time channel_info_elapsed = time.time() - channel_info_start_time
logger.debug(f"rss_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds") logger.info(f"rss_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds")
# Set feed metadata # Set feed metadata
main_name = f"{channel_title} (@{channel_username})" main_name = f"{channel_title} (@{channel_username})"
@@ -365,6 +387,7 @@ async def generate_channel_rss(channel: str | int,
messages_start_time = time.time() messages_start_time = time.time()
messages = [] messages = []
try: try:
async with POSTS_REQUEST_SEMAPHORE:
async for message in post_parser.client.get_chat_history(channel, limit=limit*2): async for message in post_parser.client.get_chat_history(channel, limit=limit*2):
messages.append(message) messages.append(message)
except Exception as e: except Exception as e:
@@ -372,7 +395,7 @@ async def generate_channel_rss(channel: str | int,
raise ValueError(f"Failed to get chat history for {channel}: {str(e)}") from e # Raise a more specific error raise ValueError(f"Failed to get chat history for {channel}: {str(e)}") from e # Raise a more specific error
messages_elapsed = time.time() - messages_start_time messages_elapsed = time.time() - messages_start_time
logger.debug(f"rss_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds") logger.info(f"rss_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
# Process messages into groups and render them # Process messages into groups and render them
processing_start_time = time.time() processing_start_time = time.time()
@@ -383,7 +406,7 @@ async def generate_channel_rss(channel: str | int,
final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text) final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
processing_elapsed = time.time() - processing_start_time processing_elapsed = time.time() - processing_start_time
logger.debug(f"rss_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds") logger.info(f"rss_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
# Generate feed entries # Generate feed entries
feed_gen_start_time = time.time() feed_gen_start_time = time.time()
@@ -405,21 +428,25 @@ async def generate_channel_rss(channel: str | int,
fe.author(name="", email=post['author']) fe.author(name="", email=post['author'])
feed_gen_elapsed = time.time() - feed_gen_start_time feed_gen_elapsed = time.time() - feed_gen_start_time
logger.debug(f"rss_feed_generation_timing: channel {channel}, feed generated in {feed_gen_elapsed:.3f} seconds") logger.info(f"rss_feed_generation_timing: channel {channel}, feed generated in {feed_gen_elapsed:.3f} seconds")
rss_feed = fg.rss_str(pretty=True) rss_feed = fg.rss_str(pretty=True)
if isinstance(rss_feed, bytes): if isinstance(rss_feed, bytes):
return rss_feed.decode('utf-8') rss_feed = rss_feed.decode('utf-8')
# Save to cache
if use_cache:
write_cache(channel, rss_feed, 'rss')
total_elapsed = time.time() - total_start_time total_elapsed = time.time() - total_start_time
logger.debug(f"rss_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds") logger.info(f"rss_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")
return rss_feed return rss_feed
except Exception as e: except Exception as e:
logger.error(f"generate_channel_rss: channel {channel}, error {str(e)}") logger.error(f"generate_channel_rss: channel {channel}, error {str(e)}")
raise 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 Enrich messages with reply to messages
""" """
@@ -440,7 +467,9 @@ async def generate_channel_html(channel: str | int,
limit: int = 20, limit: int = 20,
exclude_flags: str | None = None, exclude_flags: str | None = None,
exclude_text: str | None = None, exclude_text: str | None = None,
merge_seconds: int = 5 merge_seconds: int = 5,
use_cache: bool = True,
cache_max_age_hours: int = 2
) -> str: ) -> str:
""" """
Generate HTML feed for channel using actual messages Generate HTML feed for channel using actual messages
@@ -451,9 +480,12 @@ async def generate_channel_html(channel: str | int,
limit: Maximum number of posts to include in the RSS feed limit: Maximum number of posts to include in the RSS feed
exclude_flags: Flags to exclude from the RSS feed exclude_flags: Flags to exclude from the RSS feed
exclude_text: Text to exclude from posts exclude_text: Text to exclude from posts
use_cache: Whether to use cache
cache_max_age_hours: Maximum age of the cache in hours
Returns: Returns:
HTML feed as string HTML feed as string
""" """
total_start_time = time.time() total_start_time = time.time()
if limit < 1: if limit < 1:
@@ -485,7 +517,7 @@ async def generate_channel_html(channel: str | int,
raise ValueError(f"Failed to get chat info for {channel} in HTML generation: {str(e)}") from e raise ValueError(f"Failed to get chat info for {channel} in HTML generation: {str(e)}") from e
channel_info_elapsed = time.time() - channel_info_start_time channel_info_elapsed = time.time() - channel_info_start_time
logger.debug(f"html_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds") logger.info(f"html_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds")
# Collect messages # Collect messages
messages_start_time = time.time() messages_start_time = time.time()
@@ -498,13 +530,13 @@ async def generate_channel_html(channel: str | int,
raise ValueError(f"Failed to get chat history for {channel} in HTML generation: {str(e)}") from e raise ValueError(f"Failed to get chat history for {channel} in HTML generation: {str(e)}") from e
messages_elapsed = time.time() - messages_start_time messages_elapsed = time.time() - messages_start_time
logger.debug(f"html_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds") logger.info(f"html_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
# Enrich messages with reply to messages # Enrich messages with reply to messages
enrichment_start_time = time.time() enrichment_start_time = time.time()
messages = await _reply_enrichment(client, messages) messages = await _reply_enrichment(client, messages)
enrichment_elapsed = time.time() - enrichment_start_time enrichment_elapsed = time.time() - enrichment_start_time
logger.debug(f"html_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds") logger.info(f"html_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds")
# Process messages into groups and render them # Process messages into groups and render them
processing_start_time = time.time() processing_start_time = time.time()
@@ -517,20 +549,21 @@ async def generate_channel_html(channel: str | int,
final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text) final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
processing_elapsed = time.time() - processing_start_time processing_elapsed = time.time() - processing_start_time
logger.debug(f"html_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds") logger.info(f"html_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
# Generate HTML content # Generate HTML content
html_gen_start_time = time.time() html_gen_start_time = time.time()
html_posts = [post['html'] for post in final_posts] html_posts = [post['html'] for post in final_posts]
html = '\n<hr class="post-divider">\n'.join(html_posts) html_content = '\n<hr class="post-divider">\n'.join(html_posts)
html_gen_elapsed = time.time() - html_gen_start_time html_gen_elapsed = time.time() - html_gen_start_time
logger.debug(f"html_generation_timing: channel {channel}, HTML generated in {html_gen_elapsed:.3f} seconds") logger.info(f"html_generation_timing: channel {channel}, HTML generated in {html_gen_elapsed:.3f} seconds")
total_elapsed = time.time() - total_start_time total_elapsed = time.time() - total_start_time
logger.debug(f"html_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds") logger.info(f"html_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")
return html return html_content
except Exception as e: except Exception as e:
logger.error(f"html_generation_error: channel {channel}, error {str(e)}") logger.error(f"html_generation_error: channel {channel}, error {str(e)}")
+2 -36
View File
@@ -32,8 +32,6 @@ class TelegramClient:
api_hash=settings["tg_api_hash"], api_hash=settings["tg_api_hash"],
workdir=settings["session_path"], workdir=settings["session_path"],
) )
self.disconnect_count = 0
self.max_disconnects = 3
self._setup_connection_handlers() self._setup_connection_handlers()
def _ensure_session_directory(self): def _ensure_session_directory(self):
@@ -46,47 +44,15 @@ class TelegramClient:
def _setup_connection_handlers(self): def _setup_connection_handlers(self):
"""Sets up connection/disconnection handlers""" """Sets up connection/disconnection handlers"""
self.client.add_handler(DisconnectHandler(self._on_disconnect))
logger.info("connection_handlers: connection handlers set up") logger.info("connection_handlers: connection handlers set up")
def _on_disconnect(self, _client):
"""Handles disconnection from Telegram servers"""
self.disconnect_count += 1
logger.warning(f"connection_handler: connection lost (#{self.disconnect_count})")
if self.disconnect_count >= self.max_disconnects:
logger.critical(f"connection_handler: reached disconnect limit ({self.max_disconnects})")
self._restart_app()
async def start(self): async def start(self):
try: try:
if not self.client.is_connected: if not self.client.is_connected:
await self.client.start() await self.client.start()
logger.info("Telegram client connected successfully") logger.info("Telegram client connected successfully")
# Reset disconnect counter on successful connection
self.disconnect_count = 0
logger.info("connection_handler: connection established") logger.info("connection_handler: connection established")
except Exception as e: except Exception as e:
logger.error(f"Failed to start Telegram client: {str(e)}") logger.error(f"Failed to start Telegram client: {str(e)}")
raise # Force exit on any connection error
sys.exit(1)
def _restart_app(self):
"""Restarts the application"""
logger.warning("connection_handler: restarting application")
try:
# Properly shutdown client before restart
loop = asyncio.get_event_loop()
if loop.is_running():
loop.create_task(self.stop())
# Use os.execv to restart the process with the same arguments
os.execv(sys.executable, [sys.executable] + sys.argv)
except Exception as e:
logger.error(f"connection_handler: error during restart: {str(e)}")
# Emergency termination in case of restart failure
os.kill(os.getpid(), signal.SIGTERM)
async def stop(self):
if self.client.is_connected:
await self.client.stop()
logger.info("Telegram client disconnected")