import logging
import copy
import re
from datetime import datetime
from typing import Union, Dict, Any, List, Optional
from pyrogram.types import Message
from pyrogram.enums import MessageMediaType
from config import get_settings
Config = get_settings()
logger = logging.getLogger(__name__)
#tests
#http://127.0.0.1:8000/post/html/DragorWW_space/114 ā video
#http://127.0.0.1:8000/post/html/DragorWW_space/20 - many photos
#http://127.0.0.1:8000/post/html/DragorWW_space/58 - photos+video
#http://127.0.0.1:8000/post/html/DragorWW_space/44 - poll
#http://127.0.0.1:8000/post/html/DragorWW_space/46 - photo
#http://127.0.0.1:8000/post/html/DragorWW_space/49, http://127.0.0.1:8000/post/html/DragorWW_space/63 ā webpage
#http://127.0.0.1:8000/post/html/deckru/826 - animation
#http://127.0.0.1:8000/post/html/DragorWW_space/61 ā links
#http://127.0.0.1:8000/post/html/theyforcedme/3577 - video note
#http://127.0.0.1:8000/post/html/theyforcedme/3572 - audio
#http://127.0.0.1:8000/post/html/theyforcedme/3558 audio-note
#http://127.0.0.1:8000/html/vvzvlad_lytdybr/426 - sticker
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:
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"
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:
text = message.text.html if message.text else message.caption.html if message.caption else ''
text = text.replace('\n', '
')
html_content = []
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:
file_id = self._get_file_id(message)
if file_id:
html_content.append(f'')
if webpage := getattr(message, "web_page", None): # Web page preview
if webpage_html := self._format_webpage(webpage):
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 _format_webpage(self, webpage) -> Union[str, None]:
base_url = Config['pyrogram_bridge_url']
try:
if photo := getattr(webpage, "photo", None):
if file_id := getattr(photo, "file_id", None):
return (
f''
)
return None
except Exception as e:
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 _format_reactions_and_views(self, message: Message) -> Union[str, None]:
try:
html_parts = []
if reactions := getattr(message, "reactions", None):
reactions_html = ''
for reaction in reactions.reactions:
reactions_html += f'{reaction.emoji} {reaction.count} '
html_parts.append(reactions_html)
if views := getattr(message, "views", None):
views_html = f'(Views: {views})'
html_parts.append(views_html)
return ' '.join(html_parts) if html_parts else None
except Exception as e:
logger.error(f"reactions_views_parsing_error: {str(e)}")
return None
def _format_poll(self, poll) -> str:
try:
poll_text = f"š Poll: {poll.question}\n"
if hasattr(poll, "options") and poll.options:
for i, option in enumerate(poll.options, 1):
poll_text += f"{i}. {getattr(option, 'text', '')}\n"
poll_text += "\nā Vote in Telegram š\n"
return f'{poll_text.replace(chr(10), "
")}
'
except Exception as e:
logger.error(f"poll_parsing_error: {str(e)}")
return '[Error displaying poll]
'
def _get_file_id(self, message: Message) -> Union[str, None]:
try:
media_mapping = {
MessageMediaType.PHOTO: lambda m: m.photo.file_id,
MessageMediaType.VIDEO: lambda m: m.video.file_id,
MessageMediaType.DOCUMENT: lambda m: m.document.file_id,
MessageMediaType.AUDIO: lambda m: m.audio.file_id,
MessageMediaType.VOICE: lambda m: m.voice.file_id,
MessageMediaType.VIDEO_NOTE: lambda m: m.video_note.file_id,
MessageMediaType.ANIMATION: lambda m: m.animation.file_id,
MessageMediaType.STICKER: lambda m: m.sticker.file_id
}
if message.media in media_mapping:
return media_mapping[message.media](message)
return None
except Exception as e:
logger.error(f"file_id_extraction_error: media_type {message.media}, error {str(e)}")
return None
async def get_recent_posts(self, channel: str, limit: int = 20) -> List[Dict[Any, Any]]:
"""
Get recent posts from channel
"""
try:
messages = []
async for message in self.client.get_chat_history(channel, limit=limit):
try:
post = await self.get_post(channel, message.id, output_type='json')
if post:
messages.append(post)
except Exception as e:
logger.error(f"message_processing_error: channel {channel}, message_id {message.id}, error {str(e)}")
continue
return messages
except Exception as e:
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)
')
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'