#http://127.0.0.1:8000/html/ru2ch_ban/26586 - large video
class PostParser:
def __init__(self, client):
self.client = client
def _debug_message(self, message: Message) -> Message:
if Config["debug"]:
debug_message = copy.deepcopy(message)
debug_message.sender_chat = None
debug_message.caption_entities = None
debug_message.reactions = None
debug_message.entities = None
print(debug_message)
return
async def get_post(self, channel: str, post_id: int, output_type: str = 'json') -> Union[str, Dict[Any, Any]]:
try:
message = await self.client.get_messages(channel, post_id)
self._debug_message(message)
if not message:
logger.error(f"post_not_found: channel {channel}, post_id {post_id}")
return None
if output_type == 'html':
return self._format_html(message)
elif output_type == 'json':
return self._format_json(message)
else:
logger.error(f"Invalid output type: {output_type}")
return None
except Exception as e:
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': message.chat.username,
'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', '').strip()
username = getattr(message.sender_chat, 'username', '').strip()
return f"{title} by @{username}" if username else title
elif message.from_user:
first = getattr(message.from_user, 'first_name', '').strip()
last = getattr(message.from_user, 'last_name', '').strip()
username = getattr(message.from_user, 'username', '').strip()
name = ' '.join(filter(None, [first, last]))
return f"{name} by @{username}" if username else name
return "Unknown author"
def _generate_title(self, message: Message) -> str:
# Check for "channel created" service message
if getattr(message, "channel_chat_created", False):
return "✨ Channel created"
text = message.text or message.caption or ''
if not text:
if message.media == MessageMediaType.PHOTO: return "📷 Photo"
elif message.media == MessageMediaType.VIDEO: return "🎥 Video"
elif message.media == MessageMediaType.ANIMATION: return "🎞 GIF"
elif message.media == MessageMediaType.AUDIO: return "🎵 Audio"
elif message.media == MessageMediaType.VOICE: return "🎤 Voice message"
elif message.media == MessageMediaType.VIDEO_NOTE: return "🎥 Video message"
elif message.media == MessageMediaType.STICKER: return "🎯 Sticker"
elif message.media == MessageMediaType.POLL: return "📊 Poll"
return "📷 Media post"
# Remove URLs
text = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '', text)
# Remove HTML tags
text = re.sub('<[^<]+?>', '', text)
# Remove multiple spaces and empty lines
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 ""
max_length = 100
if len(first_line) <= max_length:
return first_line.strip()
# Cut to last space
trimmed = first_line[:max_length]
last_space = trimmed.rfind(' ')
if last_space > 0:
trimmed = trimmed[:last_space]
return f"{trimmed.strip()}..." if trimmed else ""
def _format_html(self, message: Message, naked: bool = False) -> str:
# Check for "channel created" service message
if getattr(message, "channel_chat_created", False):
service_html = 'Channel created
'
if not naked:
return self._wrap_html([service_html])
return service_html
text = message.text.html if message.text else message.caption.html if message.caption else ''
text = text.replace('\n', '
')
# Add hyperlinks to raw URLs
text = self._add_hyperlinks_to_raw_urls(text)
html_content = []
# Save media file_ids
self._save_media_file_ids(message)
if poll := getattr(message, "poll", None): # Poll formatting
if poll_html := self._format_poll(poll):
html_content.append(poll_html)
base_url = Config['pyrogram_bridge_url']
if message.media and message.media != "MessageMediaType.POLL":
file_unique_id = self._get_file_unique_id(message)
if file_unique_id is None:
logger.debug(f"File unique id not found for message {message.id}")
elif file_unique_id:
url = f"{base_url}/media/{message.chat.username}/{message.id}/{file_unique_id}"
logger.debug(f"Collected media file: {message.chat.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)
if text: # Message text
html_content.append(f'{text}
')
if not naked:
if reactions_views_html := self._format_reactions_and_views(message): # Add reactions and views
html_content.append(reactions_views_html)
if not naked:
html = self._wrap_html(html_content)
else:
html = '\n'.join(html_content)
return html
def _add_hyperlinks_to_raw_urls(self, text: str) -> str:
try:
# Find all existing 
')
elif message.media == MessageMediaType.VIDEO:
html_content.append(f'')
elif message.media == MessageMediaType.ANIMATION:
html_content.append(f'')
elif message.media == MessageMediaType.VIDEO_NOTE:
html_content.append(f'')
elif message.media == MessageMediaType.AUDIO:
mime_type = getattr(message.audio, 'mime_type', 'audio/mpeg')
html_content.append(f'')
elif message.media == MessageMediaType.VOICE:
mime_type = getattr(message.voice, 'mime_type', 'audio/ogg')
html_content.append(f'')
elif message.media == MessageMediaType.STICKER:
emoji = getattr(message.sticker, 'emoji', '')
html_content.append(f'