diff --git a/.gitignore b/.gitignore index 49f51ac..4705c9a 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ .env .DS_Store +media_file_ids.json diff --git a/post_parser.py b/post_parser.py index db5fde7..ce45110 100644 --- a/post_parser.py +++ b/post_parser.py @@ -6,6 +6,8 @@ from typing import Union, Dict, Any, List, Optional from pyrogram.types import Message from pyrogram.enums import MessageMediaType from config import get_settings +import os +import json Config = get_settings() @@ -136,6 +138,9 @@ class PostParser: 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) @@ -143,10 +148,11 @@ class PostParser: base_url = Config['pyrogram_bridge_url'] if message.media: file_id = self._get_file_id(message) + print(file_id) if file_id: html_content.append(f'
') if message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]: - html_content.append(f'Media content') + html_content.append(f'') elif message.media == MessageMediaType.VIDEO: html_content.append(f'') elif message.media == MessageMediaType.ANIMATION: @@ -215,7 +221,7 @@ class PostParser: base_url = Config['pyrogram_bridge_url'] try: if photo := getattr(webpage, "photo", None): - if file_id := getattr(photo, "file_id", None): + if file_id := getattr(photo, "file_unique_id", None): return ( f'
' f'' @@ -272,14 +278,14 @@ class PostParser: 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 + MessageMediaType.PHOTO: lambda m: m.photo.file_unique_id, + MessageMediaType.VIDEO: lambda m: m.video.file_unique_id, + MessageMediaType.DOCUMENT: lambda m: m.document.file_unique_id, + MessageMediaType.AUDIO: lambda m: m.audio.file_unique_id, + MessageMediaType.VOICE: lambda m: m.voice.file_unique_id, + MessageMediaType.VIDEO_NOTE: lambda m: m.video_note.file_unique_id, + MessageMediaType.ANIMATION: lambda m: m.animation.file_unique_id, + MessageMediaType.STICKER: lambda m: m.sticker.file_unique_id, } if message.media in media_mapping: @@ -314,3 +320,51 @@ class PostParser: 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 = { + 'file_id': None, + } + + if message.media: + if message.photo: file_data['file_id'] = message.photo.file_unique_id + elif message.video: file_data['file_id'] = message.video.file_unique_id + elif message.document: file_data['file_id'] = message.document.file_unique_id + elif message.audio: file_data['file_id'] = message.audio.file_unique_id + elif message.voice: file_data['file_id'] = message.voice.file_unique_id + elif message.video_note: file_data['file_id'] = message.video_note.file_unique_id + elif message.animation: file_data['file_id'] = message.animation.file_unique_id + elif message.sticker: file_data['file_id'] = message.sticker.file_unique_id + elif message.web_page and message.web_page.photo: + file_data['file_id'] = message.web_page.photo.file_unique_id + + if file_data['file_id']: + file_path = 'media_file_ids.json' + try: + existing_data = [] + if os.path.exists(file_path): + with open(file_path, 'r', encoding='utf-8') as f: + existing_data = json.load(f) + + # Check if file_id already exists + found = False + for item in existing_data: + if item['file_id'] == file_data['file_id']: + item['added'] = datetime.now().timestamp() + found = True + break + + # Add new entry if not found + if not found: + file_data['added'] = datetime.now().timestamp() + existing_data.append(file_data) + + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(existing_data, f, ensure_ascii=False, indent=2) + + except Exception as e: + logger.error(f"file_id_save_error: error writing to {file_path}: {str(e)}") + + except Exception as e: + logger.error(f"file_id_collection_error: message_id {message.id}, error {str(e)}") diff --git a/telegram_client.py b/telegram_client.py deleted file mode 100644 index 61abddc..0000000 --- a/telegram_client.py +++ /dev/null @@ -1,594 +0,0 @@ -import logging -from pyrogram import Client, errors -from pyrogram.types import Message, Document, Audio, Video, VideoNote, Photo, Voice, Animation, WebPage -from config import get_settings -import os -import re - - -logger = logging.getLogger(__name__) -settings = get_settings() - -class TelegramClient: - def __init__(self): - if settings["session_string"]: - self.client = Client( - name="pyro_bridge", - session_string=settings["session_string"], - api_id=settings["tg_api_id"], - api_hash=settings["tg_api_hash"], - in_memory=True - ) - else: - self.client = Client( - name="pyro_bridge", - api_id=settings["tg_api_id"], - api_hash=settings["tg_api_hash"], - workdir=settings["session_path"], - in_memory=True - ) - - async def start(self): - if not self.client.is_connected: - await self.client.start() - logger.info("Telegram client connected") - - async def stop(self): - if self.client.is_connected: - await self.client.stop() - logger.info("Telegram client disconnected") - - async def _parse_message(self, message: Message) -> dict: - # Skip service messages about pinned posts - if getattr(message, 'service', None) == 'MessageServiceType.PINNED_MESSAGE': - logger.debug(f"Skipping pinned service message: {message.id}") - return None - - # Debug raw message structure - print(f"Raw message object: \n{message}") - - # Extract text from different sources - raw_text = message.text.html if message.text else "" - - # Handle polls - poll = getattr(message, "poll", None) - if poll: - try: - raw_text += f"\n📊 Poll: {poll.question}\n" - if hasattr(poll, "options") and poll.options: - for i, option in enumerate(poll.options, 1): - raw_text += f"{i}. {getattr(option, 'text', '')}\n" - raw_text += "\n→ Vote in Telegram 🔗\n" - except Exception as e: - logger.error(f"Error parsing poll: {str(e)}") - raw_text += "\n[Error displaying poll]" - - # Try to get caption for media messages - if not raw_text and hasattr(message, "caption") and message.caption: - raw_text = message.caption.html - - # Try to get text from other attributes - if not raw_text and hasattr(message, "action"): - raw_text = str(message.action) - elif not raw_text and hasattr(message, "entities"): - # Extract HTML from entities - raw_text = message.text.html if message.text else "" - - # Replace only URLs not already in tags - processed_text = re.sub( - r'(?)', - r'\1', - raw_text - ) - - # Parse media and add video markers - media = [] - if message.media: - try: - # Изменен порядок проверки: video_note перед video - media_obj = None - if message.video_note: - media_obj = message.video_note - elif message.video: - media_obj = message.video - elif message.photo: - media_obj = message.photo - elif message.document: - media_obj = message.document - elif message.audio: - media_obj = message.audio - elif message.voice: - media_obj = message.voice - elif message.animation: - media_obj = message.animation - elif message.web_page: - media_obj = message.web_page - - if media_obj: - parsed_media = await self._parse_media(media_obj) - if parsed_media.get("url"): - media.append(parsed_media) - logger.info(f"Processed media: {parsed_media}") - except Exception as e: - logger.error(f"Media processing error: {str(e)}") - media.append({"error": str(e)}) - - # Always add reactions if present - reactions_text = "" - if getattr(message, "reactions", None): - reactions_text = "\nReactions: " + ", ".join( - f"{r.emoji}({r.count})" - for r in message.reactions.reactions - ) - processed_text += reactions_text - raw_text += reactions_text - - # Get base URL from config - base_url = settings["pyrogram_bridge_url"].rstrip('/') - - # Add media previews to text - if media: - previews = [] - for m in media: - media_type = m.get('type', 'media') - file_id = m.get('url') - - if media_type in ['video', 'animation', 'video_note'] and file_id: - # Direct video/animation embedding - previews.append( - f"
" - f"" - f"
" - ) - elif media_type == 'audio' and file_id: - # Audio player - previews.append( - f"
" - f"" - f"
" - ) - elif media_type == 'voice' and file_id: - previews.append( - f"
" - f"" - f"
" - ) - elif media_type == 'animation' and file_id: - previews.append( - f"
" - f"" - f"
" - ) - elif media_type == 'web_page' and m.get('photo_url'): - previews.append( - f"
" - f"" - f"" - f"
" - ) - elif m.get('url') and media_type == 'photo': - # Image preview without type overlay - previews.append( - f"
" - f"" - f"
" - ) - - if previews: - previews_container = f"
{''.join(previews)}
" - processed_text = previews_container + processed_text - - # Convert newlines to HTML breaks AFTER adding reactions - processed_text = processed_text.replace('\n', '
') - - - return { - "text": re.sub('<[^<]+?>', '', raw_text).strip(), - "views": message.views or 0, - "media_group_id": message.media_group_id, - "id": message.id, - "date": message.date, - "html": processed_text, - "raw_text": raw_text, - "media": media, - "author": self._get_author_info(message) - } - - def _get_best_thumb(self, thumbs: list) -> str: - """Select highest resolution thumbnail from available thumbs""" - if not thumbs: - return "" - return max(thumbs, key=lambda x: x.width * x.height).file_id - - async def _parse_media(self, media_obj) -> dict: - try: - # Для VideoNote используем явное определение типа - file_id = "" # Инициализация по умолчанию - media_type = "unknown" - - if isinstance(media_obj, VideoNote): - file_id = media_obj.file_id - media_type = "video_note" - elif isinstance(media_obj, Animation): - file_id = media_obj.file_id - media_type = "animation" - elif isinstance(media_obj, Voice): - file_id = media_obj.file_id - media_type = "voice" - elif isinstance(media_obj, WebPage): - media_type = "web_page" - if media_obj.photo: - file_id = media_obj.photo.file_id - logger.debug(f"WebPage with photo: {file_id}") - else: - logger.warning("WebPage without photo") - else: - file_id = getattr(media_obj, "file_id", "") - media_type = media_obj.__class__.__name__.lower() - logger.debug(f"Generic media type: {media_type}") - - result = { - "type": media_type, - "url": file_id, - "size": getattr(media_obj, "file_size", 0), - "mime_type": getattr(media_obj, "mime_type", ""), - } - - if not result["url"]: - logger.error(f"Empty file_id for {media_type}: {media_obj}") - return {} - - # Логирование для отладки - logger.debug(f"Parsed media: type={media_type}, file_id={file_id}") - - # Обработка превью - if hasattr(media_obj, "thumbs"): - if result["type"] == "photo": - result["url"] = media_obj.file_id # Используем оригинальный file_id для фото - else: - result["thumbnail_file_id"] = self._get_best_thumb(media_obj.thumbs) - - # Дополнительные атрибуты - for attr in ["duration", "width", "height", "file_name", "title", "performer"]: - if hasattr(media_obj, attr): - result[attr] = getattr(media_obj, attr) - - if isinstance(media_obj, Voice): - return { - "type": "voice", - "url": media_obj.file_id, - "duration": media_obj.duration, - "mime_type": media_obj.mime_type, - "size": media_obj.file_size - } - - if isinstance(media_obj, WebPage): - return { - "type": "web_page", - "url": media_obj.url, - "title": getattr(media_obj, "title", ""), - "description": getattr(media_obj, "description", ""), - "photo_url": f"{settings['pyrogram_bridge_url'].rstrip('/')}/media/{media_obj.photo.file_id.lstrip('/')}" if media_obj.photo else None - } - - return result - except Exception as e: - logger.error(f"Media parse failed: {type(e).__name__} - {str(e)}") - return {} - - async def _parse_animation(self, animation_obj) -> dict: - logger.debug(f"Parsing animation: {animation_obj}") - if not animation_obj or not hasattr(animation_obj, 'thumbs'): - logger.error("Invalid animation object") - return {"type": "animation", "error": "invalid_object"} - - thumbnail_file_id = None - if animation_obj.thumbs and len(animation_obj.thumbs) > 0: - thumbnail_file_id = animation_obj.thumbs[0].file_id - - return { - "type": "animation", - "url": getattr(animation_obj, "file_id", ""), - "size": animation_obj.file_size, - "duration": animation_obj.duration, - "thumbnail_file_id": thumbnail_file_id - } - - def _generate_title(self, raw_text: str) -> str: - if not raw_text: - return "" - # Get first non-empty line - first_line = next((line.strip() for line in raw_text.split('\n') if line.strip()), "") - # Trim and remove HTML tags - clean_line = re.sub('<[^<]+?>', '', first_line) - - # Check if message contains only reactions - if re.match(r'^Reactions: ', clean_line): - return "📷 Media post" - - max_length = 65 - if len(clean_line) <= max_length: - return clean_line.strip() - - # Cut to last space - trimmed = clean_line[:max_length] - last_space = trimmed.rfind(' ') - if last_space > 0: - trimmed = trimmed[:last_space] - - return f"{trimmed.strip()}..." if trimmed else "" - - async def get_post(self, channel: str, post_id: int) -> dict: - try: - message = await self.client.get_messages( - chat_id=channel, - message_ids=post_id - ) - if not message or message.empty: - return { - "error": "message_not_found", - "details": f"Message {post_id} in {channel} does not exist" - } - - # Check for media group - media_group = [] - if message.media_group_id: - logger.debug(f"Processing media group {message.media_group_id} for message {post_id}") - current_id = post_id - # Collect previous messages in group - while True: - prev_msg = await self.client.get_messages( - chat_id=channel, - message_ids=current_id - 1 - ) - if prev_msg and prev_msg.media_group_id == message.media_group_id: - logger.debug(f"Found previous message in group: {prev_msg.id}") - media_group.insert(0, prev_msg) - current_id -= 1 - else: - break - - # Collect next messages in group - current_id = post_id - while True: - next_msg = await self.client.get_messages( - chat_id=channel, - message_ids=current_id + 1 - ) - if next_msg and next_msg.media_group_id == message.media_group_id: - logger.debug(f"Found next message in group: {next_msg.id}") - media_group.append(next_msg) - current_id += 1 - else: - break - - # Process all messages in group - all_messages = media_group + [message] - logger.debug(f"Combined {len(all_messages)} messages in media group") - parsed_messages = [await self._parse_message(msg) for msg in all_messages] - - # Combine results - combined = { - "id": post_id, - "date": message.date.isoformat(), - "title": self._generate_title(message.text or message.caption or ""), - "raw": self._parse_raw_message(message), - "html": "".join([m["html"] for m in parsed_messages if m["html"]]), - "text": "".join([m["text"] for m in parsed_messages if m["text"]]), - "views": max(m["views"] for m in parsed_messages), - "author": parsed_messages[0]["author"] if parsed_messages else "" - } - logger.debug(f"Combined result: {combined}") - - return combined - except errors.RPCError as e: - logger.error(f"Telegram API error: {e.__class__.__name__} - {e}") - raise - except Exception as e: - logger.error(f"Unexpected error in get_post: {str(e)}") - raise - - def _parse_raw_message(self, message: Message) -> dict: - """Convert Pyrogram Message object to serializable dict""" - return { - "id": message.id, - "date": message.date.isoformat(), - "media_type": str(message.media).split('.')[-1].lower() if message.media else None, - "text": message.text or message.caption, - "author": self._get_author_info(message), - "views": message.views, - "reactions": self._parse_reactions(message), - "media": self._parse_media_metadata(message), - "sender_chat": self._parse_sender_chat(message), - "edit_date": message.edit_date.isoformat() if message.edit_date else None, - "forward_from": self._parse_forward_info(message), - "media_group_id": message.media_group_id, - "has_protected_content": message.has_protected_content - } - - def _parse_sender_chat(self, message: Message) -> dict: - if not message.sender_chat: - return None - return { - "id": message.sender_chat.id, - "title": getattr(message.sender_chat, "title", ""), - "username": getattr(message.sender_chat, "username", ""), - "photo": self._parse_chat_photo(message.sender_chat.photo) - } - - def _parse_forward_info(self, message: Message) -> dict: - if not message.forward_from_chat: - return None - return { - "id": message.forward_from_chat.id, - "title": getattr(message.forward_from_chat, "title", ""), - "username": getattr(message.forward_from_chat, "username", "") - } - - def _parse_chat_photo(self, photo) -> dict: - if not photo: - return None - return { - "small_file_id": photo.small_file_id, - "big_file_id": photo.big_file_id - } - - def _get_author_info(self, message: Message) -> dict: - # Extract author information - author_str = "" - if message.sender_chat: - title = getattr(message.sender_chat, "title", "").strip() - username = getattr(message.sender_chat, "username", "").strip() - author_str = 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])) - author_str = f"{name} by @{username}" if username else name - else: - author_str = "Unknown author" - return author_str - - - def _parse_reactions(self, message: Message) -> list: - if not message.reactions: - return [] - return [ - { - "emoji": r.emoji, - "count": r.count - } for r in message.reactions.reactions - ] - - def _parse_media_metadata(self, message: Message) -> dict: - if not message.media: - return {} - return { - "type": str(message.media).split('.')[-1].lower(), - "file_id": getattr(message.media, "file_id", None), - "file_size": getattr(message.media, "file_size", None), - "duration": getattr(message.media, "duration", None), - "photo": self._parse_photo_metadata(message.media) if hasattr(message.media, "photo") else None, - "thumbs": self._parse_thumbs(message.media) - } - - def _parse_photo_metadata(self, media_obj) -> dict: - return { - "width": media_obj.width, - "height": media_obj.height, - "file_size": media_obj.file_size, - "date": media_obj.date.isoformat() if media_obj.date else None - } - - def _parse_thumbs(self, media_obj) -> list: - if not hasattr(media_obj, "thumbs") or not media_obj.thumbs: - return [] - return [ - { - "file_id": thumb.file_id, - "width": thumb.width, - "height": thumb.height, - "file_size": thumb.file_size - } for thumb in media_obj.thumbs - ] - - async def download_media_file(self, file_id: str) -> str: - """Download media file and return local path""" - try: - if not self.client.is_connected: - await self.start() - - # Validate file_id format before downloading - if len(file_id) < 20 or not re.match(r'^[a-zA-Z0-9_-]+$', file_id): - logger.error(f"Invalid file_id format: {file_id}") - raise ValueError("Invalid file identifier format") - - return await self.client.download_media( - file_id, - in_memory=False, - block=True - ) - except errors.RPCError as e: - logger.error(f"Telegram media download error {file_id}: {type(e).__name__} - {str(e)}") - raise - except ValueError as e: - logger.error(f"Invalid file_id {file_id}: {str(e)}") - raise - except Exception as e: - logger.error(f"Media download failed {file_id}: {type(e).__name__} - {str(e)}") - raise - - async def get_channel_posts(self, channel: str, limit: int = 20) -> list: - try: - if not self.client.is_connected: - await self.start() - - posts = [] - channel_title = "" - channel_icon = "" - async for message in self.client.get_chat_history( - chat_id=channel, - limit=limit - ): - if not channel_title and message.sender_chat: - channel_title = getattr(message.sender_chat, "title", channel) - if hasattr(message.sender_chat, "photo") and message.sender_chat.photo: - channel_icon = message.sender_chat.photo.small_file_id - - parsed = await self._parse_message(message) - if parsed and parsed.get('html'): - posts.append({ - "id": message.id, - "date": message.date, - "channel_title": channel_title, - "channel_icon": channel_icon, - "title": self._generate_title(parsed.get("raw_text", "")), - "html": parsed["html"], - "media": parsed.get("media", []) - }) - return posts - - except Exception as e: - logger.error(f"Channel posts error {channel}: {type(e).__name__} - {str(e)}") - return [] - - async def _process_media(self, media, media_type) -> dict: - try: - result = { - "type": str(media_type).split('.')[-1].lower(), - "file_id": getattr(media, "file_id", ""), - "file_unique_id": getattr(media, "file_unique_id", ""), - "file_size": getattr(media, "file_size", 0), - "mime_type": getattr(media, "mime_type", None), - } - - # Handle thumbnails for media types that support them - if hasattr(media, "thumbs"): - thumbs = getattr(media, "thumbs", []) - if thumbs: - result["thumb"] = self._get_best_thumb(thumbs) - - # Add type-specific fields using safe attribute access - if isinstance(media, (Document, Audio, Video, VideoNote)): - result.update({ - "file_name": getattr(media, "file_name", None), - "duration": getattr(media, "duration", None), - "width": getattr(media, "width", None), - "height": getattr(media, "height", None), - "title": getattr(media, "title", None), - "performer": getattr(media, "performer", None), - }) - - return result - except Exception as e: - logger.error(f"Media processing failed {type(media).__name__}: {type(e).__name__} - {str(e)}") - return {} \ No newline at end of file