is not in the allowed-tags whitelist and would be stripped).
if debug:
+ raw_escaped = html.escape(str(data.get("raw_message", "")))
html_content.append(f'{data["raw_message"]}')
+ f'white-space: pre-wrap;">{raw_escaped}')
html_data = '\n'.join(html_content)
return html_data
@@ -500,11 +514,38 @@ class PostParser:
return return_html
- def process_message(self, message: Message) -> Dict[Any, Any]:
- # Compute html body once — avoids triple _generate_html_body calls
+ def process_message(self, message: Message, include_raw: bool = False, sanitize: bool = True) -> Dict[Any, Any]:
+ """Build the processed representation of a message.
+
+ Args:
+ message: the Pyrogram message.
+ include_raw: when True, compute the (expensive) full ``str(message)``
+ serialization into ``result['raw_message']``. Only JSON output and
+ debug HTML need it; feed generation must pass False to avoid
+ serializing every post.
+ sanitize: when True, run the html body and footer through a single
+ bleach pass each. Single-post HTML and JSON need this (there is no
+ whole-feed pass on those paths). Feed generation passes False and
+ relies on the final whole-feed sanitize in rss_generator, so no
+ fragment is sanitized more than once.
+ """
+ # Compute html body once — avoids triple _generate_html_body calls.
+ # The internal per-fragment sanitize passes were removed (4.4); sanitize
+ # exactly once per output boundary here when requested.
html_body = self._generate_html_body(message)
+ # NOTE (stage-4 4.4 consequence): flags are now extracted from the PRE-sanitize
+ # body (the per-fragment sanitize that used to run inside _generate_html_body was
+ # removed). Legitimate links are unaffected — bleach keeps whitelisted
+ # — so link/foreign_channel/mention flags are identical for
+ # normal content. They can differ ONLY for URL-like text that bleach would strip
+ # (e.g. a URL inside a disallowed attribute); flags are non-security (used for
+ # exclude_flags filtering / display), so this edge divergence is accepted rather
+ # than re-adding a per-message sanitize pass that 4.4 deliberately eliminated.
flags = self._extract_flags(message, html_body=html_body)
footer = self.generate_html_footer(message, flags_list=flags)
+ if sanitize:
+ html_body = self._sanitize_html(html_body)
+ footer = self._sanitize_html(footer)
result = {
'channel': self.get_channel_username(message),
'message_id': message.id,
@@ -523,8 +564,10 @@ class PostParser:
'media_group_id': message.media_group_id,
'service': getattr(message, "service", None),
'show_caption_above_media': getattr(message, 'show_caption_above_media', False),
- 'raw_message': str(message)
}
+ if include_raw:
+ # Full serialization of the message — expensive; only for JSON/debug.
+ result['raw_message'] = str(message)
# Add webpage data if available
if webpage := getattr(message, "web_page", None):
@@ -671,8 +714,10 @@ class PostParser:
if message.forward_origin: content_body.append(f"--- Forwarded post end ---") # Forward info end
content_body.append(f'') + # NOTE: sanitize is NOT applied here. Sanitization happens exactly once per + # output boundary (process_message for single-post/JSON, final whole-feed pass + # in rss_generator for feeds). See the sanitize coverage map (4.4). html_body = '\n'.join(content_body) - html_body = self._sanitize_html(html_body) return html_body def _generate_html_media(self, message: Message) -> str: @@ -751,8 +796,9 @@ class PostParser: content_media.append(webpage_html) content_media.append('') + # Not sanitized here — this fragment is embedded in the body and sanitized + # once at the output boundary (see the 4.4 sanitize coverage map). html_media = '\n'.join(content_media) - html_media = self._sanitize_html(html_media) return html_media @@ -832,8 +878,9 @@ class PostParser: flags_html = self._format_flags(current_flags) content_footer.append('
' + flags_html) + # Not sanitized here — sanitized once at the output boundary (4.4 coverage map): + # process_message for single-post/JSON, final whole-feed pass for feeds. html_footer = '\n'.join(content_footer) - html_footer = self._sanitize_html(html_footer) return html_footer @@ -919,8 +966,10 @@ class PostParser: if links: parts.append(' | '.join(links)) + # Raw fragment — embedded in the footer, sanitized once at the output + # boundary (see the 4.4 sanitize coverage map). Not sanitized here. result_html = '
'.join(parts) if parts else None - return self._sanitize_html(result_html) if result_html else None + return result_html if result_html else None except Exception as e: logger.error(f"reactions_views_links_error: message_id {message.id}, error {str(e)}") @@ -969,15 +1018,31 @@ class PostParser: logger.error(f"file_id_extraction_error: media_type {message.media}, error {str(e)}") return None - async def _persist_media_file_id_async(self, channel: str, post_id: int, file_unique_id: str, added: float) -> None: - """Persist a single media file ID record to SQLite (fire-and-forget).""" + async def _flush_pending_media_ids(self) -> None: + """Persist media file-id records collected during rendering with ONE bulk upsert. + + Called by the caller (get_post / rss_generator) after rendering completes. + Runs the blocking SQLite write in a thread. No-op when nothing was collected. + """ + entries = self._pending_media_ids + if not entries: + return try: - await asyncio.to_thread(upsert_media_file_id_sync, DB_PATH, channel, post_id, file_unique_id, added) - logger.debug(f"persist_media_file_id: upserted {channel}/{post_id}/{file_unique_id}") + await asyncio.to_thread(upsert_media_file_ids_bulk_sync, DB_PATH, entries) + logger.debug(f"persist_media_file_ids_bulk: upserted {len(entries)} records") except Exception as e: - logger.error(f"file_id_save_error: error upserting {channel}/{post_id}/{file_unique_id}, error {str(e)}") + logger.error(f"file_id_bulk_save_error: error bulk-upserting {len(entries)} records, error {str(e)}") + finally: + # Clear regardless of outcome so a retry does not double-persist a stale batch. + self._pending_media_ids = [] def _save_media_file_ids(self, message: Message) -> None: + """Collect a media file-id record for later bulk persistence. + + IMPORTANT (4.2): this runs inside the render thread, so it must NOT touch + asyncio or the DB. It only appends to self._pending_media_ids; the caller + flushes the batch via _flush_pending_media_ids() after rendering. + """ try: channel_username = self.get_channel_username(message) if not channel_username: @@ -1003,29 +1068,8 @@ class PostParser: if file_unique_id: added_ts = datetime.now().timestamp() - 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) - ) - _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) - except Exception as e: - logger.error(f"file_id_save_error: error saving {channel_username}/{message.id}/{file_unique_id}, error {str(e)}") + # Thread-safe: just append; the caller persists the batch. + self._pending_media_ids.append((channel_username, message.id, file_unique_id, added_ts)) except Exception as e: logger.error(f"file_id_collection_error: message_id {message.id}, error {str(e)}") diff --git a/rss_generator.py b/rss_generator.py index a26ca11..c2af83b 100644 --- a/rss_generator.py +++ b/rss_generator.py @@ -23,6 +23,7 @@ from pyrogram.types import Message from post_parser import PostParser from config import get_settings from tg_throttle import tg_rpc_bounded +from file_io import upsert_media_file_ids_bulk_sync, DB_PATH from bleach.css_sanitizer import CSSSanitizer from bleach import clean as HTMLSanitizer @@ -30,9 +31,12 @@ Config = get_settings() logger = logging.getLogger(__name__) -async def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]: +def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]: """ Create media groups based on time difference between messages + + Plain synchronous function (contains no await): runs inside the render thread + via _render_pipeline. Must not touch asyncio. """ # Deep-copy the input list to avoid mutating cached Message objects (the cache may # reuse the same objects across calls with different merge_seconds values) @@ -92,9 +96,11 @@ async def _create_time_based_media_groups(messages: list[Message], merge_seconds return messages_sorted -async def _create_messages_groups(messages: list[Message]) -> list[list[Message]]: +def _create_messages_groups(messages: list[Message]) -> list[list[Message]]: """ Process messages into formatted posts, handling media groups + + Plain synchronous function (contains no await): runs inside the render thread. """ processing_groups: list[list[Message]] = [] media_groups: dict[str | int, list[Message]] = {} @@ -136,9 +142,11 @@ async def _create_messages_groups(messages: list[Message]) -> list[list[Message] return processing_groups -async def _trim_messages_groups(messages_groups: list[list[Message]], limit: int): +def _trim_messages_groups(messages_groups: list[list[Message]], limit: int): """ Trim messages groups to limit + + Plain synchronous function (contains no await): runs inside the render thread. """ if len(messages_groups) > limit: # Trim groups if they exceed the specified limit messages_groups = messages_groups[:limit] @@ -193,12 +201,13 @@ def processed_message_to_tg_message(processed_message: dict) -> Message: return tg_message_mock # type: ignore -async def _render_messages_groups(messages_groups: list[list[Message]], - post_parser: PostParser, - exclude_flags: str | None = None, +def _render_messages_groups(messages_groups: list[list[Message]], + post_parser: PostParser, + exclude_flags: str | None = None, exclude_text: str | None = None): """ Render message groups into HTML format + Plain synchronous function (contains no await): runs inside the render thread. Args: messages_groups: List of message groups (each group is a list of messages) post_parser: PostParser instance @@ -213,7 +222,9 @@ async def _render_messages_groups(messages_groups: list[list[Message]], try: if len(group) == 1: # Single message - simple case one_message = group[0] - message_data = post_parser.process_message(one_message) + # Feed path: raw_message not needed and sanitize deferred to the final + # whole-feed pass, so each fragment is sanitized exactly once. + message_data = post_parser.process_message(one_message, include_raw=False, sanitize=False) html_parts = [ f'', f'' @@ -228,7 +239,7 @@ async def _render_messages_groups(messages_groups: list[list[Message]], 'flags': message_data['flags'] }) else: # Multiple messages in group - merge text and html body - processed_messages = [post_parser.process_message(msg) for msg in group] + processed_messages = [post_parser.process_message(msg, include_raw=False, sanitize=False) for msg in group] # Determine main message for header/footer/title main_message = next( @@ -309,7 +320,31 @@ async def _render_messages_groups(messages_groups: list[list[Message]], rendered_posts.sort(key=lambda x: x['date'] if x['date'] is not None else 0.0, reverse=True) return rendered_posts -async def generate_channel_rss(channel: str | int, + +def _render_pipeline(messages: list[Message], + post_parser: PostParser, + limit: int, + exclude_flags: str | None, + exclude_text: str | None, + merge_seconds: int, + time_based_merge: bool): + """ + Full synchronous feed render pipeline (grouping + trimming + rendering). + + Runs entirely in a worker thread via a single asyncio.to_thread call. It contains + NO await, NO asyncio primitives, NO create_task/get_running_loop — all the CPU-heavy + work (deepcopy, grouping, bleach-free rendering) happens here off the event loop. + Media file-id records are accumulated on post_parser._pending_media_ids and flushed + by the caller after this returns. + """ + if time_based_merge: + messages = _create_time_based_media_groups(messages, merge_seconds) + message_groups = _create_messages_groups(messages) + message_groups = _trim_messages_groups(message_groups, limit) + return _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text) + + +async def generate_channel_rss(channel: str | int, client: Client, limit: int = 20, exclude_flags: str | None = None, @@ -389,14 +424,23 @@ async def generate_channel_rss(channel: str | int, 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") - # Process messages into groups and render them + # Process messages into groups and render them. + # The whole grouping/trimming/rendering pipeline is CPU-heavy (deepcopy + + # per-message rendering) and contains no await, so run it in ONE worker + # thread to keep the event loop responsive. processing_start_time = time.time() - if Config['time_based_merge']: - messages = await _create_time_based_media_groups(messages, merge_seconds) - message_groups = await _create_messages_groups(messages) - message_groups = await _trim_messages_groups(message_groups, limit) - final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text) - + try: + final_posts = await asyncio.to_thread( + _render_pipeline, messages, post_parser, limit, + exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'], + ) + finally: + # Persist media file-ids collected during rendering with a single bulk + # upsert — in a finally so a partial render still records what it collected + # (the flush is best-effort and swallows its own errors, so it cannot mask a + # render exception). + await post_parser._flush_pending_media_ids() + 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") @@ -601,16 +645,19 @@ async def generate_channel_html(channel: str | int, enrichment_elapsed = time.time() - enrichment_start_time logger.debug(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 in ONE worker thread + # (CPU-heavy, no await) to keep the event loop responsive. processing_start_time = time.time() - if Config['time_based_merge']: - messages = await _create_time_based_media_groups(messages, merge_seconds) + try: + final_posts = await asyncio.to_thread( + _render_pipeline, messages, post_parser, limit, + exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'], + ) + finally: + # Persist media file-ids collected during rendering with a single bulk + # upsert — in a finally so a partial render still records what it collected. + await post_parser._flush_pending_media_ids() - # Process messages into groups and render them - message_groups = await _create_messages_groups(messages) - message_groups = await _trim_messages_groups(message_groups, limit) - final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text) - 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") diff --git a/tests/test_stage4_eventloop.py b/tests/test_stage4_eventloop.py new file mode 100644 index 0000000..1b091a2 --- /dev/null +++ b/tests/test_stage4_eventloop.py @@ -0,0 +1,367 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Stage 4 (event-loop hygiene for feed generation) tests. + +Covers: +- 4.1 raw_message laziness: feeds do NOT compute str(message); JSON/debug HTML do. +- 4.2 side-effect IO removed from process_message: _save_media_file_ids only appends to + self._pending_media_ids; the caller flushes once via upsert_media_file_ids_bulk_sync. + Also: the render path contains NO create_task / get_running_loop / to_thread. +- 4.3 render pipeline moved into a single thread: the four render functions are now plain + sync functions and actually execute off the main thread; deepcopy of a pickled Message + does not crash; a 100-message feed generates correctly. +- 4.4 sanitize coverage (XSS): a