vvzvlad/pyrogram-bridge
1
0
Fork 0
Code Issues 6 Pull Requests Actions 8 Packages Projects Releases Wiki Activity

Add HTML output option for RSS feed and enhance post processing

Browse Source
This commit is contained in:
vvzvlad
2025-02-05 23:06:15 +03:00
parent 7d0e8b334b
commit cebcdbd5df
3 changed files with 259 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_server.py
+8 -4
View File
@@ -15,7 +15,7 @@ from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import HTMLResponse, FileResponse
from telegram_client import TelegramClient
from config import get_settings
from rss_generator import generate_channel_rss
from rss_generator import generate_channel_rss, generate_channel_html
from post_parser import PostParser
from url_signer import verify_media_digest
@@ -419,7 +419,7 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str
@app.get("/rss/{channel}", response_class=Response)
@app.get("/rss/{channel}/{token}", response_class=Response)
async def get_rss_feed(channel: str, token: str | None = None, limit: int = 20):
async def get_rss_feed(channel: str, token: str | None = None, limit: int = 20, output_type: str = 'rss'):
if Config["token"]:
if token != Config["token"]:
logger.error(f"Invalid token for RSS feed: {token}, expected: {Config['token']}")
@@ -429,8 +429,12 @@ async def get_rss_feed(channel: str, token: str | None = None, limit: int = 20):
while True:
try:
rss_content = await generate_channel_rss(channel, client=client.client, limit=limit)
return Response(content=rss_content, media_type="application/xml")
if output_type == 'rss':
rss_content = await generate_channel_rss(channel, client=client.client, limit=limit)
return Response(content=rss_content, media_type="application/xml")
elif output_type == 'html':
rss_content = await generate_channel_html(channel, client=client.client, limit=limit)
return Response(content=rss_content, media_type="text/html")
except ValueError as e:
error_message = f"Invalid parameters for RSS feed generation: {str(e)}"
logger.error(error_message)
post_parser.py
+58 -47
View File
@@ -44,11 +44,13 @@ if not logger.handlers:
#http://127.0.0.1:8000/html/ufjqk/1070 - reply to
#http://127.0.0.1:8000/html/tetstststststststffd/4 - forwarded from channel without name
#http://127.0.0.1:8000/html/tetstststststststffd/14 - forwarded from hidden user
#https://t.me/smallpharm/4802
# https://t.me/webstrangler/3987
#https://t.me/smallpharm/4802 many pics and text
# https://t.me/webstrangler/3987 many pics without text
# https://t.me/teslacoilpro/7117
# https://t.me/red_spades/1222 many media + text
#https://t.me/ni404head/1283 file
class PostParser:
def __init__(self, client):
self.client = client
@@ -85,24 +87,6 @@ class PostParser:
logger.error(f"post_parsing_error: channel {channel}, post_id {post_id}, error {str(e)}")
raise
def _format_json(self, message: Message, naked: bool = False) -> Dict[Any, Any]:
html_content = self._format_html(message, naked=naked)
result = {
'channel': self.get_channel_username(message),
'message_id': message.id,
'date': datetime.timestamp(message.date),
'text': message.text or message.caption or '',
'html': html_content,
'title': self._generate_title(message),
'author': self._get_author_info(message),
'views': message.views,
}
if message.media_group_id:
result['media_group_id'] = message.media_group_id
return result
def _get_author_info(self, message: Message) -> str:
if message.sender_chat:
title = getattr(message.sender_chat, 'title', None)
@@ -194,31 +178,60 @@ class PostParser:
return f'<div class="message-reply">Reply to #{reply_to.id}: {reply_text}</div><br>'
return None
def _format_html(self, message: Message, naked: bool = False) -> str:
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)
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),
'author': self._get_author_info(message),
'views': message.views,
}
if message.media_group_id:
result['media_group_id'] = message.media_group_id
return result
def _format_html(self, message: Message, top_info: bool = True, bottom_info: bool = True) -> str:
html_content = []
# Add forwarded from or reply info if present
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)
# Check for "channel created" service message
# Create service message for "channel created"
if getattr(message, "channel_chat_created", False):
service_html = '<div class="message-service">Channel created</div>'
if not naked:
return self._wrap_html([service_html])
return service_html
html_content.append('<div class="message-service">Channel created</div>')
html = '\n'.join(html_content)
return html
text = message.text.html if message.text else message.caption.html if message.caption else ''
text = text.replace('\n', '<br>')
# 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 message.text:
text = message.text.html
elif message.caption:
text = message.caption.html
else:
text = ''
text = text.replace('\n', '<br>') # Replace newlines with <br>
text = self._add_hyperlinks_to_raw_urls(text) # Add hyperlinks to raw URLs
# Add hyperlinks to raw URLs
text = self._add_hyperlinks_to_raw_urls(text)
# Save media file_ids
self._save_media_file_ids(message)
self._save_media_file_ids(message) # Save media file_ids for caching
if poll := getattr(message, "poll", None): # Poll formatting
if poll_html := self._format_poll(poll):
@@ -264,14 +277,15 @@ class PostParser:
if text: # Message text
html_content.append(f'<div class="message-text">{text}</div>')
if not naked:
if bottom_info:
if reactions_views_html := self._reactions_views_links(message): # Add reactions, views and links
html_content.append(reactions_views_html)
if not naked:
html = self._wrap_html(html_content)
else:
html = '\n'.join(html_content)
#if not naked:
# html = self._wrap_html(html_content)
#else:
html = '\n'.join(html_content)
return html
@@ -421,9 +435,6 @@ class PostParser:
logger.error(f"recent_posts_error: channel {channel}, error {str(e)}")
raise
def format_message_for_feed(self, message: Message, naked: bool = False) -> Dict[Any, Any]:
return self._format_json(message, naked=naked)
def _save_media_file_ids(self, message: Message) -> None:
try:
file_data = {
rss_generator.py
+193 -56
View File
@@ -4,6 +4,7 @@ from typing import Optional
from feedgen.feed import FeedGenerator
from post_parser import PostParser
from config import get_settings
from bs4 import BeautifulSoup
Config = get_settings()
@@ -15,8 +16,134 @@ if not logger.handlers:
handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logger.addHandler(handler)
def reorganize_post_content(html):
"""
Reorganize post content to move text after all media
Args:
html: Post HTML content
Returns:
Reorganized HTML content
"""
soup = BeautifulSoup(html, 'html.parser')
# Find all media and text blocks
media_blocks = soup.find_all('div', class_='message-media')
text_blocks = soup.find_all('div', class_='message-text')
if not text_blocks: # No text to move
return html
# Create new structure
new_content = []
# Add all media first
for media in media_blocks:
media.extract()
new_content.append(str(media))
new_content.append('<br>')
# Add text last
for text in text_blocks:
text.extract()
new_content.append(str(text))
# Add remaining elements
for element in soup.contents:
if isinstance(element, str):
new_content.append(element)
elif element.name != 'br' and 'message-media' not in element.get('class', []) and 'message-text' not in element.get('class', []):
new_content.append(str(element))
return ''.join(new_content)
def merge_posts(posts):
"""Helper function to merge multiple posts into one"""
if not posts:
return None
# 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
break
merged_post = main_post.copy()
merged_html = []
for post in posts:
merged_html.append(post['html'])
merged_post['html'] = '\n<br>\n'.join(merged_html)
# Reorganize content after merging
merged_post['html'] = reorganize_post_content(merged_post['html'])
return merged_post
async def process_messages(messages, post_parser, channel):
"""
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 = []
media_groups = {}
# First pass - collect messages and media 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:
if message.media_group_id not in media_groups:
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)
except Exception as e:
logger.error(f"feed_entry_error: channel {channel}, 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)
if formatted_posts:
posts.append(merge_posts(formatted_posts))
# Sort all posts by date
posts.sort(key=lambda x: x['date'], reverse=True)
return posts
async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] = None, client = None, limit: int = 20) -> str:
"""
Generate RSS feed for channel using actual messages
@@ -25,6 +152,7 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
post_parser: Optional PostParser instance. If not provided, will create new one
client: Telegram client instance
limit: Maximum number of posts to include in the RSS feed
output_type: 'rss' or 'html'
Returns:
RSS feed as string in XML format
"""
@@ -47,7 +175,7 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
channel_username = post_parser.get_channel_username(channel_info)
if not channel_username:
return create_error_feed(channel, base_url)
except Exception as e: # raise error if channel not found
except Exception as e:
if "USERNAME_INVALID" in str(e) or "USERNAME_NOT_OCCUPIED" in str(e):
return create_error_feed(channel, base_url)
else:
@@ -59,66 +187,18 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
fg.link(href=f"https://t.me/{channel_username}", rel='alternate')
fg.description(f'Telegram channel {channel_username} RSS feed')
fg.language('ru')
fg.id(f"{base_url}/rss/{channel}")
# First collect all posts
posts = []
media_groups = {}
# Collect messages
messages = []
async for message in post_parser.client.get_chat_history(channel, limit=limit):
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
naked_html = True if message.media_group_id else False
post = post_parser.format_message_for_feed(message, naked=naked_html)
if not post:
continue
if post.get('media_group_id'):
if post['media_group_id'] not in media_groups:
media_groups[post['media_group_id']] = []
media_groups[post['media_group_id']].append(post)
else:
posts.append(post)
except Exception as e:
logger.error(f"feed_entry_error: channel {channel}, message_id {message.id}, error {str(e)}")
continue
# Merge media groups
for _group_id, group_posts in media_groups.items():
if not group_posts:
continue
# Sort posts within media group by message_id
group_posts.sort(key=lambda x: x['message_id'])
messages.append(message)
# Find post with most meaningful title
main_post = group_posts[0]
for post in group_posts:
current_title = post.get('title', '')
if current_title and current_title not in ['📷 Photo', '📹 Video', '📄 Document']:
main_post = post
break
merged_post = main_post.copy()
merged_html = []
for post in group_posts:
merged_html.append(post['html'])
merged_post['html'] = '\n'.join(merged_html)
posts.append(merged_post)
# Sort posts by date
posts.sort(key=lambda x: x['date'], reverse=True)
# Process messages into formatted posts
final_posts = await process_messages(messages, post_parser, channel)
# Generate feed entries
for post in posts:
for post in final_posts:
fe = fg.add_entry()
fe.title(post.get('title', 'Untitled post'))
@@ -144,9 +224,66 @@ async def generate_channel_rss(channel: str, post_parser: Optional[PostParser] =
except Exception as e:
logger.error(f"rss_generation_error: channel {channel}, error {str(e)}")
raise
raise
async def generate_channel_html(channel: str, post_parser: Optional[PostParser] = None, client = None, limit: int = 20) -> str:
"""
Generate HTML feed for channel using actual messages
Args:
channel: Telegram channel name
post_parser: Optional PostParser instance. If not provided, will create new one
client: Telegram client instance
limit: Maximum number of posts to include in the RSS feed
Returns:
HTML feed as string
"""
if limit < 1:
raise ValueError(f"limit must be positive, got {limit}")
if limit > 100:
raise ValueError(f"limit cannot exceed 100, got {limit}")
try:
if post_parser is None:
post_parser = PostParser(client=client)
base_url = Config['pyrogram_bridge_url']
try:
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:
return create_error_feed(channel, base_url)
except Exception as e:
if "USERNAME_INVALID" in str(e) or "USERNAME_NOT_OCCUPIED" in str(e):
return create_error_feed(channel, base_url)
else:
raise
# Set feed metadata
main_name = f"{channel_title} (@{channel_username})"
html_content = f'<h1>{main_name}</h1>'
# Collect messages
messages = []
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)
html_posts = []
for post in final_posts:
html_content = post.get('html', '')
if html_content:
html_posts.append(f'<div class="telegram-post">{html_content}</div>')
return '\n<hr class="post-divider">\n'.join(html_posts)
except Exception as e:
logger.error(f"rss_generation_error: channel {channel}, error {str(e)}")
raise
def create_error_feed(channel: str, base_url: str) -> str:
"""
Create an empty RSS feed with metadata indicating an error when the channel is not found.