feat: upgrade to Kurigram 2.2.23 and render new Telegram media types
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:
vvzvlad
2026-07-05 17:58:34 +03:00
parent 4fc8ebbd03
commit f9550d8330
4 changed files with 833 additions and 9 deletions
+34 -3
View File
@@ -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
View File
@@ -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
View File
@@ -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
+513
View File
@@ -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 = "&lt;script&gt;alert(1)&lt;/script&gt;"
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