feat: upgrade to Kurigram 2.2.23 and render new Telegram media types
Docker Image CI / build (pull_request) Has been cancelled
Docker Image CI / build (pull_request) Has been cancelled
Upgrade Kurigram 2.2.22 -> 2.2.23 (no breaking API changes; whole suite green on the new version unchanged). New in 2.2.23 and now rendered by the bridge: - LIVE_PHOTO: rendered as a looping video via the /media pipeline - Poll media (description_media): photo/video/animation/sticker attached to a poll is rendered above the poll block; find_file_id_in_message searches poll description_media AND explanation_media instead of early-returning None for polls Previously silent media types now rendered: - STORY: reposted story photo/video via the /media pipeline (render and URL derive from the same helper-selected object) - GIVEAWAY / GIVEAWAY_WINNERS: info blocks with quantity/prize/date - PAID_MEDIA: info block with star amount (content is not downloadable) - CHECKLIST: title + tasks with done marks - CONTACT, LOCATION, VENUE (with OSM links), DICE, GAME, INVOICE, UNSUPPORTED: info blocks, all user-controlled strings html-escaped Supporting changes: - titles for all new types in _media_message_title - no_image flag covers info-block types; poll with media is not no_image; LIVE_PHOTO counts as video for the video flag - _save_media_file_ids collects live_photo/story/poll media for the background cache warmer, honoring the >100MB video guard; paid media is never collected - info-block types no longer emit an empty message-media div - 53 new tests (media URLs with digest, download symmetry, XSS escaping, flags, size guard); 314 total Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
+34
-3
@@ -287,9 +287,26 @@ if __name__ == "__main__":
|
|||||||
async def find_file_id_in_message(message: Message, file_unique_id: str) -> Union[str, None]:
|
async def find_file_id_in_message(message: Message, file_unique_id: str) -> Union[str, None]:
|
||||||
"""Find file_id by checking all possible media types in message"""
|
"""Find file_id by checking all possible media types in message"""
|
||||||
if message.media == MessageMediaType.POLL:
|
if message.media == MessageMediaType.POLL:
|
||||||
logger.debug(f"Message {message.id} is a poll, skipping media search")
|
# Kurigram 2.2.23: polls may carry media in description_media and
|
||||||
|
# explanation_media (MessageContent objects). The bridge renders only
|
||||||
|
# description_media, but explanation_media is searched too: if a signed URL
|
||||||
|
# for it ever exists, the download must still work. getattr-only access —
|
||||||
|
# older Poll objects/mocks do not define these fields.
|
||||||
|
poll = getattr(message, 'poll', None)
|
||||||
|
for container_name in ('description_media', 'explanation_media'):
|
||||||
|
content = getattr(poll, container_name, None) if poll else None
|
||||||
|
if content is None:
|
||||||
|
continue
|
||||||
|
for media_attr in ('photo', 'video', 'animation', 'sticker',
|
||||||
|
'document', 'audio', 'voice', 'video_note'):
|
||||||
|
media_obj = getattr(content, media_attr, None)
|
||||||
|
if media_obj is None:
|
||||||
|
continue
|
||||||
|
if getattr(media_obj, 'file_unique_id', None) == file_unique_id:
|
||||||
|
return getattr(media_obj, 'file_id', None)
|
||||||
|
logger.debug(f"Message {message.id} is a poll, media '{file_unique_id}' not found in poll content")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
media_found = []
|
media_found = []
|
||||||
if message.photo:
|
if message.photo:
|
||||||
media_found.append(f"photo ({message.photo.file_unique_id})")
|
media_found.append(f"photo ({message.photo.file_unique_id})")
|
||||||
@@ -327,7 +344,21 @@ async def find_file_id_in_message(message: Message, file_unique_id: str) -> Unio
|
|||||||
media_found.append(f"document ({message.document.file_unique_id})")
|
media_found.append(f"document ({message.document.file_unique_id})")
|
||||||
if message.document.file_unique_id == file_unique_id:
|
if message.document.file_unique_id == file_unique_id:
|
||||||
return message.document.file_id
|
return message.document.file_id
|
||||||
|
# New media types (Kurigram 2.2.23): getattr-only access, the attributes do not
|
||||||
|
# exist on older Message objects/mocks.
|
||||||
|
if live_photo := getattr(message, 'live_photo', None):
|
||||||
|
media_found.append(f"live_photo ({getattr(live_photo, 'file_unique_id', None)})")
|
||||||
|
if getattr(live_photo, 'file_unique_id', None) == file_unique_id:
|
||||||
|
return getattr(live_photo, 'file_id', None)
|
||||||
|
if story := getattr(message, 'story', None):
|
||||||
|
for story_attr in ('photo', 'video'):
|
||||||
|
story_media = getattr(story, story_attr, None)
|
||||||
|
if story_media is None:
|
||||||
|
continue
|
||||||
|
media_found.append(f"story.{story_attr} ({getattr(story_media, 'file_unique_id', None)})")
|
||||||
|
if getattr(story_media, 'file_unique_id', None) == file_unique_id:
|
||||||
|
return getattr(story_media, 'file_id', None)
|
||||||
|
|
||||||
# If we reached here, the file_unique_id was not found
|
# If we reached here, the file_unique_id was not found
|
||||||
channel_id_log = message.chat.id if message.chat else 'unknown_chat'
|
channel_id_log = message.chat.id if message.chat else 'unknown_chat'
|
||||||
logger.warning(f"Could not find media with file_unique_id '{file_unique_id}' in message {message.id} (channel: {channel_id_log}). Found media: {', '.join(media_found) or 'None'}")
|
logger.warning(f"Could not find media with file_unique_id '{file_unique_id}' in message {message.id} (channel: {channel_id_log}). Found media: {', '.join(media_found) or 'None'}")
|
||||||
|
|||||||
+285
-5
@@ -28,6 +28,85 @@ Config = get_settings()
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Media types that never yield a renderable image/video in the feed: they are
|
||||||
|
# represented by info blocks (or nothing), so posts carrying them get "no_image".
|
||||||
|
NO_IMAGE_MEDIA_TYPES = {
|
||||||
|
MessageMediaType.GIVEAWAY,
|
||||||
|
MessageMediaType.GIVEAWAY_WINNERS,
|
||||||
|
MessageMediaType.CHECKLIST,
|
||||||
|
MessageMediaType.CONTACT,
|
||||||
|
MessageMediaType.LOCATION,
|
||||||
|
MessageMediaType.VENUE,
|
||||||
|
MessageMediaType.DICE,
|
||||||
|
MessageMediaType.GAME,
|
||||||
|
MessageMediaType.INVOICE,
|
||||||
|
MessageMediaType.UNSUPPORTED,
|
||||||
|
MessageMediaType.PAID_MEDIA,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _poll_media_object(message):
|
||||||
|
"""Return (media_obj, kind) attached to a poll's description_media, or (None, None).
|
||||||
|
|
||||||
|
Kurigram 2.2.23 polls may carry media in description_media / explanation_media
|
||||||
|
(MessageContent objects). Only description_media is considered for rendering:
|
||||||
|
explanation_media is the quiz-answer explanation attachment and does not belong
|
||||||
|
in a channel feed, so it is deliberately NOT rendered (api_server's download
|
||||||
|
lookup still covers it in case a URL for it exists).
|
||||||
|
|
||||||
|
kind is the render hint: 'img' for photo/sticker, 'video' for video/animation.
|
||||||
|
|
||||||
|
All attribute access goes through getattr: older Poll objects and test mocks do
|
||||||
|
not define these fields. A candidate is accepted only when its file_unique_id is
|
||||||
|
a non-empty str — without it nothing can be served through the /media pipeline,
|
||||||
|
and this also keeps loose MagicMock-based poll mocks from producing false
|
||||||
|
positives via auto-created attributes.
|
||||||
|
"""
|
||||||
|
poll = getattr(message, 'poll', None)
|
||||||
|
if poll is None:
|
||||||
|
return None, None
|
||||||
|
description_media = getattr(poll, 'description_media', None)
|
||||||
|
if description_media is None:
|
||||||
|
return None, None
|
||||||
|
candidates = (
|
||||||
|
('photo', 'img'),
|
||||||
|
('video', 'video'),
|
||||||
|
('animation', 'video'),
|
||||||
|
('sticker', 'img'),
|
||||||
|
)
|
||||||
|
for attr, kind in candidates:
|
||||||
|
media_obj = getattr(description_media, attr, None)
|
||||||
|
if media_obj is None:
|
||||||
|
continue
|
||||||
|
file_unique_id = getattr(media_obj, 'file_unique_id', None)
|
||||||
|
if isinstance(file_unique_id, str) and file_unique_id:
|
||||||
|
return media_obj, kind
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
|
def _story_media_object(message):
|
||||||
|
"""Return (media_obj, kind) for message.story media (video wins over photo), or (None, None).
|
||||||
|
|
||||||
|
kind is the render hint ('video' or 'img'). Rendering, URL generation and file-id
|
||||||
|
collection all go through this helper, so they always agree on WHICH story object
|
||||||
|
is used (e.g. when a story video lacks a usable file_unique_id and the helper
|
||||||
|
falls back to the photo, the tag type follows the fallback too).
|
||||||
|
|
||||||
|
Same defensive rules as _poll_media_object: getattr-only access and a non-empty
|
||||||
|
str file_unique_id requirement.
|
||||||
|
"""
|
||||||
|
story = getattr(message, 'story', None)
|
||||||
|
if story is None:
|
||||||
|
return None, None
|
||||||
|
for attr, kind in (('video', 'video'), ('photo', 'img')):
|
||||||
|
media_obj = getattr(story, attr, None)
|
||||||
|
if media_obj is None:
|
||||||
|
continue
|
||||||
|
file_unique_id = getattr(media_obj, 'file_unique_id', None)
|
||||||
|
if isinstance(file_unique_id, str) and file_unique_id:
|
||||||
|
return media_obj, kind
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
#tests
|
#tests
|
||||||
#http://127.0.0.1:8000/post/html/DragorWW_space/114 — video
|
#http://127.0.0.1:8000/post/html/DragorWW_space/114 — video
|
||||||
@@ -221,6 +300,31 @@ class PostParser:
|
|||||||
if message.media == MessageMediaType.VOICE: return "🎤 Voice"
|
if message.media == MessageMediaType.VOICE: return "🎤 Voice"
|
||||||
if message.media == MessageMediaType.VIDEO_NOTE: return "📱 Video circle"
|
if message.media == MessageMediaType.VIDEO_NOTE: return "📱 Video circle"
|
||||||
if message.media == MessageMediaType.STICKER: return "🎯 Sticker"
|
if message.media == MessageMediaType.STICKER: return "🎯 Sticker"
|
||||||
|
# New media types (Kurigram 2.2.23). New Message attributes are accessed
|
||||||
|
# via getattr only — older objects/mocks may not define them.
|
||||||
|
if message.media == MessageMediaType.LIVE_PHOTO: return "📸 Live Photo"
|
||||||
|
if message.media == MessageMediaType.STORY: return "📖 Story"
|
||||||
|
if message.media == MessageMediaType.GIVEAWAY: return "🎁 Giveaway"
|
||||||
|
if message.media == MessageMediaType.GIVEAWAY_WINNERS: return "🏆 Giveaway winners"
|
||||||
|
if message.media == MessageMediaType.PAID_MEDIA: return "⭐ Paid media"
|
||||||
|
if message.media == MessageMediaType.CHECKLIST:
|
||||||
|
checklist = getattr(message, 'checklist', None)
|
||||||
|
title = getattr(checklist, 'title', None) if checklist else None
|
||||||
|
if isinstance(title, str) and title.strip():
|
||||||
|
return f"📝 Checklist: {title.strip()[:50]}"
|
||||||
|
return "📝 Checklist"
|
||||||
|
if message.media == MessageMediaType.CONTACT: return "👤 Contact"
|
||||||
|
if message.media == MessageMediaType.LOCATION: return "📍 Location"
|
||||||
|
if message.media == MessageMediaType.VENUE:
|
||||||
|
venue = getattr(message, 'venue', None)
|
||||||
|
venue_title = getattr(venue, 'title', None) if venue else None
|
||||||
|
if isinstance(venue_title, str) and venue_title.strip():
|
||||||
|
return f"📍 {venue_title.strip()}"
|
||||||
|
return "📍 Venue"
|
||||||
|
if message.media == MessageMediaType.DICE: return "🎲 Dice"
|
||||||
|
if message.media == MessageMediaType.GAME: return "🎮 Game"
|
||||||
|
if message.media == MessageMediaType.INVOICE: return "🧾 Invoice"
|
||||||
|
if message.media == MessageMediaType.UNSUPPORTED: return "⚠️ Unsupported content"
|
||||||
|
|
||||||
# Web pages (if no text or media title)
|
# Web pages (if no text or media title)
|
||||||
if message.web_page:
|
if message.web_page:
|
||||||
@@ -357,7 +461,9 @@ class PostParser:
|
|||||||
flags.append("fwd")
|
flags.append("fwd")
|
||||||
|
|
||||||
# Add flag "video" if the message media is VIDEO or ANIMATION and the body text is up to 200 characters.
|
# 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
|
# LIVE_PHOTO (Kurigram 2.2.23) renders as a video element, so it counts too.
|
||||||
|
if (message.media in [MessageMediaType.VIDEO, MessageMediaType.ANIMATION, MessageMediaType.VIDEO_NOTE,
|
||||||
|
MessageMediaType.LIVE_PHOTO] and
|
||||||
len((message.text or message.caption or '').strip()) <= 200):
|
len((message.text or message.caption or '').strip()) <= 200):
|
||||||
flags.append("video")
|
flags.append("video")
|
||||||
|
|
||||||
@@ -366,8 +472,12 @@ class PostParser:
|
|||||||
len((message.text or message.caption or '').strip()) <= 200):
|
len((message.text or message.caption or '').strip()) <= 200):
|
||||||
flags.append("audio")
|
flags.append("audio")
|
||||||
|
|
||||||
# Add flag for posts without images
|
# Add flag for posts without images: no media at all, an info-block-only media
|
||||||
if not message.media or message.media == MessageMediaType.POLL:
|
# type (see NO_IMAGE_MEDIA_TYPES), or a poll without renderable description_media.
|
||||||
|
# A poll WITH description_media renders an image/video, so it is NOT flagged.
|
||||||
|
if (not message.media
|
||||||
|
or message.media in NO_IMAGE_MEDIA_TYPES
|
||||||
|
or (message.media == MessageMediaType.POLL and _poll_media_object(message)[0] is None)):
|
||||||
flags.append("no_image")
|
flags.append("no_image")
|
||||||
|
|
||||||
# Add flag for sticker messages
|
# Add flag for sticker messages
|
||||||
@@ -715,6 +825,7 @@ class PostParser:
|
|||||||
|
|
||||||
|
|
||||||
if poll_html: content_body.append(poll_html) # Poll
|
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
|
if message.forward_origin: content_body.append(f"--- Forwarded post end ---") # Forward info end
|
||||||
content_body.append(f'</div><br>')
|
content_body.append(f'</div><br>')
|
||||||
|
|
||||||
@@ -730,7 +841,27 @@ class PostParser:
|
|||||||
|
|
||||||
content_media = []
|
content_media = []
|
||||||
base_url = Config['pyrogram_bridge_url']
|
base_url = Config['pyrogram_bridge_url']
|
||||||
if message.media and message.media != MessageMediaType.POLL:
|
# 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'<div class="message-media">')
|
||||||
|
content_media.append(f'<div class="paid-media">⭐ Paid media ({stars_amount} stars, '
|
||||||
|
f'{items_count} item(s)) — available in Telegram</div>')
|
||||||
|
content_media.append('</div>')
|
||||||
|
# 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'<div class="message-media">')
|
content_media.append(f'<div class="message-media">')
|
||||||
|
|
||||||
file_unique_id = self._get_file_unique_id(message)
|
file_unique_id = self._get_file_unique_id(message)
|
||||||
@@ -791,6 +922,28 @@ class PostParser:
|
|||||||
else:
|
else:
|
||||||
content_media.append(f'<img src="{url}" alt="Sticker {emoji}" style="max-width:100%;'
|
content_media.append(f'<img src="{url}" alt="Sticker {emoji}" style="max-width:100%;'
|
||||||
f'width:auto; height:auto; max-height:200px; object-fit:contain;">')
|
f'width:auto; height:auto; max-height:200px; object-fit:contain;">')
|
||||||
|
elif message.media == MessageMediaType.LIVE_PHOTO:
|
||||||
|
# Live photo is effectively a short video clip.
|
||||||
|
content_media.append(f'<video controls autoplay loop muted src="{url}"'
|
||||||
|
f'style="max-width:100%; width:auto; height:auto; max-height:400px;'
|
||||||
|
f'object-fit:contain;"></video>')
|
||||||
|
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'<video controls src="{url}" style="max-width:100%; width:auto;'
|
||||||
|
f'height:auto; max-height:400px;"></video>')
|
||||||
|
elif story_media_obj is not None:
|
||||||
|
content_media.append(f'<img src="{url}" style="max-width:100%; width:auto; height:auto;'
|
||||||
|
f'max-height:400px; object-fit:contain;">')
|
||||||
|
elif message.media == MessageMediaType.POLL and poll_media_obj is not None:
|
||||||
|
if poll_media_kind == 'video':
|
||||||
|
content_media.append(f'<video controls src="{url}" style="max-width:100%; width:auto;'
|
||||||
|
f'height:auto; max-height:400px;"></video>')
|
||||||
|
else:
|
||||||
|
content_media.append(f'<img src="{url}" style="max-width:100%; width:auto; height:auto;'
|
||||||
|
f'max-height:400px; object-fit:contain;">')
|
||||||
content_media.append('</div>')
|
content_media.append('</div>')
|
||||||
|
|
||||||
if webpage := getattr(message, "web_page", None): # Web page preview
|
if webpage := getattr(message, "web_page", None): # Web page preview
|
||||||
@@ -999,6 +1152,109 @@ class PostParser:
|
|||||||
logger.error(f"poll_parsing_error: {str(e)}")
|
logger.error(f"poll_parsing_error: {str(e)}")
|
||||||
return '<div class="message-poll">[Error displaying poll]</div>'
|
return '<div class="message-poll">[Error displaying poll]</div>'
|
||||||
|
|
||||||
|
@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'<a href="{osm_url}">{latitude:.5f}, {longitude:.5f}</a>'
|
||||||
|
|
||||||
|
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"<br>{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"<br>{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 = '<br>'.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'<div class="message-special">{block}</div>'
|
||||||
|
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]:
|
def _get_file_unique_id(self, message: Message) -> Union[str, None]:
|
||||||
try:
|
try:
|
||||||
media_mapping = {
|
media_mapping = {
|
||||||
@@ -1010,7 +1266,12 @@ class PostParser:
|
|||||||
MessageMediaType.VIDEO_NOTE: lambda m: m.video_note.file_unique_id,
|
MessageMediaType.VIDEO_NOTE: lambda m: m.video_note.file_unique_id,
|
||||||
MessageMediaType.ANIMATION: lambda m: m.animation.file_unique_id,
|
MessageMediaType.ANIMATION: lambda m: m.animation.file_unique_id,
|
||||||
MessageMediaType.STICKER: lambda m: m.sticker.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
|
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:
|
if message.media in media_mapping:
|
||||||
@@ -1059,6 +1320,7 @@ class PostParser:
|
|||||||
return
|
return
|
||||||
|
|
||||||
file_unique_id = ''
|
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
|
if message.photo: file_unique_id = message.photo.file_unique_id
|
||||||
elif message.video: file_unique_id = message.video.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.document: file_unique_id = message.document.file_unique_id
|
||||||
@@ -1069,6 +1331,24 @@ class PostParser:
|
|||||||
elif message.sticker: file_unique_id = message.sticker.file_unique_id
|
elif message.sticker: file_unique_id = message.sticker.file_unique_id
|
||||||
elif message.web_page and message.web_page.photo:
|
elif message.web_page and message.web_page.photo:
|
||||||
file_unique_id = message.web_page.photo.file_unique_id
|
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:
|
if file_unique_id:
|
||||||
added_ts = datetime.now().timestamp()
|
added_ts = datetime.now().timestamp()
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ fastapi==0.115.8
|
|||||||
starlette==0.45.3
|
starlette==0.45.3
|
||||||
uvicorn==0.34.0
|
uvicorn==0.34.0
|
||||||
python-multipart==0.0.20
|
python-multipart==0.0.20
|
||||||
Kurigram==2.2.22
|
Kurigram==2.2.23
|
||||||
#git+https://github.com/KurimuzonAkuma/pyrogram.git@dev#egg=Kurigram
|
#git+https://github.com/KurimuzonAkuma/pyrogram.git@dev#egg=Kurigram
|
||||||
TgCrypto
|
TgCrypto
|
||||||
uvloop
|
uvloop
|
||||||
|
|||||||
@@ -0,0 +1,513 @@
|
|||||||
|
# flake8: noqa
|
||||||
|
# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
|
||||||
|
# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long
|
||||||
|
# pylance: disable=reportMissingImports, reportMissingModuleSource
|
||||||
|
"""
|
||||||
|
Kurigram 2.2.23 — new/uncovered media types (LIVE_PHOTO, STORY, poll media,
|
||||||
|
GIVEAWAY, GIVEAWAY_WINNERS, PAID_MEDIA, CHECKLIST, CONTACT, LOCATION, VENUE,
|
||||||
|
DICE, GAME, INVOICE, UNSUPPORTED).
|
||||||
|
|
||||||
|
Covers:
|
||||||
|
- titles for every new media type (_media_message_title via _generate_title);
|
||||||
|
- HTML rendering: live photo <video>, story <video>/<img>, poll description_media
|
||||||
|
<img>, paid media info block (no download);
|
||||||
|
- info blocks for non-downloadable types (_format_special_media) incl. XSS escaping;
|
||||||
|
- flags: no_image semantics for the new types and polls with/without media,
|
||||||
|
"video" flag for live photos;
|
||||||
|
- api_server.find_file_id_in_message: live_photo, story, poll description_media /
|
||||||
|
explanation_media lookups.
|
||||||
|
"""
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from pyrogram.enums import MessageMediaType
|
||||||
|
|
||||||
|
from post_parser import PostParser
|
||||||
|
from api_server import find_file_id_in_message
|
||||||
|
from url_signer import KeyManager, generate_media_digest
|
||||||
|
|
||||||
|
|
||||||
|
class _Str(str):
|
||||||
|
"""Minimal stand-in for Pyrogram's Str: .html returns the raw string unchanged."""
|
||||||
|
@property
|
||||||
|
def html(self):
|
||||||
|
return str(self)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _pinned_signing_key(monkeypatch):
|
||||||
|
# generate_media_digest reads/creates data/media_digest.key relative to cwd; pin
|
||||||
|
# the in-memory key so digests are deterministic and no file IO happens
|
||||||
|
# regardless of the invocation directory (repo root or tests/).
|
||||||
|
monkeypatch.setattr(KeyManager, "signing_key", "test-signing-key-new-media")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def parser():
|
||||||
|
return PostParser(SimpleNamespace())
|
||||||
|
|
||||||
|
|
||||||
|
def make_message(mid=1, media=None, text=None, username="testchan", **extra):
|
||||||
|
m = SimpleNamespace()
|
||||||
|
m.id = mid
|
||||||
|
m.date = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
m.text = _Str(text) if text is not None else None
|
||||||
|
m.caption = None
|
||||||
|
m.media = media
|
||||||
|
m.web_page = None
|
||||||
|
m.poll = None
|
||||||
|
m.service = None
|
||||||
|
m.forward_origin = None
|
||||||
|
m.reply_to_message = None
|
||||||
|
m.sender_chat = None
|
||||||
|
m.from_user = None
|
||||||
|
m.reactions = None
|
||||||
|
m.views = 100
|
||||||
|
m.media_group_id = None
|
||||||
|
m.show_caption_above_media = False
|
||||||
|
m.chat = SimpleNamespace(id=-1001234567890, username=username)
|
||||||
|
for attr in ("photo", "video", "document", "audio", "voice",
|
||||||
|
"video_note", "animation", "sticker"):
|
||||||
|
setattr(m, attr, None)
|
||||||
|
# New Kurigram 2.2.23 attributes (live_photo, story, giveaway, checklist, ...)
|
||||||
|
# are deliberately NOT set by default: production code must survive their
|
||||||
|
# absence via getattr (that is exactly what old mocks look like).
|
||||||
|
for key, value in extra.items():
|
||||||
|
setattr(m, key, value)
|
||||||
|
return m
|
||||||
|
|
||||||
|
|
||||||
|
def media_url(mid, fuid, username="testchan"):
|
||||||
|
file = f"{username}/{mid}/{fuid}"
|
||||||
|
return f"http://test.example.com/media/{file}/{generate_media_digest(file)}"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1. Titles for every new media type
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_title_live_photo(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.LIVE_PHOTO)) == "📸 Live Photo"
|
||||||
|
|
||||||
|
def test_title_story(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.STORY)) == "📖 Story"
|
||||||
|
|
||||||
|
def test_title_giveaway(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.GIVEAWAY)) == "🎁 Giveaway"
|
||||||
|
|
||||||
|
def test_title_giveaway_winners(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.GIVEAWAY_WINNERS)) == "🏆 Giveaway winners"
|
||||||
|
|
||||||
|
def test_title_paid_media(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.PAID_MEDIA)) == "⭐ Paid media"
|
||||||
|
|
||||||
|
def test_title_checklist_with_title(parser):
|
||||||
|
msg = make_message(media=MessageMediaType.CHECKLIST,
|
||||||
|
checklist=SimpleNamespace(title="Shopping list", tasks=[]))
|
||||||
|
assert parser._generate_title(msg) == "📝 Checklist: Shopping list"
|
||||||
|
|
||||||
|
def test_title_checklist_long_title_truncated(parser):
|
||||||
|
long_title = "x" * 80
|
||||||
|
msg = make_message(media=MessageMediaType.CHECKLIST,
|
||||||
|
checklist=SimpleNamespace(title=long_title, tasks=[]))
|
||||||
|
assert parser._generate_title(msg) == f"📝 Checklist: {'x' * 50}"
|
||||||
|
|
||||||
|
def test_title_checklist_without_object(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.CHECKLIST)) == "📝 Checklist"
|
||||||
|
|
||||||
|
def test_title_contact(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.CONTACT)) == "👤 Contact"
|
||||||
|
|
||||||
|
def test_title_location(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.LOCATION)) == "📍 Location"
|
||||||
|
|
||||||
|
def test_title_venue_with_title(parser):
|
||||||
|
msg = make_message(media=MessageMediaType.VENUE,
|
||||||
|
venue=SimpleNamespace(title="Blue Bottle Cafe", address="1 Main St"))
|
||||||
|
assert parser._generate_title(msg) == "📍 Blue Bottle Cafe"
|
||||||
|
|
||||||
|
def test_title_venue_fallback(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.VENUE)) == "📍 Venue"
|
||||||
|
|
||||||
|
def test_title_dice(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.DICE)) == "🎲 Dice"
|
||||||
|
|
||||||
|
def test_title_game(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.GAME)) == "🎮 Game"
|
||||||
|
|
||||||
|
def test_title_invoice(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.INVOICE)) == "🧾 Invoice"
|
||||||
|
|
||||||
|
def test_title_unsupported(parser):
|
||||||
|
assert parser._generate_title(make_message(media=MessageMediaType.UNSUPPORTED)) == "⚠️ Unsupported content"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 2. LIVE_PHOTO rendering
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_live_photo_renders_video_with_signed_url(parser):
|
||||||
|
msg = make_message(101, media=MessageMediaType.LIVE_PHOTO,
|
||||||
|
live_photo=SimpleNamespace(file_unique_id="lp_uid_1", file_id="lp_fid_1"))
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<video" in html_media
|
||||||
|
assert media_url(101, "lp_uid_1") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_live_photo_collected_for_media_ids(parser):
|
||||||
|
msg = make_message(102, media=MessageMediaType.LIVE_PHOTO,
|
||||||
|
live_photo=SimpleNamespace(file_unique_id="lp_uid_2", file_id="lp_fid_2"))
|
||||||
|
parser._generate_html_media(msg)
|
||||||
|
assert [(c, p, f) for c, p, f, _ in parser._pending_media_ids] == [("testchan", 102, "lp_uid_2")]
|
||||||
|
|
||||||
|
|
||||||
|
def test_live_photo_gets_video_flag_not_no_image(parser):
|
||||||
|
msg = make_message(103, media=MessageMediaType.LIVE_PHOTO,
|
||||||
|
live_photo=SimpleNamespace(file_unique_id="lp_uid_3", file_id="lp_fid_3"))
|
||||||
|
flags = parser._extract_flags(msg)
|
||||||
|
assert "video" in flags
|
||||||
|
assert "no_image" not in flags
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 3. STORY rendering
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_story_with_video_renders_video(parser):
|
||||||
|
msg = make_message(111, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=SimpleNamespace(file_unique_id="st_vid"),
|
||||||
|
photo=None))
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<video" in html_media
|
||||||
|
assert media_url(111, "st_vid") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_story_with_photo_renders_img(parser):
|
||||||
|
msg = make_message(112, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=None,
|
||||||
|
photo=SimpleNamespace(file_unique_id="st_pic")))
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<img" in html_media
|
||||||
|
assert "<video" not in html_media
|
||||||
|
assert media_url(112, "st_pic") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_story_video_wins_over_photo(parser):
|
||||||
|
msg = make_message(113, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=SimpleNamespace(file_unique_id="st_vid2"),
|
||||||
|
photo=SimpleNamespace(file_unique_id="st_pic2")))
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert media_url(113, "st_vid2") in html_media
|
||||||
|
assert "st_pic2" not in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_story_video_without_uid_falls_back_to_photo_as_img(parser):
|
||||||
|
# Review fix 3: the tag choice must follow the SAME object selection as the URL.
|
||||||
|
# A story video with an unusable file_unique_id is skipped by the helper in
|
||||||
|
# favour of the photo — the render must emit <img>, not <video>.
|
||||||
|
msg = make_message(114, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=SimpleNamespace(file_unique_id=None),
|
||||||
|
photo=SimpleNamespace(file_unique_id="st_pic3")))
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<img" in html_media
|
||||||
|
assert "<video" not in html_media
|
||||||
|
assert media_url(114, "st_pic3") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_story_large_video_not_collected_for_media_ids(parser):
|
||||||
|
# Review fix 2: the >100MB "don't cache" rule applies to story videos too.
|
||||||
|
msg = make_message(115, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=SimpleNamespace(file_unique_id="st_big",
|
||||||
|
file_size=200 * 1024 * 1024),
|
||||||
|
photo=None))
|
||||||
|
parser._save_media_file_ids(msg)
|
||||||
|
assert parser._pending_media_ids == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_story_small_video_collected_for_media_ids(parser):
|
||||||
|
msg = make_message(116, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(video=SimpleNamespace(file_unique_id="st_small",
|
||||||
|
file_size=5 * 1024 * 1024),
|
||||||
|
photo=None))
|
||||||
|
parser._save_media_file_ids(msg)
|
||||||
|
assert [(c, p, f) for c, p, f, _ in parser._pending_media_ids] == [("testchan", 116, "st_small")]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 4. POLL with/without description_media
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _poll_with_photo(fuid="poll_pic", fid="poll_pic_fid"):
|
||||||
|
return SimpleNamespace(
|
||||||
|
question=_Str("Pick one?"),
|
||||||
|
options=[],
|
||||||
|
description_media=SimpleNamespace(photo=SimpleNamespace(file_unique_id=fuid, file_id=fid)),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_poll_with_description_photo_renders_img(parser):
|
||||||
|
msg = make_message(121, media=MessageMediaType.POLL, poll=_poll_with_photo())
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<img" in html_media
|
||||||
|
assert media_url(121, "poll_pic") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
def test_poll_with_description_photo_has_no_no_image_flag(parser):
|
||||||
|
msg = make_message(122, media=MessageMediaType.POLL, poll=_poll_with_photo())
|
||||||
|
flags = parser._extract_flags(msg)
|
||||||
|
assert "no_image" not in flags
|
||||||
|
assert "poll" in flags
|
||||||
|
|
||||||
|
|
||||||
|
def test_poll_without_media_keeps_no_image_flag(parser):
|
||||||
|
msg = make_message(123, media=MessageMediaType.POLL,
|
||||||
|
poll=SimpleNamespace(question=_Str("Plain poll?"), options=[]))
|
||||||
|
flags = parser._extract_flags(msg)
|
||||||
|
assert "no_image" in flags
|
||||||
|
assert "poll" in flags
|
||||||
|
|
||||||
|
|
||||||
|
def test_poll_media_collected_for_media_ids(parser):
|
||||||
|
msg = make_message(124, media=MessageMediaType.POLL, poll=_poll_with_photo("poll_pic4"))
|
||||||
|
parser._generate_html_media(msg)
|
||||||
|
assert [(c, p, f) for c, p, f, _ in parser._pending_media_ids] == [("testchan", 124, "poll_pic4")]
|
||||||
|
|
||||||
|
|
||||||
|
def test_poll_with_video_description_renders_video(parser):
|
||||||
|
poll = SimpleNamespace(
|
||||||
|
question=_Str("Video poll?"),
|
||||||
|
options=[],
|
||||||
|
description_media=SimpleNamespace(video=SimpleNamespace(file_unique_id="poll_vid", file_id="poll_vid_fid")),
|
||||||
|
)
|
||||||
|
msg = make_message(125, media=MessageMediaType.POLL, poll=poll)
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "<video" in html_media
|
||||||
|
assert media_url(125, "poll_vid") in html_media
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 5. api_server.find_file_id_in_message (async)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
async def test_find_file_id_live_photo():
|
||||||
|
msg = make_message(201, media=MessageMediaType.LIVE_PHOTO,
|
||||||
|
live_photo=SimpleNamespace(file_unique_id="lp_uid", file_id="lp_fid"))
|
||||||
|
assert await find_file_id_in_message(msg, "lp_uid") == "lp_fid"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_story_video():
|
||||||
|
msg = make_message(202, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(photo=None,
|
||||||
|
video=SimpleNamespace(file_unique_id="sv_uid", file_id="sv_fid")))
|
||||||
|
assert await find_file_id_in_message(msg, "sv_uid") == "sv_fid"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_story_photo():
|
||||||
|
msg = make_message(203, media=MessageMediaType.STORY,
|
||||||
|
story=SimpleNamespace(photo=SimpleNamespace(file_unique_id="sp_uid", file_id="sp_fid"),
|
||||||
|
video=None))
|
||||||
|
assert await find_file_id_in_message(msg, "sp_uid") == "sp_fid"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_poll_description_photo():
|
||||||
|
msg = make_message(204, media=MessageMediaType.POLL, poll=_poll_with_photo("pd_uid", "pd_fid"))
|
||||||
|
assert await find_file_id_in_message(msg, "pd_uid") == "pd_fid"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_poll_explanation_video():
|
||||||
|
poll = SimpleNamespace(
|
||||||
|
question=_Str("Quiz?"),
|
||||||
|
options=[],
|
||||||
|
explanation_media=SimpleNamespace(video=SimpleNamespace(file_unique_id="ex_uid", file_id="ex_fid")),
|
||||||
|
)
|
||||||
|
msg = make_message(205, media=MessageMediaType.POLL, poll=poll)
|
||||||
|
assert await find_file_id_in_message(msg, "ex_uid") == "ex_fid"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_poll_without_media_returns_none():
|
||||||
|
msg = make_message(206, media=MessageMediaType.POLL,
|
||||||
|
poll=SimpleNamespace(question=_Str("Plain?"), options=[]))
|
||||||
|
assert await find_file_id_in_message(msg, "whatever") is None
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_file_id_poll_none_object_returns_none():
|
||||||
|
msg = make_message(207, media=MessageMediaType.POLL) # message.poll stays None
|
||||||
|
assert await find_file_id_in_message(msg, "whatever") is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 6. XSS: user-controlled strings in info blocks are escaped
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
XSS = "<script>alert(1)</script>"
|
||||||
|
XSS_ESCAPED = "<script>alert(1)</script>"
|
||||||
|
|
||||||
|
|
||||||
|
def test_contact_name_is_escaped(parser):
|
||||||
|
msg = make_message(301, media=MessageMediaType.CONTACT,
|
||||||
|
contact=SimpleNamespace(first_name=XSS, last_name="Doe",
|
||||||
|
phone_number="+1234567890", user_id=None, vcard=None))
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert XSS not in body
|
||||||
|
assert XSS_ESCAPED in body
|
||||||
|
assert "+1234567890" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_checklist_task_text_is_escaped(parser):
|
||||||
|
checklist = SimpleNamespace(
|
||||||
|
title="My list",
|
||||||
|
tasks=[
|
||||||
|
SimpleNamespace(id=1, text=XSS, completed_by=None, completion_date=None),
|
||||||
|
SimpleNamespace(id=2, text="done task", completed_by=SimpleNamespace(id=7), completion_date=None),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
msg = make_message(302, media=MessageMediaType.CHECKLIST, checklist=checklist)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert XSS not in body
|
||||||
|
assert f"☐ {XSS_ESCAPED}" in body
|
||||||
|
assert "☑ done task" in body
|
||||||
|
assert "📝 My list" in body
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 7. Giveaway / giveaway winners info blocks
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_giveaway_block_quantity_and_months(parser):
|
||||||
|
giveaway = SimpleNamespace(quantity=10, months=3, stars=None,
|
||||||
|
until_date=datetime(2026, 2, 1, tzinfo=timezone.utc),
|
||||||
|
description="Win big!")
|
||||||
|
msg = make_message(311, media=MessageMediaType.GIVEAWAY, giveaway=giveaway)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🎁 Giveaway: 10 prize(s)" in body
|
||||||
|
assert "3 months Premium" in body
|
||||||
|
assert "until 01/02/2026" in body
|
||||||
|
assert "Win big!" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_giveaway_block_with_stars(parser):
|
||||||
|
giveaway = SimpleNamespace(quantity=5, months=None, stars=500,
|
||||||
|
until_date=None, description=None)
|
||||||
|
msg = make_message(312, media=MessageMediaType.GIVEAWAY, giveaway=giveaway)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🎁 Giveaway: 5 prize(s)" in body
|
||||||
|
assert "500 Stars" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_giveaway_winners_block(parser):
|
||||||
|
winners = SimpleNamespace(winner_count=2, quantity=5, prize_description="Cool prize",
|
||||||
|
unclaimed_prize_count=3, winners=[])
|
||||||
|
msg = make_message(313, media=MessageMediaType.GIVEAWAY_WINNERS, giveaway_winners=winners)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🏆 Giveaway winners: 2 of 5" in body
|
||||||
|
assert "Cool prize" in body
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Other info blocks and paid media
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_paid_media_renders_info_block_without_download(parser):
|
||||||
|
paid = SimpleNamespace(stars_amount=50,
|
||||||
|
media=[SimpleNamespace(width=100, height=100, duration=None, thumbnail=None),
|
||||||
|
SimpleNamespace(width=200, height=200, duration=5, thumbnail=None)])
|
||||||
|
msg = make_message(321, media=MessageMediaType.PAID_MEDIA, paid_media=paid)
|
||||||
|
html_media = parser._generate_html_media(msg)
|
||||||
|
assert "⭐ Paid media (50 stars, 2 item(s)) — available in Telegram" in html_media
|
||||||
|
assert "/media/" not in html_media # nothing downloadable
|
||||||
|
assert parser._pending_media_ids == [] # nothing collected for the download cache
|
||||||
|
assert "no_image" in parser._extract_flags(msg)
|
||||||
|
|
||||||
|
|
||||||
|
def test_location_block_has_osm_link(parser):
|
||||||
|
msg = make_message(322, media=MessageMediaType.LOCATION,
|
||||||
|
location=SimpleNamespace(latitude=55.75123456, longitude=37.61761111))
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "📍 Location:" in body
|
||||||
|
assert "https://www.openstreetmap.org/?mlat=55.75123456&mlon=37.61761111" in body
|
||||||
|
assert "55.75123, 37.61761" in body
|
||||||
|
assert "no_image" in parser._extract_flags(msg)
|
||||||
|
|
||||||
|
|
||||||
|
def test_venue_block_with_osm_link(parser):
|
||||||
|
venue = SimpleNamespace(title="Blue Bottle", address="1 Main St",
|
||||||
|
location=SimpleNamespace(latitude=1.5, longitude=2.5))
|
||||||
|
msg = make_message(323, media=MessageMediaType.VENUE, venue=venue)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "📍 Blue Bottle, 1 Main St" in body
|
||||||
|
assert "openstreetmap.org/?mlat=1.5&mlon=2.5" in body
|
||||||
|
|
||||||
|
|
||||||
|
# Review fix 1: info-block-only media types must not open an empty
|
||||||
|
# <div class="message-media"> container — only the special block is rendered.
|
||||||
|
|
||||||
|
def test_venue_has_no_empty_media_div(parser):
|
||||||
|
venue = SimpleNamespace(title="Cafe", address="2 Side St",
|
||||||
|
location=SimpleNamespace(latitude=1.0, longitude=2.0))
|
||||||
|
msg = make_message(328, media=MessageMediaType.VENUE, venue=venue)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert 'class="message-media"' not in body
|
||||||
|
assert 'class="message-special"' in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_contact_has_no_empty_media_div(parser):
|
||||||
|
msg = make_message(329, media=MessageMediaType.CONTACT,
|
||||||
|
contact=SimpleNamespace(first_name="Jane", last_name="Doe",
|
||||||
|
phone_number="+1999", user_id=None, vcard=None))
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert 'class="message-media"' not in body
|
||||||
|
assert 'class="message-special"' in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_paid_media_block_survives_media_div_gate(parser):
|
||||||
|
# PAID_MEDIA is in NO_IMAGE_MEDIA_TYPES but has its own render branch — the
|
||||||
|
# info block (inside its message-media container) must keep rendering.
|
||||||
|
paid = SimpleNamespace(stars_amount=10, media=[SimpleNamespace(width=1, height=1,
|
||||||
|
duration=None, thumbnail=None)])
|
||||||
|
msg = make_message(330, media=MessageMediaType.PAID_MEDIA, paid_media=paid)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert 'class="message-media"' in body
|
||||||
|
assert "⭐ Paid media (10 stars, 1 item(s)) — available in Telegram" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_dice_block(parser):
|
||||||
|
msg = make_message(324, media=MessageMediaType.DICE,
|
||||||
|
dice=SimpleNamespace(emoji="🎯", value=6))
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🎲 🎯: 6" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_game_block_with_title(parser):
|
||||||
|
msg = make_message(325, media=MessageMediaType.GAME,
|
||||||
|
game=SimpleNamespace(title="Tetris"))
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🎮 Game: Tetris" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoice_block(parser):
|
||||||
|
msg = make_message(326, media=MessageMediaType.INVOICE)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "🧾 Invoice" in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_unsupported_block(parser):
|
||||||
|
msg = make_message(327, media=MessageMediaType.UNSUPPORTED)
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "⚠️ This post contains content not supported by the bridge" in body
|
||||||
|
assert "no_image" in parser._extract_flags(msg)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Regression: messages without any of the new attributes still render
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_plain_text_message_unaffected(parser):
|
||||||
|
msg = make_message(331, text="just a plain post with enough text")
|
||||||
|
body = parser._generate_html_body(msg)
|
||||||
|
assert "just a plain post with enough text" in body
|
||||||
|
assert "message-special" not in body
|
||||||
|
assert "paid-media" not in body
|
||||||
|
flags = parser._extract_flags(msg)
|
||||||
|
assert "no_image" in flags
|
||||||
Reference in New Issue
Block a user