diff --git a/post_parser.py b/post_parser.py
index ecf98eb..548636c 100644
--- a/post_parser.py
+++ b/post_parser.py
@@ -3,11 +3,12 @@ import copy
import re
import os
import json
-import bleach
from datetime import datetime
from typing import Union, Dict, Any, List
from pyrogram.types import Message
+from bleach.css_sanitizer import CSSSanitizer
+from bleach import clean as HTMLSanitizer
from pyrogram.enums import MessageMediaType
from config import get_settings
from url_signer import generate_media_digest
@@ -56,30 +57,6 @@ if not logger.handlers:
class PostParser:
def __init__(self, client):
self.client = client
- self.allowed_tags = ['p', 'a', 'b', 'i', 'strong', 'em', 'ul', 'ol', 'li', 'br', 'div', 'span', 'img', 'video', 'audio', 'source']
- self.allowed_attributes = {
- 'a': ['href', 'title', 'target'],
- 'img': {
- 'src': True,
- 'alt': True,
- 'style': ['max-width', 'max-height', 'object-fit']
- },
- 'video': {
- 'controls': True,
- 'src': True,
- 'style': ['max-width', 'max-height']
- },
- 'audio': {
- 'controls': True,
- 'style': ['width', 'max-width']
- },
- 'source': ['src', 'type'],
- 'div': {
- 'class': True,
- 'style': ['margin']
- },
- 'span': ['class']
- }
def _debug_message(self, message: Message) -> Message:
if Config["debug"]:
@@ -90,15 +67,19 @@ class PostParser:
debug_message.entities = None
print(debug_message)
return
+
+ def channel_name_prepare(self, channel: str):
+ if isinstance(channel, str) and channel.startswith('-100'): # Convert numeric channel ID to int
+ channel_id = int(channel)
+ return channel_id
+ else:
+ return channel
async def get_post(self, channel: str, post_id: int, output_type: str = 'json') -> Union[str, Dict[Any, Any]]:
print(f"Getting post {channel}, {post_id}")
try:
- channel_id = channel
- if isinstance(channel, str) and channel.startswith('-100'): # Convert numeric channel ID to int
- channel_id = int(channel)
-
- message = await self.client.get_messages(channel_id, post_id)
+ channel = self.channel_name_prepare(channel)
+ message = await self.client.get_messages(channel, post_id)
self._debug_message(message)
@@ -109,7 +90,7 @@ class PostParser:
if output_type == 'html':
return self._format_html(message)
elif output_type == 'json':
- return self._format_json(message)
+ return self.process_message(message)
else:
logger.error(f"Invalid output type: {output_type}")
return None
@@ -160,7 +141,7 @@ class PostParser:
text = '\n'.join(line.strip() for line in text.split('\n') if line.strip())
# Get first non-empty line
- first_line = text.split('\n')[0] if text else ""
+ first_line = text.split('\n', maxsplit=1)[0] if text else ""
max_length = 100
if len(first_line) <= max_length:
@@ -210,74 +191,113 @@ class PostParser:
return None
+ def _extract_reactions(self, message: Message) -> Union[str, None]:
+ if reactions := getattr(message, 'reactions', None):
+ return {r.emoji: r.count for r in reactions.reactions}
+ return None
- def format_message_for_feed(self, message: Message, top_info: bool = True, bottom_info: bool = True) -> Dict[Any, Any]:
- return self._format_json(message, top_info=top_info, bottom_info=bottom_info)
- def _format_json(self, message: Message, top_info: bool = True, bottom_info: bool = True) -> Dict[Any, Any]:
- html_content = self._format_html(message, top_info=top_info, bottom_info=bottom_info)
+ def _format_html(self, message: Message) -> str:
+ html_content = []
+ data = self.process_message(message)['html']
+
+ html_content.append(f'
')
+ html_content.append(f'{data["media"]}
')
+ html_content.append(f'{data["body"]}
')
+ html_content.append(f'')
+ html_content = '\n'.join(html_content)
+ return html_content
+
+
+ def process_message(self, message: Message) -> Dict[Any, Any]:
result = {
'channel': self.get_channel_username(message),
'message_id': message.id,
'date': datetime.timestamp(message.date),
'date_formatted': message.date.strftime('%Y-%m-%d %H:%M:%S'),
'text': message.text or message.caption or '',
- 'html': html_content,
- 'title': self._generate_title(message),
+ 'html': {
+ 'title': self._generate_title(message),
+ 'header': self._generate_html_header(message),
+ 'body': self._generate_html_body(message),
+ 'media': self._generate_html_media(message),
+ 'footer': self._generate_html_footer(message)
+ },
'author': self._get_author_info(message),
'views': message.views,
+ 'reactions': self._extract_reactions(message),
+ 'media_group_id': message.media_group_id
}
-
- if message.media_group_id:
- result['media_group_id'] = message.media_group_id
-
+
return result
def _sanitize_html(self, html: str) -> str:
+ allowed_tags = ['p', 'a', 'b', 'i', 'strong', 'em', 'ul', 'ol', 'li', 'br', 'div', 'span', 'img', 'video', 'audio', 'source']
+ allowed_attributes = {
+ 'a': ['href', 'title', 'target'],
+ 'img': {'src': True, 'alt': True, 'style': True},
+ 'video': {'controls': True, 'src': True, 'style': True},
+ 'audio': {'controls': True, 'style': True},
+ 'source': ['src', 'type'],
+ 'div': {'class': True, 'style': True},
+ 'span': ['class']
+ }
+
try:
- return bleach.clean(
+ css_sanitizer = CSSSanitizer(
+ allowed_css_properties=["max-width", "max-height", "object-fit", "width", "height"]
+ )
+ sanitized_html = HTMLSanitizer(
html,
- tags=self.allowed_tags,
- attributes=self.allowed_attributes,
+ tags=allowed_tags,
+ attributes=allowed_attributes,
+ css_sanitizer=css_sanitizer,
strip=True,
)
+ return sanitized_html
except Exception as e:
logger.error(f"html_sanitization_error: {str(e)}")
return html
- def _format_html(self, message: Message, top_info: bool = True, bottom_info: bool = True) -> str:
- html_content = []
-
- # Create service message for "channel created"
- if getattr(message, "channel_chat_created", False):
- html_content.append('Channel created
')
- html = '\n'.join(html_content)
- return self._sanitize_html(html)
-
+ def _generate_html_header(self, message: Message) -> str:
+ content_header = []
# Add forwarded from or reply info if present
- if top_info:
- if forward_html := self._format_forward_info(message):
- html_content.append(forward_html)
- elif reply_html := self._format_reply_info(message):
- html_content.append(reply_html)
+ if forward_html := self._format_forward_info(message):
+ content_header.append(forward_html)
+ elif reply_html := self._format_reply_info(message):
+ content_header.append(reply_html)
+ html_header = '\n'.join(content_header)
+ html_header = self._sanitize_html(html_header)
+ return html_header
- if message.text:
- text = message.text.html
- elif message.caption:
- text = message.caption.html
- else:
- text = ''
+ def _generate_html_body(self, message: Message) -> str:
+ content_body = []
+ if message.text: text = message.text.html
+ elif message.caption: text = message.caption.html
+ else: text = ''
text = text.replace('\n', '
') # Replace newlines with
text = self._add_hyperlinks_to_raw_urls(text) # Add hyperlinks to raw URLs
- self._save_media_file_ids(message) # Save media file_ids for caching
-
- if poll := getattr(message, "poll", None): # Poll formatting
+ if text: # Message text
+ content_body.append(f'{text}
')
+
+ if poll := getattr(message, "poll", None): # Poll message
if poll_html := self._format_poll(poll):
- html_content.append(poll_html)
-
+ content_body.append(poll_html)
+
+ if getattr(message, "channel_chat_created", False): # Create service message for "channel created"
+ content_body.append('Channel created
')
+
+ html_body = '\n'.join(content_body)
+ html_body = self._sanitize_html(html_body)
+ return html_body
+
+ def _generate_html_media(self, message: Message) -> str:
+ self._save_media_file_ids(message) # Save media file_ids for caching
+
+ content_media = []
base_url = Config['pyrogram_bridge_url']
if message.media and message.media != "MessageMediaType.POLL":
file_unique_id = self._get_file_unique_id(message)
@@ -290,54 +310,51 @@ class PostParser:
url = f"{base_url}/media/{file}/{digest}"
logger.debug(f"Collected media file: {channel_username}/{message.id}/{file_unique_id}")
- html_content.append(f'')
if webpage := getattr(message, "web_page", None): # Web page preview
if webpage_html := self._format_webpage(webpage, message):
- html_content.append(webpage_html)
+ content_media.append(webpage_html)
- if text: # Message text
- html_content.append(f'{text}
')
+ html_media = '\n'.join(content_media)
+ html_media = self._sanitize_html(html_media)
+ return html_media
+
+ def _generate_html_footer(self, message: Message) -> str:
+ content_footer = []
+ if reactions_views_html := self._reactions_views_links(message): # Add reactions, views and links
+ content_footer.append(reactions_views_html)
+ html_footer = '\n'.join(content_footer)
+ html_footer = self._sanitize_html(html_footer)
+ return html_footer
- if bottom_info:
- if reactions_views_html := self._reactions_views_links(message): # Add reactions, views and links
- html_content.append(reactions_views_html)
-
- html = '\n'.join(html_content)
-
- return self._sanitize_html(html)
def _add_hyperlinks_to_raw_urls(self, text: str) -> str:
try:
- # Find all existing tags
- a_tags = re.finditer(r']*>.*?', text)
- # Store their positions
- excluded_ranges = [(m.start(), m.end()) for m in a_tags]
-
- # Find all URLs that are not already in HTML tags
+ a_tags = re.finditer(r']*>.*?', text) # Find all existing tags
+ excluded_ranges = [(m.start(), m.end()) for m in a_tags] # Store their positions
result = text
offset = 0
- for match in re.finditer(r'https?://[^\s<>"\']+', text):
+ for match in re.finditer(r'https?://[^\s<>"\']+', text): # Find all URLs that are not already in HTML tags
start, end = match.span()
# Check if URL is inside an tag
@@ -378,16 +395,6 @@ class PostParser:
logger.error(f"webpage_parsing_error: url {getattr(webpage, 'url', 'unknown')}, error {str(e)}")
return None
- def _wrap_html(self, html_content: list) -> str:
- struct = [
- '', '',
- '', '', '',
- '', *html_content, '',
- ''
- ]
- html_content = '\n'.join(struct)
- return html_content
-
def _reactions_views_links(self, message: Message) -> Union[str, None]:
try:
parts = []
@@ -405,13 +412,11 @@ class PostParser:
channel_identifier = self.get_channel_username(message)
if channel_identifier:
links = []
- if channel_identifier.startswith('-100'):
- # For channels with only ID
+ if channel_identifier.startswith('-100'): # For channels with only ID
channel_id = channel_identifier[4:] # Remove '-100' prefix for web links
links.append(f'Open in Telegram')
links.append(f'Open in Web')
- else:
- # For channels with username
+ else: # For channels with username
links.append(f'Open in Telegram')
links.append(f'Open in Web')
parts.append(' '.join(links))
@@ -495,24 +500,15 @@ class PostParser:
if message.video and message.video.file_size > 100 * 1024 * 1024:
return
- if message.photo:
- file_data['file_unique_id'] = message.photo.file_unique_id
- elif message.video:
- file_data['file_unique_id'] = message.video.file_unique_id
- elif message.document:
- file_data['file_unique_id'] = message.document.file_unique_id
- elif message.audio:
- file_data['file_unique_id'] = message.audio.file_unique_id
- elif message.voice:
- file_data['file_unique_id'] = message.voice.file_unique_id
- elif message.video_note:
- file_data['file_unique_id'] = message.video_note.file_unique_id
- elif message.animation:
- file_data['file_unique_id'] = message.animation.file_unique_id
- elif message.sticker:
- file_data['file_unique_id'] = message.sticker.file_unique_id
- elif message.web_page and message.web_page.photo:
- file_data['file_unique_id'] = message.web_page.photo.file_unique_id
+ if message.photo: file_data['file_unique_id'] = message.photo.file_unique_id
+ elif message.video: file_data['file_unique_id'] = message.video.file_unique_id
+ elif message.document: file_data['file_unique_id'] = message.document.file_unique_id
+ elif message.audio: file_data['file_unique_id'] = message.audio.file_unique_id
+ elif message.voice: file_data['file_unique_id'] = message.voice.file_unique_id
+ elif message.video_note: file_data['file_unique_id'] = message.video_note.file_unique_id
+ elif message.animation: file_data['file_unique_id'] = message.animation.file_unique_id
+ elif message.sticker: file_data['file_unique_id'] = message.sticker.file_unique_id
+ elif message.web_page and message.web_page.photo: file_data['file_unique_id'] = message.web_page.photo.file_unique_id
if file_data['file_unique_id']:
file_data['channel'] = channel_username
diff --git a/requirements.txt b/requirements.txt
index ea93f4f..2e69374 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,4 +8,4 @@ feedgen==1.0.0
python-dateutil==2.9.0.post0
python-magic==0.4.27
beautifulsoup4
-bleach==6.1.0
\ No newline at end of file
+bleach[css]==6.1.0
diff --git a/rss_generator.py b/rss_generator.py
index fa0fee9..1dbe918 100644
--- a/rss_generator.py
+++ b/rss_generator.py
@@ -61,55 +61,18 @@ def reorganize_post_content(html):
return ''.join(new_content)
-
-def merge_posts(posts):
- """Helper function to merge multiple posts into one"""
- if not posts:
- logger.debug("No posts to merge")
- return None
-
- logger.debug(f"Merging {len(posts)} posts")
-
- # Find post with most meaningful title
- main_post = posts[0]
- for post in posts:
- current_title = post.get('title', '')
- if current_title and current_title not in ['📷 Photo', '📹 Video', '📄 Document']:
- main_post = post
- logger.debug(f"Selected main post with title: {current_title}")
- break
-
- merged_post = main_post.copy()
- merged_html = []
- for post in posts:
- merged_html.append(post['html'])
-
- merged_post['html'] = '\n
\n'.join(merged_html)
-
- # Reorganize content after merging
- merged_post['html'] = reorganize_post_content(merged_post['html'])
- logger.debug(f"Merged {len(posts)} posts successfully")
- return merged_post
-
-async def process_messages(messages, post_parser, channel):
+async def create_messages_groups(messages):
"""
Process messages into formatted posts, handling media groups
- Args:
- messages: List of raw messages
- post_parser: PostParser instance
- channel: Channel name for logging
- Returns:
- List of formatted and merged posts
"""
- posts = []
+ processing_groups = []
media_groups = {}
- # First pass - collect messages and media groups
+ # First pass - collect messages and organize into processing groups
for message in messages:
try:
# Skip service messages about pinned posts
if message.service and 'PINNED_MESSAGE' in str(message.service):
- logger.debug(f"Skipping pinned service message {message.id} in channel {channel}")
continue
if message.media_group_id:
@@ -117,41 +80,94 @@ async def process_messages(messages, post_parser, channel):
media_groups[message.media_group_id] = []
media_groups[message.media_group_id].append(message)
else:
- formatted = post_parser.format_message_for_feed(
- message,
- top_info=True,
- bottom_info=True
- )
- if formatted:
- posts.append(formatted)
+ # Single message becomes its own processing group
+ processing_groups.append([message])
except Exception as e:
- logger.error(f"feed_entry_error: channel {channel}, message_id {message.id}, error {str(e)}")
+ logger.error(f"message_processing_error: channel {message.chat.username}, message_id {message.id}, error {str(e)}")
continue
- # Process media groups
- for _group_id, group_messages in media_groups.items():
- if not group_messages:
- continue
-
- group_messages.sort(key=lambda x: x.id)
- formatted_posts = []
-
- for i, msg in enumerate(group_messages):
- formatted = post_parser.format_message_for_feed(
- msg,
- top_info=(i == 0),
- bottom_info=(i == len(group_messages) - 1)
- )
- if formatted:
- formatted_posts.append(formatted)
+ # Sort messages within media groups by message ID in descending order
+ for media_group in media_groups.values():
+ media_group.sort(key=lambda x: x.id, reverse=False)
+ processing_groups.append(media_group)
+
+ # Sort processing groups by date of first message in each group
+ processing_groups.sort(
+ key=lambda group: group[0].date,
+ reverse=True
+ )
+
+ return processing_groups
+
+async def render_messages_groups(messages_groups, post_parser):
+ """
+ Render message groups into HTML format
+ Args:
+ messages_groups: List of message groups (each group is a list of messages)
+ post_parser: PostParser instance
+ Returns:
+ List of rendered posts
+ """
+ rendered_posts = []
+
+ for group in messages_groups:
+ try:
+ if len(group) == 1:
+ # Single message - simple case
+ message_data = post_parser.process_message(group[0])
+ html_parts = [
+ f'',
+ f'{message_data["html"]["media"]}
',
+ f'{message_data["html"]["body"]}
',
+ f''
+ ]
+ rendered_posts.append({
+ 'html': '\n'.join(html_parts),
+ 'date': message_data['date'],
+ 'message_id': message_data['message_id'],
+ 'title': message_data['html']['title'],
+ 'text': message_data['text'],
+ 'author': message_data['author']
+ })
+ else:
+ # Multiple messages in group - need to merge
+ processed_messages = [post_parser.process_message(msg) for msg in group]
- if formatted_posts:
- posts.append(merge_posts(formatted_posts))
+ # Find main message (one with text)
+ main_message = next(
+ (msg for msg in processed_messages if msg['text']),
+ processed_messages[0] # fallback to first if none has text
+ )
+
+ # Collect all media sections
+ all_media = [msg['html']['media'] for msg in processed_messages if msg['html']['media']]
+ combined_media = '\n'.join(all_media)
+
+ # Combine parts
+ html_parts = [
+ f'',
+ f'{combined_media}
',
+ f'{main_message["html"]["body"]}
',
+ f''
+ ]
+
+ rendered_posts.append({
+ 'html': '\n'.join(html_parts),
+ 'date': main_message['date'],
+ 'message_id': main_message['message_id'],
+ 'title': main_message['html']['title'],
+ 'text': main_message['text'],
+ 'author': main_message['author']
+ })
+
+ except Exception as e:
+ logger.error(f"message_group_rendering_error: error {str(e)}")
+ continue
- # Sort all posts by date
- posts.sort(key=lambda x: x['date'], reverse=True)
- return posts
+ # Sort by date
+ rendered_posts.sort(key=lambda x: x['date'], reverse=True)
+ return rendered_posts
async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] = None, client = None, limit: int = 20) -> str:
"""
@@ -179,12 +195,8 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
base_url = Config['pyrogram_bridge_url']
try:
- # Преобразуем числовой ID канала обратно в int, если это возможно
- channel_id = channel
- if isinstance(channel, str) and channel.startswith('-100'):
- channel_id = int(channel)
-
- channel_info = await post_parser.client.get_chat(channel_id)
+ channel = post_parser.channel_name_prepare(channel)
+ channel_info = await post_parser.client.get_chat(channel)
channel_title = channel_info.title or f"Telegram: {channel}"
channel_username = post_parser.get_channel_username(channel_info)
if not channel_username:
@@ -205,30 +217,29 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
# Collect messages
messages = []
- async for message in post_parser.client.get_chat_history(channel_id, limit=limit):
+ async for message in post_parser.client.get_chat_history(channel, limit=limit):
messages.append(message)
- # Process messages into formatted posts
- final_posts = await process_messages(messages, post_parser, channel)
+ # Process messages into groups and render them
+ message_groups = await create_messages_groups(messages)
+ final_posts = await render_messages_groups(message_groups, post_parser)
# Generate feed entries
for post in final_posts:
fe = fg.add_entry()
- fe.title(post.get('title', 'Untitled post'))
+ fe.title(post['title'])
post_link = f"https://t.me/{channel_username}/{post['message_id']}"
fe.link(href=post_link)
- html_content = post.get('html', '')
- text_content = post.get('text', '')
- fe.description(text_content.replace('\n', ' '))
- fe.content(content=html_content, type='CDATA')
+ fe.description(post['text'].replace('\n', ' '))
+ fe.content(content=post['html'], type='CDATA')
pub_date = datetime.fromtimestamp(post['date'], tz=timezone.utc)
fe.pubDate(pub_date)
fe.guid(post_link, permalink=True)
- if post.get('author') and post['author'] != main_name:
+ if post['author'] and post['author'] != main_name:
fe.author(name="", email=post['author'])
rss_feed = fg.rss_str(pretty=True)
@@ -264,7 +275,6 @@ async def generate_channel_html(channel: str, post_parser: Optional[PostParser]
base_url = Config['pyrogram_bridge_url']
try:
- # Преобразуем числовой ID канала обратно в int, если это возможно
channel_id = channel
if isinstance(channel, str) and channel.startswith('-100'):
channel_id = int(channel)
@@ -289,18 +299,16 @@ async def generate_channel_html(channel: str, post_parser: Optional[PostParser]
async for message in post_parser.client.get_chat_history(channel_id, limit=limit):
messages.append(message)
- # Process messages into formatted posts
- final_posts = await process_messages(messages, post_parser, channel)
+ # Process messages into groups and render them
+ message_groups = await create_messages_groups(messages)
+ final_posts = await render_messages_groups(message_groups, post_parser)
- html_posts = []
- for post in final_posts:
- html_content = post.get('html', '')
- if html_content:
- html_posts.append(f'{html_content}
')
+ # Generate HTML content
+ html_posts = [post['html'] for post in final_posts]
return '\n
\n'.join(html_posts)
except Exception as e:
- logger.error(f"rss_generation_error: channel {channel}, error {str(e)}")
+ logger.error(f"html_generation_error: channel {channel}, error {str(e)}")
raise
def create_error_feed(channel: str, base_url: str) -> str: