')
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("
")
show_caption_above = getattr(message, 'show_caption_above_media', False)
if show_caption_above:
if text_html:
content_body.append(text_html)
content_body.append("
")
if media_html:
content_body.append(media_html)
content_body.append("
")
else:
if media_html:
content_body.append(media_html)
content_body.append("
")
if text_html:
content_body.append(text_html)
content_body.append("
")
if poll_html: content_body.append(poll_html) # Poll
if special_html := self._format_special_media(message): content_body.append(special_html) # Special media info blocks
if message.forward_origin: content_body.append(f"--- Forwarded post end ---") # Forward info end
content_body.append(f'
is not in the allowed-tags whitelist and would be stripped).
if debug:
raw_escaped = html.escape(str(data.get("raw_message", "")))
html_content.append(f'{raw_escaped}')
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 = ['')
return_html = ' '.join(flags_html)
return return_html
def process_message(self, message: Message, include_raw: bool = False, sanitize: bool = True) -> Dict[Any, Any]:
"""Build the processed representation of a message.
Args:
message: the Pyrogram message.
include_raw: when True, compute the (expensive) full ``str(message)``
serialization into ``result['raw_message']``. Only JSON output and
debug HTML need it; feed generation must pass False to avoid
serializing every post.
sanitize: when True, run the html body and footer through a single
bleach pass each. Single-post HTML and JSON need this (there is no
feed-level pass on those paths). Feed generation passes False and
relies on the per-post sanitize in rss_generator._render_pipeline
(per-post for BOTH RSS and HTML), so no fragment is sanitized more
than once.
"""
# Compute html body once — avoids triple _generate_html_body calls.
# The internal per-fragment sanitize passes were removed (4.4); sanitize
# exactly once per output boundary here when requested.
html_body = self._generate_html_body(message)
# NOTE (stage-4 4.4 consequence): flags are now extracted from the PRE-sanitize
# body (the per-fragment sanitize that used to run inside _generate_html_body was
# removed). Legitimate links are unaffected — bleach keeps whitelisted
# — so link/foreign_channel/mention flags are identical for
# normal content. They can differ ONLY for URL-like text that bleach would strip
# (e.g. a URL inside a disallowed attribute); flags are non-security (used for
# exclude_flags filtering / display), so this edge divergence is accepted rather
# than re-adding a per-message sanitize pass that 4.4 deliberately eliminated.
flags = self._extract_flags(message, html_body=html_body)
footer = self.generate_html_footer(message, flags_list=flags)
if sanitize:
html_body = self._sanitize_html(html_body)
footer = self._sanitize_html(footer)
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': html_body,
'footer': footer
},
'flags': flags,
'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),
'show_caption_above_media': getattr(message, 'show_caption_above_media', False),
}
if include_raw:
# Full serialization of the message — expensive; only for JSON/debug.
result['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:
# Delegate to the single project-wide bleach config (sanitizer.sanitize_html).
# This drops the former fail-open branch: sanitize_html is fail-closed and
# html.escape()s on any bleach error (registry §3.2).
return sanitize_html(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''
return f''
# 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''
# 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''
# 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''
# 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''
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'
')
# NOTE: sanitize is NOT applied here. Sanitization happens exactly once per
# output boundary (process_message for single-post/JSON; in rss_generator the
# per-post pass in _render_pipeline for BOTH RSS and HTML). See the map (4.4).
html_body = '\n'.join(content_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']
# Poll media (Kurigram 2.2.23): a poll may carry description_media which IS
# renderable through the regular /media pipeline.
poll_media_obj, poll_media_kind = _poll_media_object(message)
if message.media == MessageMediaType.PAID_MEDIA:
# Paid media cannot be downloaded (it is paid content) — render an info
# block instead of a media element. Attributes via getattr only.
paid_media = getattr(message, 'paid_media', None)
stars_amount = getattr(paid_media, 'stars_amount', 0) if paid_media else 0
paid_items = getattr(paid_media, 'media', None) if paid_media else None
items_count = len(paid_items) if isinstance(paid_items, (list, tuple)) else 0
content_media.append(f'')
# Info-block-only media types (NO_IMAGE_MEDIA_TYPES) are rendered by
# _format_special_media — do not open an empty message-media container for
# them. PAID_MEDIA (also in that set) is handled by the branch above; POLL is
# not in the set and is let in only when it carries renderable poll media.
elif (message.media
and message.media not in NO_IMAGE_MEDIA_TYPES
and (message.media != MessageMediaType.POLL or poll_media_obj is not None)):
content_media.append(f'')
else:
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.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-файл]')
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'')
elif message.media == MessageMediaType.LIVE_PHOTO:
# Live photo is effectively a short video clip.
content_media.append(f'')
elif message.media == MessageMediaType.STORY:
# Choose the tag from the SAME helper that produced the URL's
# file_unique_id, so the tag type always matches the URL object.
story_media_obj, story_media_kind = _story_media_object(message)
if story_media_obj is not None and story_media_kind == 'video':
content_media.append(f'')
elif story_media_obj is not None:
content_media.append(f'')
elif message.media == MessageMediaType.POLL and poll_media_obj is not None:
if poll_media_kind == 'video':
content_media.append(f'')
else:
content_media.append(f'')
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('')
# Not sanitized here — this fragment is embedded in the body and sanitized
# once at the output boundary (see the 4.4 sanitize coverage map).
html_media = '\n'.join(content_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)
# Guard: skip photo if channel_username is unavailable to avoid broken URLs
if channel_username:
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)
# Not sanitized here — sanitized once at the output boundary (4.4 coverage map):
# process_message for single-post/JSON; per-post in _render_pipeline for feeds
# (both RSS and HTML).
html_footer = '\n'.join(content_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()
# An empty reactions object (reactions.reactions == []) produces no
# spans; appending the empty string would emit a leading
# ' | ' separator in the footer. Skip it
# (registry §3.15). Affects single posts too.
if reactions_html:
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'')
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))
# Raw fragment — embedded in the footer, sanitized once at the output
# boundary (see the 4.4 sanitize coverage map). Not sanitized here.
result_html = '
'.join(parts) if parts else None
return 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):
q = poll.question
question_str = q.text if hasattr(q, 'text') else str(q)
poll_text = f"📊 Poll: {question_str}\n"
if hasattr(poll, "options") and poll.options:
for i, option in enumerate(poll.options, 1):
opt_text = getattr(option, 'text', '')
opt_str = opt_text.text if hasattr(opt_text, 'text') else str(opt_text)
poll_text += f"{i}. {opt_str}\n"
poll_text += "\n→ Vote in Telegram 🔗\n"
return f''
return None
except Exception as e:
logger.error(f"poll_parsing_error: {str(e)}")
return ''
@staticmethod
def _format_osm_link(location) -> Union[str, None]:
"""Build an OpenStreetMap link for a location object, or None if coords are unusable."""
latitude = getattr(location, 'latitude', None) if location else None
longitude = getattr(location, 'longitude', None) if location else None
if not isinstance(latitude, (int, float)) or not isinstance(longitude, (int, float)):
return None
osm_url = f"https://www.openstreetmap.org/?mlat={latitude}&mlon={longitude}#map=16/{latitude}/{longitude}"
return f'{latitude:.5f}, {longitude:.5f}'
def _format_special_media(self, message: Message) -> Union[str, None]:
"""Render an info block for media types that carry no downloadable file.
Covers giveaways, giveaway winners, checklists, contacts, locations, venues,
dice, games, invoices and UNSUPPORTED content (Kurigram 2.2.23). Each block is
gated on message.media so unrelated messages never render these. All new
Message attributes are accessed via getattr only (older objects/mocks do not
define them) and ALL user-controlled strings go through html.escape.
"""
try:
media = getattr(message, 'media', None)
block = None
if media == MessageMediaType.GIVEAWAY and (giveaway := getattr(message, 'giveaway', None)):
quantity = getattr(giveaway, 'quantity', None)
months = getattr(giveaway, 'months', None)
stars = getattr(giveaway, 'stars', None)
until_date = getattr(giveaway, 'until_date', None)
description = getattr(giveaway, 'description', None)
block = f"🎁 Giveaway: {quantity} prize(s)"
if months: block += f" × {months} months Premium"
elif stars: block += f" × {stars} Stars"
if until_date is not None and hasattr(until_date, 'strftime'):
block += f" — until {until_date.strftime('%d/%m/%Y')}"
if isinstance(description, str) and description.strip():
block += f"
{html.escape(description.strip())}"
elif media == MessageMediaType.GIVEAWAY_WINNERS and (winners := getattr(message, 'giveaway_winners', None)):
winner_count = getattr(winners, 'winner_count', None)
quantity = getattr(winners, 'quantity', None)
prize_description = getattr(winners, 'prize_description', None)
block = f"🏆 Giveaway winners: {winner_count} of {quantity}"
if isinstance(prize_description, str) and prize_description.strip():
block += f"
{html.escape(prize_description.strip())}"
elif media == MessageMediaType.CHECKLIST and (checklist := getattr(message, 'checklist', None)):
title = getattr(checklist, 'title', None)
title_str = title if isinstance(title, str) else ''
lines = [f"📝 {html.escape(title_str)}" if title_str else "📝 Checklist"]
for task in (getattr(checklist, 'tasks', None) or []):
completed = bool(getattr(task, 'completed_by', None) or getattr(task, 'completion_date', None))
mark = "☑" if completed else "☐"
task_text = getattr(task, 'text', '')
task_str = task_text if isinstance(task_text, str) else str(task_text)
lines.append(f"{mark} {html.escape(task_str)}")
block = '
'.join(lines)
elif media == MessageMediaType.CONTACT and (contact := getattr(message, 'contact', None)):
first_name = getattr(contact, 'first_name', None) or ''
last_name = getattr(contact, 'last_name', None) or ''
phone_number = getattr(contact, 'phone_number', None) or ''
full_name = ' '.join(part for part in [str(first_name), str(last_name)] if part)
block = f"👤 {html.escape(full_name)}"
if phone_number:
block += f" — {html.escape(str(phone_number))}"
elif media == MessageMediaType.LOCATION and (location := getattr(message, 'location', None)):
osm_link = self._format_osm_link(location)
block = f"📍 Location: {osm_link}" if osm_link else "📍 Location"
elif media == MessageMediaType.VENUE and (venue := getattr(message, 'venue', None)):
venue_title = getattr(venue, 'title', None) or ''
venue_address = getattr(venue, 'address', None) or ''
venue_label = ', '.join(part for part in [str(venue_title), str(venue_address)] if part)
block = f"📍 {html.escape(venue_label)}" if venue_label else "📍 Venue"
if osm_link := self._format_osm_link(getattr(venue, 'location', None)):
block += f" — {osm_link}"
elif media == MessageMediaType.DICE and (dice := getattr(message, 'dice', None)):
dice_emoji = getattr(dice, 'emoji', None) or '🎲'
dice_value = getattr(dice, 'value', None)
block = f"🎲 {html.escape(str(dice_emoji))}: {dice_value}"
elif media == MessageMediaType.GAME:
game_title = getattr(getattr(message, 'game', None), 'title', None)
if isinstance(game_title, str) and game_title.strip():
block = f"🎮 Game: {html.escape(game_title.strip())}"
else:
block = "🎮 Game"
elif media == MessageMediaType.INVOICE:
block = "🧾 Invoice"
elif media == MessageMediaType.UNSUPPORTED:
block = "⚠️ This post contains content not supported by the bridge — open it in Telegram."
if block is None:
return None
return f''
except Exception as e:
logger.error(f"special_media_parsing_error: message_id {getattr(message, 'id', 'unknown')}, error {str(e)}")
return None
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.WEB_PAGE: lambda m: m.web_page.photo.file_unique_id if m.web_page and m.web_page.photo else None,
# New media types (Kurigram 2.2.23): getattr-only access, the
# attributes do not exist on older Message objects/mocks.
MessageMediaType.LIVE_PHOTO: lambda m: getattr(getattr(m, 'live_photo', None), 'file_unique_id', None),
MessageMediaType.STORY: lambda m: getattr(_story_media_object(m)[0], 'file_unique_id', None),
MessageMediaType.POLL: lambda m: getattr(_poll_media_object(m)[0], 'file_unique_id', 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
async def _flush_pending_media_ids(self) -> None:
"""Persist media file-id records collected during rendering with ONE bulk upsert.
Called by the caller (get_post / rss_generator) after rendering completes.
Runs the blocking SQLite write in a thread. No-op when nothing was collected.
"""
entries = self._pending_media_ids
if not entries:
return
try:
await asyncio.to_thread(upsert_media_file_ids_bulk_sync, DB_PATH, entries)
logger.debug(f"persist_media_file_ids_bulk: upserted {len(entries)} records")
except Exception as e:
logger.error(f"file_id_bulk_save_error: error bulk-upserting {len(entries)} records, error {str(e)}")
finally:
# Clear regardless of outcome so a retry does not double-persist a stale batch.
self._pending_media_ids = []
def _save_media_file_ids(self, message: Message) -> None:
"""Collect a media file-id record for later bulk persistence.
IMPORTANT (4.2): this runs inside the render thread, so it must NOT touch
asyncio or the DB. It only appends to self._pending_media_ids; the caller
flushes the batch via _flush_pending_media_ids() after rendering.
"""
try:
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
file_unique_id = ''
new_media_obj = None # selected object among the Kurigram 2.2.23 media sources
if message.photo: file_unique_id = message.photo.file_unique_id
elif message.video: file_unique_id = message.video.file_unique_id
elif message.document: file_unique_id = message.document.file_unique_id
elif message.audio: file_unique_id = message.audio.file_unique_id
elif message.voice: file_unique_id = message.voice.file_unique_id
elif message.video_note: file_unique_id = message.video_note.file_unique_id
elif message.animation: file_unique_id = message.animation.file_unique_id
elif message.sticker: file_unique_id = message.sticker.file_unique_id
elif message.web_page and message.web_page.photo:
file_unique_id = message.web_page.photo.file_unique_id
# New media types (Kurigram 2.2.23): getattr-only, the attributes do
# not exist on older Message objects/mocks. paid_media is deliberately
# NOT collected — it cannot be downloaded (paid content).
elif getattr(message, 'live_photo', None):
new_media_obj = message.live_photo
elif (story_media := _story_media_object(message)[0]) is not None:
new_media_obj = story_media
elif (poll_media := _poll_media_object(message)[0]) is not None:
new_media_obj = poll_media
if new_media_obj is not None:
# The >100MB message.video guard above does not cover these
# sources (live photo, story video, poll description video) —
# apply the same "don't cache large videos" rule here.
file_size = getattr(new_media_obj, 'file_size', None)
if isinstance(file_size, int) and file_size > 100 * 1024 * 1024:
return
file_unique_id = getattr(new_media_obj, 'file_unique_id', '') or ''
if file_unique_id:
added_ts = datetime.now().timestamp()
# Thread-safe: just append; the caller persists the batch.
self._pending_media_ids.append((channel_username, message.id, file_unique_id, added_ts))
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