From 65a69ed37a673a0b31ecea0f5a66e6abf5db6644 Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Sat, 19 Apr 2025 04:31:50 +0300 Subject: [PATCH] refactoring --- post_parser.py | 117 +++++++++++++++++++++++++++++-------------------- 1 file changed, 70 insertions(+), 47 deletions(-) diff --git a/post_parser.py b/post_parser.py index 68ff6b0..3829f2b 100644 --- a/post_parser.py +++ b/post_parser.py @@ -188,7 +188,8 @@ class PostParser: return f"πŸ“Š Poll: {poll_question}" return "πŸ“Š Poll" if message.media == MessageMediaType.DOCUMENT: - if message.document is not None and hasattr(message.document, 'mime_type') and message.document.mime_type == 'application/pdf': + is_mime = (message.document is not None and hasattr(message.document, 'mime_type')) + if is_mime and message.document is not None and message.document.mime_type == 'application/pdf': return "πŸ“„ PDF Document" return "πŸ“Ž Document" if message.media == MessageMediaType.PHOTO: return "πŸ“· Photo" @@ -279,13 +280,15 @@ class PostParser: def _format_reply_info(self, message: Message) -> Union[str, None]: - if getattr(message, "service", None) and 'PINNED_MESSAGE' in str(message.service) and (reply_to := getattr(message, "reply_to_message", None)): + is_pinned = (getattr(message, "service", None) and 'PINNED_MESSAGE' in str(message.service)) + reply_to = getattr(message, "reply_to_message", None) + if is_pinned and reply_to: reply_text = reply_to.text or reply_to.caption or '' if len(reply_text) > 100: reply_text = reply_text[:100] + '...' return f'
Pinned: {reply_text}

' - if reply_to := getattr(message, "reply_to_message", None): + if reply_to: reply_text = reply_to.text or reply_to.caption or '' if len(reply_text) > 100: reply_text = reply_text[:100] + '...' @@ -363,8 +366,11 @@ class PostParser: flags.append("clownpoo") break - # Check if the message text contains "#Ρ€Π΅ΠΊΠ»Π°ΠΌΠ°", "ΠŸΠ°Ρ€Ρ‚Π½Π΅Ρ€ΡΠΊΠΈΠΉ пост", "ΠΏΠΎ ΠΏΡ€ΠΎΠΌΠΎΠΊΠΎΠ΄Ρƒ", "скидка Π½Π° курс", "Ρ€Π΅Π³ΠΈΡΡ‚Ρ€ΠΈΡ€ΡƒΠΉΡ‚Π΅ΡΡŒ Ρ‚ΡƒΡ‚" in a case-insensitive manner. - if re.search(r'(?i)(#Ρ€Π΅ΠΊΠ»Π°ΠΌΠ°|#ΠΏΡ€ΠΎΠΌΠΎ|О\s+Ρ€Π΅ΠΊΠ»Π°ΠΌΠΎΠ΄Π°Ρ‚Π΅Π»Π΅|партнСрский\s+пост|ΠΏΠΎ\s+ΠΏΡ€ΠΎΠΌΠΎΠΊΠΎΠ΄Ρƒ|erid|скидка\s+Π½Π°\s+курс|Ρ€Π΅Π³ΠΈΡΡ‚Ρ€ΠΈΡ€ΡƒΠΉΡ‚Π΅ΡΡŒ\s+Ρ‚ΡƒΡ‚)', message_text_str): + # Check if the message text contains advert. + if re.search(r'(?i)(#Ρ€Π΅ΠΊΠ»Π°ΠΌΠ°|#ΠΏΡ€ΠΎΠΌΠΎ|О\s+Ρ€Π΅ΠΊΠ»Π°ΠΌΠΎΠ΄Π°Ρ‚Π΅Π»Π΅|партнСрский\s+пост)', message_text_str): + flags.append("advert") + + if re.search(r'(?i)(ΠΏΠΎ\s+ΠΏΡ€ΠΎΠΌΠΎΠΊΠΎΠ΄Ρƒ|erid|скидка\s+Π½Π°\s+курс|Ρ€Π΅Π³ΠΈΡΡ‚Ρ€ΠΈΡ€ΡƒΠΉΡ‚Π΅ΡΡŒ\s+Ρ‚ΡƒΡ‚)', message_text_str): flags.append("advert") # Check for paywall-related words and tags @@ -429,7 +435,9 @@ class PostParser: except Exception as e: logger.error(f"tme_link_extraction_error: message_id {message.id}, error {str(e)}") - return flags + # Deduplicate flags + deduplicated_flags = list(dict.fromkeys(flags)) + return deduplicated_flags def _format_html(self, data: Dict[str, Any], debug: bool = False) -> str: html_content = [] @@ -442,7 +450,9 @@ class PostParser: # Add raw JSON debug output if debug is enabled if debug: - html_content.append(f'
{data["raw_message"]}
') + html_content.append(f'
{data["raw_message"]}
') html_data = '\n'.join(html_content) return html_data @@ -497,7 +507,10 @@ class PostParser: def _sanitize_html(self, html_raw: str) -> str: - allowed_tags = ['p', 'a', 'b', 'i', 'strong', 'em', 'ul', 'ol', 'li', 'br', 'div', 'span', 'img', 'video', 'audio', 'source'] + 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', 'alt', 'style'], @@ -622,38 +635,52 @@ class PostParser: content_media.append(f'
') # Check if document is a PDF file - if message.media == MessageMediaType.DOCUMENT and message.document is not None and hasattr(message.document, 'mime_type') and message.document.mime_type == 'application/pdf': + if (message.media == MessageMediaType.DOCUMENT and + message.document is not None and hasattr(message.document, 'mime_type') and + message.document.mime_type == 'application/pdf'): # Only attempt to create link if channel_username is available if channel_username: - if channel_username.startswith('-100'): tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}" - else: tg_link = f"https://t.me/{channel_username}/{message.id}" - content_media.append(f'
[PDF-Ρ„Π°ΠΉΠ»]
') + if channel_username.startswith('-100'): + tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}" + else: + tg_link = f"https://t.me/{channel_username}/{message.id}" + content_media.append(f'
') + content_media.append(f'[PDF-Ρ„Π°ΠΉΠ»]
') else: # Handle case where channel_username is None (e.g., log or add placeholder) - logger.warning(f"Could not generate PDF link for message {message.id} because channel username is missing.") + logger.warning(f"Could not generate PDF link for {message.id}: ch username is missing.") content_media.append(f'
[PDF-Ρ„Π°ΠΉΠ»]
') # Add placeholder without link elif message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]: - content_media.append(f'') + content_media.append(f'') elif message.media == MessageMediaType.VIDEO: - content_media.append(f'') + content_media.append(f'') elif message.media == MessageMediaType.ANIMATION: - content_media.append(f'') + content_media.append(f'') elif message.media == MessageMediaType.VIDEO_NOTE: - content_media.append(f'') + content_media.append(f'') elif message.media == MessageMediaType.AUDIO: mime_type = getattr(message.audio, 'mime_type', 'audio/mpeg') - content_media.append(f'') + content_media.append(f'') content_media.append('
') elif message.media == MessageMediaType.VOICE: mime_type = getattr(message.voice, 'mime_type', 'audio/ogg') - content_media.append(f'') + content_media.append(f'') content_media.append('
') elif message.media == MessageMediaType.STICKER: emoji = getattr(message.sticker, 'emoji', '') if getattr(message.sticker, 'is_video', False): - content_media.append(f'') + content_media.append(f'') else: - content_media.append(f'Sticker {emoji}') + content_media.append(f'Sticker {emoji}') content_media.append('
') if webpage := getattr(message, "web_page", None): # Web page preview @@ -674,7 +701,9 @@ class PostParser: if is_telegram_message: # Telegram message preview with distinctive styling - html_parts = ['
'] + html_parts = [("""
""")] html_parts.append(f'
πŸ“± Telegram
') else: # Regular webpage preview @@ -686,23 +715,25 @@ class PostParser: # Add title with link if available if title := getattr(webpage, "title", None): url = getattr(webpage, "url", "#") - html_parts.append(f'
{title}
') - + html_parts.append(f'
') + html_parts.append(f'{title}
') # Add description if available if description := getattr(webpage, "description", None): # Process the description to handle line breaks and possibly HTML processed_description = description.replace('\n', '
') processed_description = self._add_hyperlinks_to_raw_urls(processed_description) - html_parts.append(f'
{processed_description}
') + html_parts.append(f'
{processed_description}
') # Display URL for non-telegram links or when display_url is available if not is_telegram_message: display_url = getattr(webpage, "display_url", None) url = getattr(webpage, "url", None) if display_url: - html_parts.append(f'
{display_url}
') + html_parts.append(f'
{display_url}
') elif url: - html_parts.append(f'
{url}
') + html_parts.append(f'
{url}
') # Add photo if available if photo := getattr(webpage, "photo", None): @@ -711,8 +742,10 @@ class PostParser: file = f"{channel_username}/{message.id}/{file_unique_id}" digest = generate_media_digest(file) url = f"{base_url}/media/{file}/{digest}" - html_parts.append(f'
' - f'
') + html_parts.append(f'
') + html_parts.append(f'') + html_parts.append(f'
') html_parts.append('
') return '\n'.join(html_parts) @@ -746,7 +779,8 @@ class PostParser: result = text offset = 0 - for match in re.finditer(r'https?://[^\s<>"\']+', text): # Find all URLs that are not already in HTML tags + # Find all URLs that are not already in HTML tags + for match in re.finditer(r'https?://[^\s<>"\']+', text): start, end = match.span() # Check if URL is inside an tag @@ -824,20 +858,8 @@ class PostParser: return self._sanitize_html(result_html) if result_html else None except Exception as e: - # Log the type of the object causing the error - obj_type = type(message).__name__ - # Try to get an identifier safely - log_id = None - try: - if isinstance(message, dict): - log_id = message.get('message_id', 'unknown_dict_id') - else: - log_id = getattr(message, 'id', 'unknown_object_id') - except Exception: - log_id = "id_retrieval_failed" - - logger.error(f"reactions_views_parsing_error: Type '{obj_type}', MsgID '{log_id}', Error: {str(e)}", exc_info=True) - logger.error(f"Problematic object data: {str(message)}") + logger.error(f"reactions_views_links_error: message_id {message.id}, error {str(e)}") + logger.error(f"reactions_views_links_error message: {str(message)}") return None def _format_poll(self, poll) -> str: @@ -901,7 +923,8 @@ class PostParser: 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 + 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 @@ -919,8 +942,8 @@ class PostParser: found = False for item in existing_data: if (item.get('channel') == file_data['channel'] and - item.get('post_id') == file_data['post_id'] and - item.get('file_unique_id') == file_data['file_unique_id']): + item.get('post_id') == file_data['post_id'] and + item.get('file_unique_id') == file_data['file_unique_id']): item['added'] = datetime.now().timestamp() found = True break