' if reply_to: reply_text = reply_to.text or reply_to.caption or '' if len(reply_text) > 100: reply_text = reply_text[:100] + '...' channel_username = getattr(reply_to.sender_chat, "username", None) if channel_username: reply_link = f'#{reply_to.id}' return f'

Reply to {reply_link}: {reply_text}

' return f'

Reply to #{reply_to.id}: {reply_text}

' return None def _extract_reactions(self, message: Message) -> Union[Dict[str, int], None]: if reactions := getattr(message, 'reactions', None): return {r.emoji: r.count for r in reactions.reactions} return None def _extract_flags(self, message: Message) -> List[str]: #Tests: tests/postparser_extract_flags.py # Use raw text/caption for some checks before HTML processing message_text_str = str(message.text or message.caption or '') # Use HTML body for checks involving formatted text or links within HTML message_body_html = self._generate_html_body(message) flags = [] # Add "fwd" flag for forwarded messages if message.forward_origin: flags.append("fwd") # Add flag "video" if the message media is VIDEO or ANIMATION and the body text is up to 200 characters. if (message.media in [MessageMediaType.VIDEO, MessageMediaType.ANIMATION, MessageMediaType.VIDEO_NOTE] and len((message.text or message.caption or '').strip()) <= 200): flags.append("video") # Add flag "story" if the message contains a story if message.media == MessageMediaType.STORY: flags.append("story") # Add flag "audio" if the message media is AUDIO if (message.media in [MessageMediaType.AUDIO, MessageMediaType.VOICE] and len((message.text or message.caption or '').strip()) <= 200): flags.append("audio") # Add flag for posts without images if not message.media or message.media == MessageMediaType.POLL: flags.append("no_image") # Add flag for sticker messages if message.media == MessageMediaType.STICKER: flags.append("sticker") # Add flag for poll messages if message.media == MessageMediaType.POLL: flags.append("poll") # Check if the message text contains variations of the word "стрим", "вебинар" # or "онлайн-лекция" in a case-insensitive manner. if re.search(r'(?i)\b(стрим\w*|livestream|онлайн-лекци[яю]|вебинар\w*)\b', message_text_str): flags.append("stream") # Check if the message text contains the word "донат" in a case-insensitive manner. if re.search(r'(?i)\bдонат\w*\b', message_text_str): flags.append("donat") # Check for pay.cloudtips.ru links and add donat flag if re.search(r'(?i)pay\.cloudtips\.ru', message_text_str): flags.append("donat") # Check for t.me/boost links and add donat flag if re.search(r'https?://(?:www\.)?t\.me/boost/', message_text_str): flags.append("donat") # Check if the post's reactions contain more clown emojis (🤡) or poo emojis (💩). if reactions := getattr(message, "reactions", None): for reaction in getattr(reactions, "reactions", []): if reaction.emoji == "🤡" and reaction.count >= 30: flags.append("clownpoo") break if reaction.emoji == "💩" and reaction.count >= 30: flags.append("clownpoo") break # 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 if re.search(r'(?i)(Дзен\.Премиум|Sponsr|Бусти|Boosty)', message_text_str): flags.append("paywall") # --- Link Flags --- # Check if the message consists ONLY of a link (text/caption) or only has a webpage preview is_only_link = False if message.web_page and not message_text_str.strip(): # Case 1: Only a web_page preview, no text/caption is_only_link = True elif message_text_str.strip(): # Case 2: Text/caption exists, check if it's just a single non-t.me URL # Regex matches string starting and ending with a non-t.me URL url_pattern = r'^https?://(?!(?:www\.)?t\.me)[^\s<>"\']+$' if re.fullmatch(url_pattern, message_text_str.strip()): is_only_link = True if is_only_link: flags.append("only_link") # Check if the message contains any http/https links (excluding t.me) # Only add 'link' if 'only_link' isn't already set if not is_only_link: # Search within the generated HTML body to catch links in hrefs as well if (re.search(r'https?://(?!(?:www\.)?t\.me)[^\s<>"\']+', message_body_html) or re.search(r'href=[\"\']https?://(?!(?:www\.)?t\.me)[^\"\']+[\"\']', message_body_html)): flags.append("link") # --- End Link Flags --- # Check if the message contains channel mentions in the format @name if re.search(r'@[a-zA-Z][a-zA-Z0-9_]{3,}', message_body_html): flags.append("mention") try: # Find links with a '+' after t.me/ indicating a hidden channel link. hidden_links = re.findall(r'https?://(?:www\.)?t\.me/\+([A-Za-z0-9]+)', message_body_html) # Find links without a '+' after t.me/ indicating an open (foreign) channel link. open_links = re.findall(r'https?://(?:www\.)?t\.me/(?!\+)([A-Za-z0-9_]+)', message_body_html) # Find links with pattern t.me/boost/channel_name boost_links = re.findall(r'https?://(?:www\.)?t\.me/boost/([A-Za-z0-9_]+)', message_body_html) if hidden_links: flags.append("hid_channel") current_channel = self.get_channel_username(message) # Check regular open links for open_link in open_links: if current_channel is None or (open_link.lower() != current_channel.lower() and open_link.lower() != 'boost'): flags.append("foreign_channel") break # Check boost links separately - only consider as foreign if the boosted channel is not current channel if "foreign_channel" not in flags: for boost_channel in boost_links: if current_channel is None or boost_channel.lower() != current_channel.lower(): flags.append("foreign_channel") break except Exception as e: logger.error(f"tme_link_extraction_error: message_id {message.id}, error {str(e)}") # 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 = [] if debug: html_content.append(f'

Title: {data["html"]["title"]}

') html_content.append(f'

{data["html"]["body"]}

') html_content.append(f'

{data["html"]["footer"]}

') # Add raw JSON debug output if debug is enabled if debug: html_content.append(f'

{data["raw_message"]}

') html_data = '\n'.join(html_content) return html_data def _format_flags(self, flags_list: list) -> str: if not Config['show_post_flags']: return '' return_html = "" if flags_list: flags_html = ['

'] for flag in flags_list: flags_html.append(f'🏷 {flag}') flags_html.append('

') return_html = ' '.join(flags_html) return return_html def process_message(self, message: Message) -> Dict[Any, Any]: result = { 'channel': self.get_channel_username(message), 'message_id': message.id, 'date': datetime.timestamp(message.date) if message.date else None, 'date_formatted': message.date.strftime('%Y-%m-%d %H:%M:%S') if message.date else None, 'text': message.text or message.caption or '', 'html': { 'title': self._generate_title(message), 'body': self._generate_html_body(message), 'footer': self.generate_html_footer(message) }, 'flags': self._extract_flags(message), 'author': self._get_author_info(message), 'views': message.views, 'reactions': self._extract_reactions(message), 'media_group_id': message.media_group_id, 'service': getattr(message, "service", None), 'raw_message': str(message) } # Add webpage data if available if webpage := getattr(message, "web_page", None): result['webpage'] = { 'type': getattr(webpage, 'type', None), 'url': getattr(webpage, 'url', None), 'display_url': getattr(webpage, 'display_url', None), 'site_name': getattr(webpage, 'site_name', None), 'title': getattr(webpage, 'title', None), 'description': getattr(webpage, 'description', None), 'has_large_media': getattr(webpage, 'has_large_media', False), 'is_telegram_link': getattr(webpage, 'type', '') == 'telegram_message' } return result 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_attributes = { 'a': ['href', 'title', 'target'], 'img': ['src', 'alt', 'style'], 'video': ['controls', 'src', 'style'], 'audio': ['controls', 'style'], 'source': ['src', 'type'], 'div': ['class', 'style'], 'span': ['class'] } try: css_sanitizer = CSSSanitizer( allowed_css_properties=["max-width", "max-height", "object-fit", "width", "height"] ) sanitized_html = HTMLSanitizer( html_raw, tags=allowed_tags, attributes=allowed_attributes, protocols=['http', 'https', 'tg'], css_sanitizer=css_sanitizer, strip=True, ) return sanitized_html except Exception as e: logger.error(f"html_sanitization_error: error {str(e)}") return html_raw def _format_forward_info(self, message: Message) -> Union[str, None]: if forward_origin := getattr(message, "forward_origin", None): # Case 1: Channel or supergroup forward (has chat attribute) if sender_chat := getattr(forward_origin, "chat", None): forward_title = getattr(sender_chat, "title", "Unknown channel") forward_username = getattr(sender_chat, "username", None) if forward_username: forward_link = f'{forward_title} (@{forward_username})' return f'

--- Forwarded from {forward_link} ---

' return f'

--- Forwarded from {forward_title} ---

' # Case 2: Hidden user (MessageOriginHiddenUser) if hasattr(forward_origin, "sender_user_name") and forward_origin.sender_user_name: sender_name = forward_origin.sender_user_name return f'

--- Forwarded from {sender_name} ---

' # Case 3: Regular user (MessageOriginUser) if hasattr(forward_origin, "sender_user") and forward_origin.sender_user: user = forward_origin.sender_user name_parts = [] if hasattr(user, "first_name") and user.first_name: name_parts.append(user.first_name) if hasattr(user, "last_name") and user.last_name: name_parts.append(user.last_name) sender_name = " ".join(name_parts) if name_parts else "Unknown user" if hasattr(user, "username") and user.username: sender_name = f"{sender_name} (@{user.username})" return f'

--- Forwarded from {sender_name} ---

' # Case 4: Channel without username (MessageOriginChannel) if hasattr(forward_origin, "chat_id") and hasattr(forward_origin, "title"): title = forward_origin.title if forward_origin.title else "Unknown channel" return f'

--- Forwarded from {title} ---

' # Case 5: Default for any other type of forward_origin origin_type = getattr(forward_origin, "type", "unknown") logger.debug(f"unhandled_forward_type: {origin_type}, data: {forward_origin}") return f'

--- Forwarded message ---

' return None def _get_post_text_with_urls(self, message: Message) -> Union[str, None]: if message.text: text = message.text.html elif message.caption: text = message.caption.html else: return None text = text.replace('\n', '
') # Replace newlines with
text_html = self._add_hyperlinks_to_raw_urls(text) return text_html def _generate_html_body(self, message: Message) -> str: content_body = [] text_html = self._get_post_text_with_urls(message) poll_html = self._format_poll(message) forward_html = self._format_forward_info(message) reply_html = self._format_reply_info(message) media_html = self._generate_html_media(message) content_body.append(f'

') if forward_html: content_body.append(forward_html) # Forward info content_body.append("
") if reply_html: content_body.append(reply_html) # Reply info content_body.append("
") if text_html: content_body.append(text_html) # Post text content_body.append("
") content_body.append(media_html) # Media content_body.append("
") if poll_html: content_body.append(poll_html) # Poll if message.forward_origin: content_body.append(f"--- Forwarded post end ---") # Forward info end content_body.append(f'

') html_body = '\n'.join(content_body) html_body = self._sanitize_html(html_body) return html_body def _generate_html_media(self, message: Message) -> str: self._save_media_file_ids(message) # Save media file_ids for caching content_media = [] base_url = Config['pyrogram_bridge_url'] if message.media and message.media != "MessageMediaType.POLL": content_media.append(f'

') # Handle stories if message.media == MessageMediaType.STORY and hasattr(message, "story"): story = message.story if story and hasattr(story, "video") and story.video: story_file_id = story.video.file_unique_id if story_file_id: sender_username = getattr(story.sender_chat, "username", None) channel_username = sender_username or self.get_channel_username(message) if channel_username: file = f"{channel_username}/{story.id}/{story_file_id}" digest = generate_media_digest(file) url = f"{base_url}/media/{file}/{digest}" content_media.append(f'

') content_media.append(f'

📱 Story from @{channel_username}

') logger.debug(f"Collected story video file: {channel_username}/{story.id}/{story_file_id}") content_media.append('

') return '\n'.join(content_media) 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: channel_username = self.get_channel_username(message) file = f"{channel_username}/{message.id}/{file_unique_id}" digest = generate_media_digest(file) url = f"{base_url}/media/{file}/{digest}" logger.debug(f"Collected media file: {channel_username}/{message.id}/{file_unique_id}") # 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'): # 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'

') 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.id}: ch username is missing.") # Add placeholder without link content_media.append(f'

[PDF-файл]

') elif message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]: content_media.append(f'

') elif message.media == MessageMediaType.VIDEO: content_media.append(f'

') elif message.media == MessageMediaType.ANIMATION: content_media.append(f'

') elif message.media == MessageMediaType.VIDEO_NOTE: 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('
') elif message.media == MessageMediaType.VOICE: mime_type = getattr(message.voice, 'mime_type', 'audio/ogg') 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'

') else: content_media.append(f' Sticker {emoji}

') content_media.append('') if webpage := getattr(message, "web_page", None): # Web page preview if webpage_html := self._format_webpage(webpage, message): if len((message.text or message.caption or '').strip()) <= 10: content_media.append(f'

') content_media.append(webpage_html) content_media.append('

') html_media = '\n'.join(content_media) html_media = self._sanitize_html(html_media) return html_media def _format_webpage(self, webpage, message) -> Union[str, None]: base_url = Config['pyrogram_bridge_url'] try: # Check if this is a Telegram message link is_telegram_message = getattr(webpage, "type", "") == "telegram_message" if is_telegram_message: # Telegram message preview with distinctive styling html_parts = [("""

""")] html_parts.append(f'

📱 Telegram

') else: # Regular webpage preview html_parts = ['

'] # Add site name if available if site_name := getattr(webpage, "site_name", None): html_parts.append(f'

{site_name}

') # Add title with link if available if title := getattr(webpage, "title", None): url = getattr(webpage, "url", "#") 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}

') # 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}

') elif url: html_parts.append(f'

{url}

') # Add photo if available if photo := getattr(webpage, "photo", None): if file_unique_id := getattr(photo, "file_unique_id", None): channel_username = self.get_channel_username(message) 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'

') html_parts.append(f'') html_parts.append(f'

') html_parts.append('

') return '\n'.join(html_parts) except Exception as e: logger.error(f"webpage_parsing_error: url {getattr(webpage, 'url', 'unknown')}, error {str(e)}") return None def generate_html_footer(self, message: Message, flags_list: Optional[List[str]] = None) -> str: content_footer = [] content_footer.append('
') if reactions_views_html := self._reactions_views_links(message): # Add reactions, views, date and links content_footer.append(reactions_views_html) # Use provided flags_list if available, otherwise extract from message current_flags = flags_list if flags_list is not None else self._extract_flags(message) if current_flags: flags_html = self._format_flags(current_flags) content_footer.append('
' + flags_html) html_footer = '\n'.join(content_footer) html_footer = self._sanitize_html(html_footer) return html_footer def _add_hyperlinks_to_raw_urls(self, text: str) -> str: try: a_tags = re.finditer(r']*>.*?', text) # Find all existing tags excluded_ranges = [(m.start(), m.end()) for m in a_tags] # Store their positions result = text offset = 0 # 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 is_in_tag = any(tag_start <= start and end <= tag_end for tag_start, tag_end in excluded_ranges) if not is_in_tag: url = match.group() replacement = f'{url}' # Apply replacement considering offset result = result[:start + offset] + replacement + result[end + offset:] offset += len(replacement) - (end - start) return result except Exception as e: logger.error(f"url_processing_error: error {str(e)}") return text def _reactions_views_links(self, message: Message) -> Union[str, None]: try: parts = [] # First line: reactions + views + date first_line_parts = [] # Add reactions reactions_html = "" if reactions := getattr(message, "reactions", None): for reaction in reactions.reactions: if getattr(reaction, "is_paid", False): emoji = "⭐" elif hasattr(reaction, "emoji") and reaction.emoji: emoji = reaction.emoji elif hasattr(reaction, "custom_emoji_id"): emoji = "❓" # Then check custom emoji else: emoji = "❓" # Default for unknown cases reactions_html += f'{emoji} {reaction.count} ' reactions_html = reactions_html.rstrip() first_line_parts.append(reactions_html) # Add views if views := getattr(message, "views", None): first_line_parts.append(f'{views} 👁') # Add date if message.date: formatted_date = message.date.strftime("%d/%m/%y, %H:%M:%S") first_line_parts.append(f'{formatted_date}') if message.id: first_line_parts.append(f'#{message.id}') if first_line_parts: parts.append(' | '.join(first_line_parts)) # Second line: links channel_identifier = self.get_channel_username(message) if channel_identifier: links = [] base_url = Config['pyrogram_bridge_url'] if channel_identifier.startswith('-100'): channel_id = channel_identifier[4:] links.append(f'Open in Telegram') links.append(f'Open in Web') else: links.append(f'Open in Telegram') links.append(f'Open in Web') if Config['show_bridge_link']: token = Config['token'] links.append(f'Open in Bridge') if links: parts.append(' | '.join(links)) result_html = '
'.join(parts) if parts else None return self._sanitize_html(result_html) if result_html else None except Exception as e: logger.error(f"reactions_views_links_error: message_id {message.id}, error {str(e)}") logger.error(f"reactions_views_links_error message_object: {str(message)}") # Keep original log for context return None def _format_poll(self, message: Message) -> Union[str, None]: #TODO: refactoring to parts.append try: if poll := getattr(message, "poll", None): 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), "
")}

' return None except Exception as e: logger.error(f"poll_parsing_error: {str(e)}") return '

[Error displaying poll]

' def _get_file_unique_id(self, message: Message) -> Union[str, None]: try: media_mapping = { 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, MessageMediaType.STORY: lambda m: m.story.video.file_unique_id if m.story and hasattr(m.story, "video") else None, MessageMediaType.WEB_PAGE: lambda m: m.web_page.photo.file_unique_id if m.web_page and m.web_page.photo else None } 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 def _save_media_file_ids(self, message: Message) -> None: try: file_data = { 'channel': '', 'post_id': 0, 'file_unique_id': '', 'added': .0 } channel_username = self.get_channel_username(message) if not channel_username: logger.error(f"channel_username_error: no username found for chat in message {message.id}") return if message.media: # Skip large videos - they shouldn't be cached permanently if message.video and message.video.file_size and message.video.file_size > 100 * 1024 * 1024: return if message.photo: file_data['file_unique_id'] = message.photo.file_unique_id elif message.video: file_data['file_unique_id'] = message.video.file_unique_id elif message.document: file_data['file_unique_id'] = message.document.file_unique_id elif message.audio: file_data['file_unique_id'] = message.audio.file_unique_id elif message.voice: file_data['file_unique_id'] = message.voice.file_unique_id 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.story and hasattr(message.story, "video") and message.story.video: file_data['file_unique_id'] = message.story.video.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 file_data['post_id'] = message.id file_data['added'] = datetime.now().timestamp() # For stories, we need to save with original story ID and channel if message.media == MessageMediaType.STORY and hasattr(message, "story"): story = message.story if story and hasattr(story, "video") and story.video: story_channel = getattr(story.sender_chat, "username", None) if story_channel: # Create a separate entry for the story story_file_data = file_data.copy() story_file_data['channel'] = story_channel story_file_data['post_id'] = story.id file_path = os.path.join(os.path.abspath("./data"), '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 story file already exists by all three fields found = False for item in existing_data: if (item.get('channel') == story_file_data['channel'] and item.get('post_id') == story_file_data['post_id'] and item.get('file_unique_id') == story_file_data['file_unique_id']): item['added'] = datetime.now().timestamp() found = True break # Add new entry if not found if not found: existing_data.append(story_file_data) logger.debug(f"Saved story media file: {story_channel}/{story.id}/{story_file_data['file_unique_id']}") 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"story_file_id_save_error: error writing to {file_path}, error {str(e)}") # Standard file saving process file_path = os.path.join(os.path.abspath("./data"), '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 already exists by all three fields 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['added'] = datetime.now().timestamp() found = True break # Add new entry if not found if not found: 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}, error {str(e)}") except Exception as e: logger.error(f"file_id_collection_error: message_id {message.id}, error {str(e)}") def get_channel_username(self, message: Message) -> Union[str, None]: """Extract channel username or ID from message""" chat = message.chat if hasattr(message, 'chat') else message if not chat: return None # First try to get username if hasattr(chat, 'usernames') and chat.usernames: # Check many usernames active_usernames = [u.username for u in chat.usernames if u.active] if active_usernames: return active_usernames[0] if hasattr(chat, 'username') and chat.username: return chat.username # Check single username # Return numeric ID only if no username found if isinstance(chat.id, int) and str(chat.id).startswith('-100'): return str(chat.id) logger.error(f"channel_username_error: no username or valid ID found for chat") return None