diff --git a/api_server.py b/api_server.py index d60a36e..ea8ce36 100644 --- a/api_server.py +++ b/api_server.py @@ -190,13 +190,9 @@ async def find_file_id_in_message(message: Message, file_unique_id: str) -> Unio return None -def delayed_delete_file(file_path: str, delay: int = 300) -> None: - """ - Delete temporary file after a delay to ensure complete file delivery. - Delay is set to 5 minutes by default. - NOTE: Runs in a background thread via Starlette BackgroundTask. - """ - time.sleep(delay) +async def delayed_delete_file(file_path: str, delay: int = 300) -> None: + """Delete a temporary file after a delay. Runs as an async background task.""" + await asyncio.sleep(delay) try: os.remove(file_path) logger.info(f"Deleted temporary file {file_path} after delay of {delay} seconds") @@ -955,6 +951,29 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re try: # Wrap the download and prepare call import time as _time + + # Pre-semaphore cache check: serve already-cached files without acquiring the semaphore + base_cache_dir = os.path.abspath("./data/cache") + channel_dir = os.path.join(base_cache_dir, str(channel)) + post_dir = os.path.join(channel_dir, str(post_id)) + cache_path = os.path.join(post_dir, file_unique_id) + if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0: + # File is already in cache — skip semaphore and serve directly + logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}") + # Fire-and-forget timestamp update with error handling to avoid silent failures + async def _update_access(_ch, _pid, _fid): + try: + await asyncio.to_thread( + update_media_file_access_sync, + DB_PATH, str(_ch), _pid, _fid, + datetime.now().timestamp() + ) + except Exception as _e: + logger.warning(f"Failed to update access time for {_ch}/{_pid}/{_fid}: {_e}") + asyncio.create_task(_update_access(channel, post_id, file_unique_id)) + return await prepare_file_response(cache_path, request=request, + media_key=(str(channel), post_id, file_unique_id)) + _sem_wait_start = _time.monotonic() async with HTTP_DOWNLOAD_SEMAPHORE: # limit concurrent live HTTP downloads _sem_wait = _time.monotonic() - _sem_wait_start diff --git a/file_io.py b/file_io.py index 2925c34..965bc96 100644 --- a/file_io.py +++ b/file_io.py @@ -17,6 +17,7 @@ def _open_db(db_path: str) -> sqlite3.Connection: """Open a SQLite connection with WAL journal mode enabled.""" conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout = 5000") # Wait up to 5 seconds on lock return conn diff --git a/post_parser.py b/post_parser.py index ab6ca59..1e054e3 100644 --- a/post_parser.py +++ b/post_parser.py @@ -28,6 +28,9 @@ Config = get_settings() logger = logging.getLogger(__name__) +# Module-level counter for in-flight persist tasks (used only in diagnostics) +_persist_pending_count = 0 + #tests #http://127.0.0.1:8000/post/html/DragorWW_space/114 — video @@ -325,11 +328,15 @@ class PostParser: return result return None - def _extract_flags(self, message: Message) -> List[str]: #Tests: tests/postparser_extract_flags.py + def _extract_flags(self, message: Message, html_body: Optional[str] = None) -> List[str]: #Tests: tests/postparser_extract_flags.py # Use raw text/caption for some checks before HTML processing message_text_str = str(message.text or message.caption or '') # Use HTML body for checks involving formatted text or links within HTML - message_body_html = self._generate_html_body(message) + # If html_body is provided (pre-computed), use it directly to avoid redundant generation + if html_body is None: + message_body_html = self._generate_html_body(message) + else: + message_body_html = html_body flags = [] # Add "fwd" flag for forwarded messages @@ -491,6 +498,10 @@ class PostParser: return return_html def process_message(self, message: Message) -> Dict[Any, Any]: + # Compute html body once — avoids triple _generate_html_body calls + html_body = self._generate_html_body(message) + flags = self._extract_flags(message, html_body=html_body) + footer = self.generate_html_footer(message, flags_list=flags) result = { 'channel': self.get_channel_username(message), 'message_id': message.id, @@ -499,10 +510,10 @@ class PostParser: 'text': message.text or message.caption or '', 'html': { 'title': self._generate_title(message), - 'body': self._generate_html_body(message), - 'footer': self.generate_html_footer(message) + 'body': html_body, + 'footer': footer }, - 'flags': self._extract_flags(message), + 'flags': flags, 'author': self._get_author_info(message), 'views': message.views, 'reactions': self._extract_reactions(message), @@ -992,16 +1003,21 @@ class PostParser: try: loop = asyncio.get_running_loop() if loop.is_running(): + global _persist_pending_count task = loop.create_task( self._persist_media_file_id_async(channel_username, message.id, file_unique_id, added_ts) ) - # Count pending fire-and-forget tasks to surface backlog issues - all_tasks = asyncio.all_tasks(loop) - persist_tasks = [t for t in all_tasks if '_persist_media_file_id_async' in str(t.get_coro())] - task_count = len(persist_tasks) - if task_count > 5: - logger.warning(f"persist_task_queue: {task_count} pending _persist_media_file_id_async tasks (msg {message.id})") - logger.debug(f"persist_task_created: queued for {channel_username}/{message.id}/{file_unique_id}, total pending: {task_count}") + _persist_pending_count += 1 + + def _on_task_done(t): + global _persist_pending_count + _persist_pending_count -= 1 + + task.add_done_callback(_on_task_done) + + if _persist_pending_count > 5: + logger.warning(f"persist_task_queue: {_persist_pending_count} pending _persist_media_file_id_async tasks (msg {message.id})") + logger.debug(f"persist_task_created: queued for {channel_username}/{message.id}/{file_unique_id}, total pending: {_persist_pending_count}") else: # Fallback sync path (should not occur during normal FastAPI runtime) upsert_media_file_id_sync(DB_PATH, channel_username, message.id, file_unique_id, added_ts) diff --git a/rss_generator.py b/rss_generator.py index 66a3059..6fe5777 100644 --- a/rss_generator.py +++ b/rss_generator.py @@ -477,17 +477,47 @@ async def generate_channel_rss(channel: str | int, async def _reply_enrichment(client: Client, messages: list[Message]) -> list[Message]: """ - Enrich messages with reply to messages + Enrich messages with reply-to messages. + + Instead of one API call per message, replies are batched: all message IDs + that need enrichment are grouped by chat_id and fetched in a single + client.get_messages() call per chat_id. """ + # Collect messages that need reply enrichment, grouped by chat_id + chat_messages: dict[int, list[Message]] = {} for message in messages: if message.reply_to_message_id and message.chat: - full_message = await client.get_messages(message.chat.id, message.id) - if isinstance(full_message, list): - if full_message and full_message[0].reply_to_message: - message.reply_to_message = full_message[0].reply_to_message - else: - if full_message and full_message.reply_to_message: - message.reply_to_message = full_message.reply_to_message + chat_id = message.chat.id + if chat_id not in chat_messages: + chat_messages[chat_id] = [] + chat_messages[chat_id].append(message) + + if not chat_messages: + # No messages with replies — return unchanged + return messages + + # Build a lookup: {(chat_id, message_id): full_message} using one batch call per chat + reply_lookup: dict[tuple[int, int], Message] = {} + for chat_id, chat_msgs in chat_messages.items(): + ids_to_fetch = [m.id for m in chat_msgs] + try: + fetched = await client.get_messages(chat_id, ids_to_fetch) + # get_messages may return a single Message or a list + if not isinstance(fetched, list): + fetched = [fetched] + for fm in fetched: + if fm and not getattr(fm, 'empty', False): + reply_lookup[(chat_id, fm.id)] = fm + except Exception as e: + logger.error(f"reply_enrichment_batch_error: chat_id {chat_id}, ids {ids_to_fetch}, error {str(e)}") + + # Apply reply_to_message from lookup + for message in messages: + if message.reply_to_message_id and message.chat: + key = (message.chat.id, message.id) + full_message = reply_lookup.get(key) + if full_message and full_message.reply_to_message: + message.reply_to_message = full_message.reply_to_message return messages diff --git a/tg_cache.py b/tg_cache.py index a9734a7..435f7cf 100644 --- a/tg_cache.py +++ b/tg_cache.py @@ -42,11 +42,11 @@ def _save_history_to_cache(channel_id: Union[str, int], messages: List[Message], try: cache_file = _get_history_cache_file_path(channel_id) - # Create cache metadata + # Create cache metadata — store messages directly (no inner pickle.dumps) cache_data = { 'timestamp': time.time(), 'limit': limit, - 'messages': pickle.dumps(messages) + 'messages': messages } with open(cache_file, 'wb') as f: @@ -94,8 +94,12 @@ def _get_history_from_cache(channel_id: Union[str, int], limit: int, max_age_hou logger.info(f"history_cache_limit_mismatch: channel {channel_id}, cached limit {cached_limit}, requested limit {limit}") return None - # Restore message list from pickle - messages = pickle.loads(cache_data['messages']) + # Restore message list; handle old cache files that used double-pickle (bytes = old format) + raw = cache_data['messages'] + if isinstance(raw, bytes): + messages = pickle.loads(raw) + else: + messages = raw logger.info(f"history_cache_hit: channel {channel_id}, limit {limit}, messages {len(messages)}, age {cache_age:.1f}s") return messages