vvzvlad/pyrogram-bridge
1
0
Fork 0
Code Issues 6 Pull Requests Actions 8 Packages Projects Releases Wiki Activity

refactor(render): этап 5a — таблица медиа-типов (байт-в-байт) (#32)

Browse Source 
Три параллельные ручно-синхронизируемые «лестницы» выбора media-объекта
(_get_file_unique_id dict-map, _save_media_file_ids if/elif, _generate_html_media
elif-каскад) сведены в одну точку правды: MEDIA_SOURCES (селектор → (media_obj,
render_kind), 12 записей) + RenderCtx + RENDERERS (7 рендереров подняты дословно
из старых веток). Все три консьюмера теперь ходят через таблицу; uid единообразно
через getattr(file_unique_id).

Байт-в-байт: fragment-oracle (media_fragment_replay + media_fragments.json, 24
кейса) без изменений, feed-golden без изменений. Единственная унификация —
_save_media_file_ids теперь ключуется на message.media (как две другие лестницы),
что заодно чинит латентную рассинхронизацию (в корпусе не встречается → оракул 0).
test_stage5_media_table.py (48 тестов). Полный сюит 413 passed, PYTHONHASHSEED 1/42.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
agent_coder
2026-07-06 18:25:45 +03:00
parent bf7b5d9442
commit 78e9bc2583
4 changed files with 813 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
post_parser.py
+181 -116
View File
@@ -14,8 +14,9 @@ import re
import os
import html
import inspect
from dataclasses import dataclass
from datetime import datetime
from typing import Union, Dict, Any, List, Optional
from typing import Union, Dict, Any, List, Optional, Callable, Tuple
from pyrogram.types import Message
from pyrogram.enums import MessageMediaType
from sanitizer import sanitize_html
@@ -107,6 +108,138 @@ def _story_media_object(message):
return None, None
# --------------------------------------------------------------------------- #
# Single source of truth for media handling (render-pipeline refactor, stage 5).
#
# MEDIA_SOURCES maps a media type to a selector returning (media object, render
# kind). It replaces the three hand-synced ladders that used to live in
# _get_file_unique_id (dict map), _save_media_file_ids (if/elif) and
# _generate_html_media (elif cascade). All three now go through this table.
#
# kind=None means "selector-only entry": the object participates in file-id
# extraction/collection but is NOT rendered by the dispatcher. The ONLY kind=None
# entry is WEB_PAGE (its preview is rendered by _format_webpage). PAID_MEDIA has
# NO entry at all: its info block is a dedicated branch in _generate_html_media
# and it is deliberately never collected/downloaded (paid content).
# --------------------------------------------------------------------------- #
def _select_document(message):
"""DOCUMENT selector: PDF documents render as a t.me link ('pdf'), everything
else as an image ('img_400'). The selected object is always message.document."""
document = message.document
if (document is not None and hasattr(document, 'mime_type')
and document.mime_type == 'application/pdf'):
return document, 'pdf'
return document, 'img_400'
def _select_sticker(message):
"""STICKER selector: video stickers loop as <video> ('video_loop_200'), image
stickers render as <img> ('img_200_sticker')."""
sticker = message.sticker
if getattr(sticker, 'is_video', False):
return sticker, 'video_loop_200'
return sticker, 'img_200_sticker'
# Helper kinds returned by _story_media_object / _poll_media_object map to render kinds.
_HELPER_KIND_TO_RENDER = {'video': 'video_400', 'img': 'img_400'}
def _select_story(message):
"""STORY selector: object choice (video over photo) is delegated to
_story_media_object; its helper kind is mapped to a render kind."""
media_obj, helper_kind = _story_media_object(message)
return media_obj, _HELPER_KIND_TO_RENDER.get(helper_kind)
def _select_poll_media(message):
"""POLL selector: object comes from _poll_media_object (description_media); its
helper kind is mapped to a render kind."""
media_obj, helper_kind = _poll_media_object(message)
return media_obj, _HELPER_KIND_TO_RENDER.get(helper_kind)
MEDIA_SOURCES: Dict[Any, Callable[[Message], Tuple[Any, Optional[str]]]] = {
MessageMediaType.PHOTO: lambda m: (m.photo, 'img_400'),
MessageMediaType.VIDEO: lambda m: (m.video, 'video_400'),
MessageMediaType.ANIMATION: lambda m: (m.animation, 'video_400'),
MessageMediaType.VIDEO_NOTE: lambda m: (m.video_note, 'video_400'),
MessageMediaType.AUDIO: lambda m: (m.audio, 'audio'),
MessageMediaType.VOICE: lambda m: (m.voice, 'audio'),
MessageMediaType.DOCUMENT: _select_document,
MessageMediaType.STICKER: _select_sticker,
MessageMediaType.LIVE_PHOTO: lambda m: (getattr(m, 'live_photo', None), 'video_loop_400'),
MessageMediaType.STORY: _select_story,
MessageMediaType.POLL: _select_poll_media,
MessageMediaType.WEB_PAGE: lambda m: (getattr(m.web_page, 'photo', None), None),
}
@dataclass
class RenderCtx:
"""Everything a renderer needs. _generate_html_media assembles it (URL signing,
channel_username guard, and the mime/emoji/tg_link values chosen BY MEDIA TYPE);
a renderer only formats the string and never inspects the Message."""
url: str
tg_link: Optional[str] = None # t.me deep link — only the 'pdf' renderer uses it
emoji: str = '' # sticker alt text
mime: Optional[str] = None # audio/voice <source> type; default chosen by media type
# Renderers return list[str] so the byte structure of the '\n'.join in
# _generate_html_media is preserved (audio emits TWO items: <audio> tag + <br>;
# pdf emits its two-append div block). Bodies are lifted VERBATIM from the old
# if/elif branches, including the `src="{url}"style=` concatenation artifacts —
# byte-for-byte fidelity is the stage-5a contract.
def _render_img_400(ctx: 'RenderCtx') -> List[str]:
return [f'<img src="{ctx.url}" style="max-width:100%; width:auto; height:auto;'
f'max-height:400px; object-fit:contain;">']
def _render_video_400(ctx: 'RenderCtx') -> List[str]:
return [f'<video controls src="{ctx.url}" style="max-width:100%; width:auto;'
f'height:auto; max-height:400px;"></video>']
def _render_audio(ctx: 'RenderCtx') -> List[str]:
return [f'<audio controls style="width:100%; max-width:400px;">'
f'<source src="{ctx.url}" type="{ctx.mime}"></audio>',
'<br>']
def _render_video_loop_200(ctx: 'RenderCtx') -> List[str]:
return [f'<video controls autoplay loop muted src="{ctx.url}"'
f'style="max-width:100%; width:auto; height:auto; max-height:200px;'
f'object-fit:contain;"></video>']
def _render_img_200_sticker(ctx: 'RenderCtx') -> List[str]:
return [f'<img src="{ctx.url}" alt="Sticker {ctx.emoji}" style="max-width:100%;'
f'width:auto; height:auto; max-height:200px; object-fit:contain;">']
def _render_video_loop_400(ctx: 'RenderCtx') -> List[str]:
return [f'<video controls autoplay loop muted src="{ctx.url}"'
f'style="max-width:100%; width:auto; height:auto; max-height:400px;'
f'object-fit:contain;"></video>']
def _render_pdf(ctx: 'RenderCtx') -> List[str]:
return [f'<div class="document-pdf" style="padding: 10px;">',
f'<a href="{ctx.tg_link}" target="_blank">[PDF-файл]</a></div>']
RENDERERS: Dict[str, Callable[['RenderCtx'], List[str]]] = {
'img_400': _render_img_400,
'video_400': _render_video_400,
'audio': _render_audio,
'video_loop_200': _render_video_loop_200,
'img_200_sticker': _render_img_200_sticker,
'video_loop_400': _render_video_loop_400,
'pdf': _render_pdf,
}
#tests
#http://127.0.0.1:8000/post/html/DragorWW_space/114 — video
#http://127.0.0.1:8000/post/html/DragorWW_space/20 - many photos
@@ -848,72 +981,31 @@ class PostParser:
logger.debug(f"Collected media file: {channel_username}/{message.id}/{file_unique_id}")
# Check if document is a PDF file
if (message.media == MessageMediaType.DOCUMENT and
message.document is not None and hasattr(message.document, 'mime_type') and
message.document.mime_type == 'application/pdf'):
# Only attempt to create link if channel_username is available
if channel_username.startswith('-100'):
tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}"
else:
tg_link = f"https://t.me/{channel_username}/{message.id}"
content_media.append(f'<div class="document-pdf" style="padding: 10px;">')
content_media.append(f'<a href="{tg_link}" target="_blank">[PDF-файл]</a></div>')
elif message.media in [MessageMediaType.PHOTO, MessageMediaType.DOCUMENT]:
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.VIDEO:
content_media.append(f'<video controls src="{url}" style="max-width:100%; width:auto;'
f'height:auto; max-height:400px;"></video>')
elif message.media == MessageMediaType.ANIMATION:
content_media.append(f'<video controls src="{url}" style="max-width:100%; width:auto;'
f'height:auto; max-height:400px;"></video>')
elif message.media == MessageMediaType.VIDEO_NOTE:
content_media.append(f'<video controls src="{url}" style="max-width:100%; width:auto;'
f'height:auto; max-height:400px;"></video>')
elif message.media == MessageMediaType.AUDIO:
mime_type = getattr(message.audio, 'mime_type', 'audio/mpeg')
content_media.append(f'<audio controls style="width:100%; max-width:400px;">'
f'<source src="{url}" type="{mime_type}"></audio>')
content_media.append('<br>')
elif message.media == MessageMediaType.VOICE:
mime_type = getattr(message.voice, 'mime_type', 'audio/ogg')
content_media.append(f'<audio controls style="width:100%; max-width:400px;">'
f'<source src="{url}" type="{mime_type}"></audio>')
content_media.append('<br>')
elif message.media == MessageMediaType.STICKER:
emoji = getattr(message.sticker, 'emoji', '')
if getattr(message.sticker, 'is_video', False):
content_media.append(f'<video controls autoplay loop muted src="{url}"'
f'style="max-width:100%; width:auto; height:auto; max-height:200px;'
f'object-fit:contain;"></video>')
else:
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;">')
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;">')
# Dispatch through the media table: the selector gives the render
# kind, the RENDERERS map gives the fragment. mime/emoji/tg_link
# defaults are chosen here BY MEDIA TYPE (the renderer never guesses);
# kind None (WEB_PAGE) emits no element — the empty container matches
# the old cascade where no branch matched.
selector = MEDIA_SOURCES.get(message.media)
kind = selector(message)[1] if selector is not None else None
renderer = RENDERERS.get(kind) if kind else None
if renderer is not None:
ctx = RenderCtx(url=url)
if kind == 'pdf':
if channel_username.startswith('-100'):
ctx.tg_link = f"https://t.me/c/{channel_username[4:]}/{message.id}"
else:
ctx.tg_link = f"https://t.me/{channel_username}/{message.id}"
elif kind == 'audio':
if message.media == MessageMediaType.VOICE:
ctx.mime = getattr(message.voice, 'mime_type', 'audio/ogg')
else:
ctx.mime = getattr(message.audio, 'mime_type', 'audio/mpeg')
elif kind == 'img_200_sticker':
ctx.emoji = getattr(message.sticker, 'emoji', '')
content_media.extend(renderer(ctx))
content_media.append('</div>')
if webpage := getattr(message, "web_page", None): # Web page preview
if webpage_html := self._format_webpage(webpage, message):
if len((message.text or message.caption or '').strip()) <= 10:
@@ -1231,28 +1323,12 @@ class PostParser:
def _get_file_unique_id(self, message: Message) -> Union[str, None]:
try:
media_mapping = {
MessageMediaType.PHOTO: lambda m: m.photo.file_unique_id,
MessageMediaType.VIDEO: lambda m: m.video.file_unique_id,
MessageMediaType.DOCUMENT: lambda m: m.document.file_unique_id,
MessageMediaType.AUDIO: lambda m: m.audio.file_unique_id,
MessageMediaType.VOICE: lambda m: m.voice.file_unique_id,
MessageMediaType.VIDEO_NOTE: lambda m: m.video_note.file_unique_id,
MessageMediaType.ANIMATION: lambda m: m.animation.file_unique_id,
MessageMediaType.STICKER: lambda m: m.sticker.file_unique_id,
MessageMediaType.WEB_PAGE: lambda m: m.web_page.photo.file_unique_id if m.web_page and m.web_page.photo else None,
# New media types (Kurigram 2.2.23): getattr-only access, the
# attributes do not exist on older Message objects/mocks.
MessageMediaType.LIVE_PHOTO: lambda m: getattr(getattr(m, 'live_photo', None), 'file_unique_id', None),
MessageMediaType.STORY: lambda m: getattr(_story_media_object(m)[0], 'file_unique_id', None),
MessageMediaType.POLL: lambda m: getattr(_poll_media_object(m)[0], 'file_unique_id', None),
}
if message.media in media_mapping:
return media_mapping[message.media](message)
return None
selector = MEDIA_SOURCES.get(message.media)
if selector is None:
return None
selected_obj, _kind = selector(message)
return getattr(selected_obj, 'file_unique_id', None)
except Exception as e:
logger.error(f"file_id_extraction_error: media_type {message.media}, error {str(e)}")
return None
@@ -1289,40 +1365,29 @@ class PostParser:
return
if message.media:
# Skip large videos - they shouldn't be cached permanently
# Skip large videos - they shouldn't be cached permanently.
# 5a keeps this message.video-only guard verbatim; registry §3.13
# generalises the >100MB rule to every selected object in 5b.
if message.video and message.video.file_size and message.video.file_size > 100 * 1024 * 1024:
return
file_unique_id = ''
new_media_obj = None # selected object among the Kurigram 2.2.23 media sources
if message.photo: file_unique_id = message.photo.file_unique_id
elif message.video: file_unique_id = message.video.file_unique_id
elif message.document: file_unique_id = message.document.file_unique_id
elif message.audio: file_unique_id = message.audio.file_unique_id
elif message.voice: file_unique_id = message.voice.file_unique_id
elif message.video_note: file_unique_id = message.video_note.file_unique_id
elif message.animation: file_unique_id = message.animation.file_unique_id
elif message.sticker: file_unique_id = message.sticker.file_unique_id
elif message.web_page and message.web_page.photo:
file_unique_id = message.web_page.photo.file_unique_id
# New media types (Kurigram 2.2.23): getattr-only, the attributes do
# not exist on older Message objects/mocks. paid_media is deliberately
# NOT collected — it cannot be downloaded (paid content).
elif getattr(message, 'live_photo', None):
new_media_obj = message.live_photo
elif (story_media := _story_media_object(message)[0]) is not None:
new_media_obj = story_media
elif (poll_media := _poll_media_object(message)[0]) is not None:
new_media_obj = poll_media
# Object selection now goes through the single MEDIA_SOURCES table
# instead of the old truthy-attribute if/elif ladder. paid_media has no
# table entry and is deliberately NOT collected (paid content).
selector = MEDIA_SOURCES.get(message.media)
selected_obj = selector(message)[0] if selector is not None else None
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)
# The message.video guard above does not cover the Kurigram 2.2.23
# sources (live photo, story video, poll description video) — apply the
# same "don't cache large videos" rule to them (5b unifies this — §3.13).
if message.media in (MessageMediaType.LIVE_PHOTO,
MessageMediaType.STORY,
MessageMediaType.POLL):
file_size = getattr(selected_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 ''
file_unique_id = getattr(selected_obj, 'file_unique_id', '') or ''
if file_unique_id:
added_ts = datetime.now().timestamp()
tests/media_fragment_replay.py
+181
View File
@@ -0,0 +1,181 @@
# flake8: noqa
# pylint: disable=import-outside-toplevel, missing-function-docstring, line-too-long
"""Stage-5 fragment-level snapshot helpers (render-pipeline refactor epic, issue #32/#34).
Captures the raw `_generate_html_media` output (BEFORE sanitize) for every media
render kind and edge branch, as a SEPARATE oracle layer from the stage-0 feed goldens
(spec §4: "два слоя эталонов, не смешивать"). The 5a refactor (MEDIA_SOURCES table +
renderers) must reproduce these fragments byte-for-byte; 5b intentionally changes two
of them (§3.13 large-file guard is collection-only, §3.14 unclosed-div close).
Cases whose type exists in the recorded corpus could be pulled from it, but the media
FRAGMENT is a pure function of a single Message, so deterministic hand-built mocks give
the same bytes with far less machinery and also reach the mock-only exotic types
(PAID_MEDIA, LIVE_PHOTO, STORY) — exactly the set the spec allows mocks for.
Run `python -m tests.media_fragment_replay` from the repo root to (re)generate the snapshot.
"""
import os
import json
from types import SimpleNamespace
TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(TESTS_DIR)
SNAPSHOT_PATH = os.path.join(TESTS_DIR, "test_data", "media_fragments.json")
# Fixed signing key so digests in the snapshot reproduce on any checkout (mirrors the
# stage-0 golden pin). Applied by the generator and by the comparison test.
FRAGMENT_SIGNING_KEY = "stage5-fragment-fixed-signing-key-000000000000"
class _Str(str):
"""Minimal stand-in for Pyrogram's Str: .html returns the raw string unchanged."""
@property
def html(self):
return str(self)
def _msg(mid, media, *, text=None, username="testchan", chat_id=-1001234567890, **extra):
"""Build a Message-like mock with the attributes _generate_html_media touches.
Top-level media attributes default to None (truthy checks in _save_media_file_ids);
new Kurigram 2.2.23 attributes (live_photo/story/giveaway/...) are intentionally
absent unless passed, so getattr-only production code is exercised as on real objects.
"""
m = SimpleNamespace()
m.id = mid
m.media = media
m.text = _Str(text) if text is not None else None
m.caption = None
m.web_page = None
m.poll = None
m.paid_media = None
m.forward_origin = None
m.show_caption_above_media = False
if chat_id is None and username is None:
m.chat = None
else:
m.chat = SimpleNamespace(id=chat_id, username=username)
for attr in ("photo", "video", "document", "audio", "voice",
"video_note", "animation", "sticker"):
setattr(m, attr, None)
for key, value in extra.items():
setattr(m, key, value)
return m
def _obj(**kw):
return SimpleNamespace(**kw)
# --------------------------------------------------------------------------- #
# Fragment cases. Each entry: name -> Message factory.
# Covers the full spec §5a "Шаг 0" list + edge branches.
# --------------------------------------------------------------------------- #
def _webpage(photo=None, url="https://example.com", title="Example",
description=None, site_name=None, wp_type=""):
return _obj(photo=photo, url=url, title=title, description=description,
site_name=site_name, type=wp_type, display_url=None)
def build_cases():
from pyrogram.enums import MessageMediaType as T
cases = {}
cases["photo"] = lambda: _msg(1, T.PHOTO, photo=_obj(file_unique_id="ph_uid", file_id="ph_fid"))
cases["video"] = lambda: _msg(2, T.VIDEO, video=_obj(file_unique_id="vid_uid", file_id="vid_fid", file_size=1024))
cases["animation"] = lambda: _msg(3, T.ANIMATION, animation=_obj(file_unique_id="ani_uid", file_id="ani_fid"))
cases["video_note"] = lambda: _msg(4, T.VIDEO_NOTE, video_note=_obj(file_unique_id="vn_uid", file_id="vn_fid"))
cases["audio_default_mime"] = lambda: _msg(5, T.AUDIO, audio=_obj(file_unique_id="au_uid", file_id="au_fid"))
cases["audio_explicit_mime"] = lambda: _msg(6, T.AUDIO, audio=_obj(file_unique_id="au2_uid", file_id="au2_fid", mime_type="audio/flac"))
cases["voice_default_mime"] = lambda: _msg(7, T.VOICE, voice=_obj(file_unique_id="vo_uid", file_id="vo_fid"))
cases["voice_explicit_mime"] = lambda: _msg(8, T.VOICE, voice=_obj(file_unique_id="vo2_uid", file_id="vo2_fid", mime_type="audio/wav"))
cases["sticker_img"] = lambda: _msg(9, T.STICKER, sticker=_obj(file_unique_id="st_uid", file_id="st_fid", emoji="😀", is_video=False))
cases["sticker_video"] = lambda: _msg(10, T.STICKER, sticker=_obj(file_unique_id="stv_uid", file_id="stv_fid", emoji="🎬", is_video=True))
cases["document_pdf_public"] = lambda: _msg(11, T.DOCUMENT, username="pubchan", chat_id=None,
document=_obj(file_unique_id="pdf_uid", file_id="pdf_fid", mime_type="application/pdf"))
cases["document_pdf_private"] = lambda: _msg(12, T.DOCUMENT, username=None, chat_id=-1009876543210,
document=_obj(file_unique_id="pdf2_uid", file_id="pdf2_fid", mime_type="application/pdf"))
cases["document_normal"] = lambda: _msg(13, T.DOCUMENT, document=_obj(file_unique_id="doc_uid", file_id="doc_fid", mime_type="image/png"))
cases["live_photo"] = lambda: _msg(14, T.LIVE_PHOTO, live_photo=_obj(file_unique_id="lp_uid", file_id="lp_fid"))
cases["story_video"] = lambda: _msg(15, T.STORY, story=_obj(video=_obj(file_unique_id="sv_uid", file_id="sv_fid"), photo=None))
cases["story_photo"] = lambda: _msg(16, T.STORY, story=_obj(video=None, photo=_obj(file_unique_id="sp_uid", file_id="sp_fid")))
cases["poll_media_img"] = lambda: _msg(17, T.POLL, poll=_obj(question=_Str("Q?"), options=[],
description_media=_obj(photo=_obj(file_unique_id="pl_uid", file_id="pl_fid"))))
cases["poll_media_video"] = lambda: _msg(18, T.POLL, poll=_obj(question=_Str("Q?"), options=[],
description_media=_obj(video=_obj(file_unique_id="plv_uid", file_id="plv_fid"))))
cases["paid_media"] = lambda: _msg(19, T.PAID_MEDIA,
paid_media=_obj(stars_amount=50, media=[_obj(), _obj()]))
# WEB_PAGE with photo: opens an EMPTY message-media div (no elif matches WEB_PAGE),
# plus the webpage-preview block (short text gate <=10).
cases["webpage_with_photo"] = lambda: _msg(20, T.WEB_PAGE, text="hi",
web_page=_webpage(photo=_obj(file_unique_id="wp_uid", file_id="wp_fid")))
# WEB_PAGE without photo: file_unique_id is None -> message-media div opened and
# left UNCLOSED (§3.14 target). Short text so the preview block renders.
cases["webpage_without_photo"] = lambda: _msg(21, T.WEB_PAGE, text="hi",
web_page=_webpage(photo=None))
# WEB_PAGE with photo but long text (>10): preview gate closed, only the empty media div.
cases["webpage_photo_long_text"] = lambda: _msg(22, T.WEB_PAGE,
text="this text is definitely longer than ten characters",
web_page=_webpage(photo=_obj(file_unique_id="wpl_uid", file_id="wpl_fid")))
# file_unique_id is None on a normal media type -> unclosed div (§3.14 target).
cases["file_unique_id_none"] = lambda: _msg(23, T.PHOTO, photo=_obj(file_unique_id=None, file_id="x_fid"))
# channel_username is None but uid present -> the div IS closed on the guard branch.
cases["channel_username_none"] = lambda: _msg(24, T.PHOTO, username=None, chat_id=555,
photo=_obj(file_unique_id="cu_uid", file_id="cu_fid"))
return cases
# --------------------------------------------------------------------------- #
# Capture / compare.
# --------------------------------------------------------------------------- #
def capture_fragments():
"""Return {case_name: {"html": fragment, "collected": [[chan, id, fuid], ...]}}.
Covers all three ladders in one shot: _generate_html_media renders the fragment
(render ladder + _get_file_unique_id ladder) and calls _save_media_file_ids
(collection ladder), whose result is read off _pending_media_ids."""
from post_parser import PostParser
parser = PostParser(SimpleNamespace())
out = {}
for name, factory in build_cases().items():
parser._pending_media_ids = []
html = parser._generate_html_media(factory())
collected = [[c, p, f] for c, p, f, _ in parser._pending_media_ids]
out[name] = {"html": html, "collected": collected}
return out
def pin_signing_key(monkeypatch):
from url_signer import KeyManager
monkeypatch.setattr(KeyManager, "signing_key", FRAGMENT_SIGNING_KEY)
def load_snapshot():
with open(SNAPSHOT_PATH, encoding="utf-8") as f:
return json.load(f)
def _bootstrap_standalone():
import sys
import time
if ROOT not in sys.path:
sys.path.insert(0, ROOT)
import tests.mock_config as _mock_config
sys.modules["config"] = _mock_config
os.environ["TZ"] = "UTC"
time.tzset()
def generate_snapshot():
from url_signer import KeyManager
KeyManager.signing_key = FRAGMENT_SIGNING_KEY
data = capture_fragments()
os.makedirs(os.path.dirname(SNAPSHOT_PATH), exist_ok=True)
with open(SNAPSHOT_PATH, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)
print(f"wrote {len(data)} fragment snapshots to {SNAPSHOT_PATH}")
if __name__ == "__main__":
_bootstrap_standalone()
generate_snapshot()
tests/test_data/media_fragments.json
+218
View File
@@ -0,0 +1,218 @@
{
"animation": {
"collected": [
[
"testchan",
3,
"ani_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls src=\"http://test.example.com/media/testchan/3/ani_uid/ef4319c9\" style=\"max-width:100%; width:auto;height:auto; max-height:400px;\"></video>\n</div>"
},
"audio_default_mime": {
"collected": [
[
"testchan",
5,
"au_uid"
]
],
"html": "<div class=\"message-media\">\n<audio controls style=\"width:100%; max-width:400px;\"><source src=\"http://test.example.com/media/testchan/5/au_uid/be14c993\" type=\"audio/mpeg\"></audio>\n<br>\n</div>"
},
"audio_explicit_mime": {
"collected": [
[
"testchan",
6,
"au2_uid"
]
],
"html": "<div class=\"message-media\">\n<audio controls style=\"width:100%; max-width:400px;\"><source src=\"http://test.example.com/media/testchan/6/au2_uid/015e2737\" type=\"audio/flac\"></audio>\n<br>\n</div>"
},
"channel_username_none": {
"collected": [],
"html": "<div class=\"message-media\">\n</div>"
},
"document_normal": {
"collected": [
[
"testchan",
13,
"doc_uid"
]
],
"html": "<div class=\"message-media\">\n<img src=\"http://test.example.com/media/testchan/13/doc_uid/4c9075c7\" style=\"max-width:100%; width:auto; height:auto;max-height:400px; object-fit:contain;\">\n</div>"
},
"document_pdf_private": {
"collected": [
[
"-1009876543210",
12,
"pdf2_uid"
]
],
"html": "<div class=\"message-media\">\n<div class=\"document-pdf\" style=\"padding: 10px;\">\n<a href=\"https://t.me/c/9876543210/12\" target=\"_blank\">[PDF-файл]</a></div>\n</div>"
},
"document_pdf_public": {
"collected": [
[
"pubchan",
11,
"pdf_uid"
]
],
"html": "<div class=\"message-media\">\n<div class=\"document-pdf\" style=\"padding: 10px;\">\n<a href=\"https://t.me/pubchan/11\" target=\"_blank\">[PDF-файл]</a></div>\n</div>"
},
"file_unique_id_none": {
"collected": [],
"html": "<div class=\"message-media\">"
},
"live_photo": {
"collected": [
[
"testchan",
14,
"lp_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls autoplay loop muted src=\"http://test.example.com/media/testchan/14/lp_uid/7cbfcf53\"style=\"max-width:100%; width:auto; height:auto; max-height:400px;object-fit:contain;\"></video>\n</div>"
},
"paid_media": {
"collected": [],
"html": "<div class=\"message-media\">\n<div class=\"paid-media\">⭐ Paid media (50 stars, 2 item(s)) — available in Telegram</div>\n</div>"
},
"photo": {
"collected": [
[
"testchan",
1,
"ph_uid"
]
],
"html": "<div class=\"message-media\">\n<img src=\"http://test.example.com/media/testchan/1/ph_uid/08e6b2ef\" style=\"max-width:100%; width:auto; height:auto;max-height:400px; object-fit:contain;\">\n</div>"
},
"poll_media_img": {
"collected": [
[
"testchan",
17,
"pl_uid"
]
],
"html": "<div class=\"message-media\">\n<img src=\"http://test.example.com/media/testchan/17/pl_uid/492d238c\" style=\"max-width:100%; width:auto; height:auto;max-height:400px; object-fit:contain;\">\n</div>"
},
"poll_media_video": {
"collected": [
[
"testchan",
18,
"plv_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls src=\"http://test.example.com/media/testchan/18/plv_uid/abb609a3\" style=\"max-width:100%; width:auto;height:auto; max-height:400px;\"></video>\n</div>"
},
"sticker_img": {
"collected": [
[
"testchan",
9,
"st_uid"
]
],
"html": "<div class=\"message-media\">\n<img src=\"http://test.example.com/media/testchan/9/st_uid/03dc35d2\" alt=\"Sticker 😀\" style=\"max-width:100%;width:auto; height:auto; max-height:200px; object-fit:contain;\">\n</div>"
},
"sticker_video": {
"collected": [
[
"testchan",
10,
"stv_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls autoplay loop muted src=\"http://test.example.com/media/testchan/10/stv_uid/cedd7494\"style=\"max-width:100%; width:auto; height:auto; max-height:200px;object-fit:contain;\"></video>\n</div>"
},
"story_photo": {
"collected": [
[
"testchan",
16,
"sp_uid"
]
],
"html": "<div class=\"message-media\">\n<img src=\"http://test.example.com/media/testchan/16/sp_uid/d8a5d9c9\" style=\"max-width:100%; width:auto; height:auto;max-height:400px; object-fit:contain;\">\n</div>"
},
"story_video": {
"collected": [
[
"testchan",
15,
"sv_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls src=\"http://test.example.com/media/testchan/15/sv_uid/1aba6362\" style=\"max-width:100%; width:auto;height:auto; max-height:400px;\"></video>\n</div>"
},
"video": {
"collected": [
[
"testchan",
2,
"vid_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls src=\"http://test.example.com/media/testchan/2/vid_uid/e0c6ba2e\" style=\"max-width:100%; width:auto;height:auto; max-height:400px;\"></video>\n</div>"
},
"video_note": {
"collected": [
[
"testchan",
4,
"vn_uid"
]
],
"html": "<div class=\"message-media\">\n<video controls src=\"http://test.example.com/media/testchan/4/vn_uid/a5ee9944\" style=\"max-width:100%; width:auto;height:auto; max-height:400px;\"></video>\n</div>"
},
"voice_default_mime": {
"collected": [
[
"testchan",
7,
"vo_uid"
]
],
"html": "<div class=\"message-media\">\n<audio controls style=\"width:100%; max-width:400px;\"><source src=\"http://test.example.com/media/testchan/7/vo_uid/f16627e0\" type=\"audio/ogg\"></audio>\n<br>\n</div>"
},
"voice_explicit_mime": {
"collected": [
[
"testchan",
8,
"vo2_uid"
]
],
"html": "<div class=\"message-media\">\n<audio controls style=\"width:100%; max-width:400px;\"><source src=\"http://test.example.com/media/testchan/8/vo2_uid/30fed86a\" type=\"audio/wav\"></audio>\n<br>\n</div>"
},
"webpage_photo_long_text": {
"collected": [
[
"testchan",
22,
"wpl_uid"
]
],
"html": "<div class=\"message-media\">\n</div>"
},
"webpage_with_photo": {
"collected": [
[
"testchan",
20,
"wp_uid"
]
],
"html": "<div class=\"message-media\">\n</div>\n<div class=\"webpage-preview\">\n<div class=\"webpage-preview\" style=\"border-left: 3px solid #ccc; padding-left: 10px; margin: 10px 0;\">\n<div class=\"webpage-title\" style=\"font-weight:bold; margin:5px 0;\">\n<a href=\"https://example.com\" target=\"_blank\">Example</a></div>\n<div class=\"webpage-url\" style=\"color:#666; font-size:0.9em;margin-bottom:5px;\">https://example.com</div>\n<div class=\"webpage-photo\" style=\"margin-top:10px;\">\n<a href=\"https://example.com\" target=\"_blank\">\n<img src=\"http://test.example.com/media/testchan/20/wp_uid/a3dc7111\" style=\"max-width:100%; width:auto;height:auto; max-height:200px; object-fit:contain;\"></a></div>\n</div>\n</div>"
},
"webpage_without_photo": {
"collected": [],
"html": "<div class=\"message-media\">\n<div class=\"webpage-preview\">\n<div class=\"webpage-preview\" style=\"border-left: 3px solid #ccc; padding-left: 10px; margin: 10px 0;\">\n<div class=\"webpage-title\" style=\"font-weight:bold; margin:5px 0;\">\n<a href=\"https://example.com\" target=\"_blank\">Example</a></div>\n<div class=\"webpage-url\" style=\"color:#666; font-size:0.9em;margin-bottom:5px;\">https://example.com</div>\n</div>\n</div>"
}
}
tests/test_stage5_media_table.py
+233
View File
@@ -0,0 +1,233 @@
# flake8: noqa
# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
# pylint: disable=redefined-outer-name, line-too-long
"""Stage-5 media table tests (render-pipeline refactor epic, issue #32/#34).
Two layers:
* 5a — the MEDIA_SOURCES table + renderers reproduce the three old ladders
(_get_file_unique_id, _save_media_file_ids, _generate_html_media) byte-for-byte:
a fragment-level snapshot oracle (tests/test_data/media_fragments.json) plus
table/selector/invariant unit tests.
* 5b — the registered fixes (§3.13 large-file guard for every object, §3.14 the
message-media div is closed in every branch) live in test_stage5b_media_fixes.py.
The cross-module invariant test pins the boundary with api_server.find_file_id_in_message:
every object a selector returns must be resolvable there by its file_unique_id, or a new
table entry would mint a /media URL the download path answers with 404.
"""
from types import SimpleNamespace
import pytest
from pyrogram.enums import MessageMediaType
from post_parser import (
PostParser, MEDIA_SOURCES, RENDERERS, RenderCtx,
_select_document, _select_sticker, _select_story, _select_poll_media,
)
from api_server import find_file_id_in_message
from url_signer import KeyManager
from tests import media_fragment_replay as fr
@pytest.fixture(autouse=True)
def _pin_signing_key(monkeypatch):
# Deterministic media-URL digests regardless of checkout / cwd.
monkeypatch.setattr(KeyManager, "signing_key", fr.FRAGMENT_SIGNING_KEY)
@pytest.fixture
def parser():
return PostParser(SimpleNamespace())
# --------------------------------------------------------------------------- #
# 1. Fragment-level snapshot oracle — the 5a byte-for-byte contract.
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("case_name", sorted(fr.build_cases().keys()))
def test_media_fragment_matches_snapshot(parser, case_name):
"""_generate_html_media (render ladder + _get_file_unique_id ladder) and the
_save_media_file_ids collection ladder reproduce the frozen pre-refactor bytes."""
snapshot = fr.load_snapshot()
factory = fr.build_cases()[case_name]
parser._pending_media_ids = []
html = parser._generate_html_media(factory())
collected = [[c, p, f] for c, p, f, _ in parser._pending_media_ids]
assert html == snapshot[case_name]["html"], f"fragment HTML diverged for {case_name}"
assert collected == snapshot[case_name]["collected"], f"collection diverged for {case_name}"
def test_snapshot_covers_every_spec_case():
"""Guard: the snapshot and the case builders stay in lockstep (a dropped case
would otherwise silently reduce coverage)."""
snapshot = fr.load_snapshot()
assert set(snapshot.keys()) == set(fr.build_cases().keys())
# The spec §5a "Шаг 0" edge branches must all be present.
for required in ("photo", "video", "animation", "video_note", "audio_default_mime",
"voice_default_mime", "sticker_img", "sticker_video",
"document_pdf_public", "document_pdf_private", "document_normal",
"live_photo", "story_video", "story_photo", "poll_media_img",
"poll_media_video", "paid_media", "webpage_with_photo",
"webpage_without_photo", "webpage_photo_long_text",
"file_unique_id_none", "channel_username_none"):
assert required in snapshot, f"missing spec fragment case: {required}"
# --------------------------------------------------------------------------- #
# 2. Table structure: every render kind has a renderer; the only kind=None entry
# is WEB_PAGE; PAID_MEDIA has no entry.
# --------------------------------------------------------------------------- #
def test_every_render_kind_has_a_renderer():
# Fixed-kind lambda entries (branchy selectors are covered by the selector unit
# tests below and the fragment oracle).
static = {
MessageMediaType.PHOTO: 'img_400',
MessageMediaType.VIDEO: 'video_400',
MessageMediaType.ANIMATION: 'video_400',
MessageMediaType.VIDEO_NOTE: 'video_400',
MessageMediaType.AUDIO: 'audio',
MessageMediaType.VOICE: 'audio',
MessageMediaType.LIVE_PHOTO: 'video_loop_400',
}
for mt, expected_kind in static.items():
assert expected_kind in RENDERERS, f"{mt} kind {expected_kind} has no renderer"
# Selector-produced kinds (document/sticker/story/poll) all resolve to renderers
# or, for WEB_PAGE, to None.
for kind in ('img_400', 'video_400', 'audio', 'pdf', 'video_loop_200',
'img_200_sticker', 'video_loop_400'):
assert kind in RENDERERS
def test_web_page_is_the_only_kind_none_entry():
assert MEDIA_SOURCES[MessageMediaType.WEB_PAGE](SimpleNamespace(web_page=None))[1] is None
def test_paid_media_has_no_table_entry():
assert MessageMediaType.PAID_MEDIA not in MEDIA_SOURCES
# --------------------------------------------------------------------------- #
# 3. Selector unit tests (the branchy selectors).
# --------------------------------------------------------------------------- #
def test_select_document_pdf_vs_image():
pdf = SimpleNamespace(document=SimpleNamespace(mime_type="application/pdf", file_unique_id="d1"))
png = SimpleNamespace(document=SimpleNamespace(mime_type="image/png", file_unique_id="d2"))
assert _select_document(pdf)[1] == 'pdf'
assert _select_document(png)[1] == 'img_400'
assert _select_document(pdf)[0] is pdf.document
def test_select_sticker_video_vs_image():
vid = SimpleNamespace(sticker=SimpleNamespace(is_video=True, file_unique_id="s1"))
img = SimpleNamespace(sticker=SimpleNamespace(is_video=False, file_unique_id="s2"))
assert _select_sticker(vid)[1] == 'video_loop_200'
assert _select_sticker(img)[1] == 'img_200_sticker'
def test_select_story_maps_helper_kind():
vid = SimpleNamespace(story=SimpleNamespace(video=SimpleNamespace(file_unique_id="v"), photo=None))
pic = SimpleNamespace(story=SimpleNamespace(video=None, photo=SimpleNamespace(file_unique_id="p")))
none = SimpleNamespace(story=None)
assert _select_story(vid)[1] == 'video_400'
assert _select_story(pic)[1] == 'img_400'
assert _select_story(none) == (None, None)
def test_select_poll_media_maps_helper_kind():
img = SimpleNamespace(poll=SimpleNamespace(description_media=SimpleNamespace(photo=SimpleNamespace(file_unique_id="p"))))
vid = SimpleNamespace(poll=SimpleNamespace(description_media=SimpleNamespace(video=SimpleNamespace(file_unique_id="v"))))
none = SimpleNamespace(poll=None)
assert _select_poll_media(img)[1] == 'img_400'
assert _select_poll_media(vid)[1] == 'video_400'
assert _select_poll_media(none) == (None, None)
# --------------------------------------------------------------------------- #
# 4. Renderer byte structure (audio emits two items; pdf emits its two-append block).
# --------------------------------------------------------------------------- #
def test_audio_renderer_emits_tag_and_br():
out = RENDERERS['audio'](RenderCtx(url="U", mime="audio/mpeg"))
assert len(out) == 2 and out[1] == '<br>'
assert 'type="audio/mpeg"' in out[0]
def test_pdf_renderer_emits_two_item_block():
out = RENDERERS['pdf'](RenderCtx(url="U", tg_link="https://t.me/x/1"))
assert len(out) == 2
assert out[0] == '<div class="document-pdf" style="padding: 10px;">'
assert out[1] == '<a href="https://t.me/x/1" target="_blank">[PDF-файл]</a></div>'
# --------------------------------------------------------------------------- #
# 5. Cross-module invariant: every selector object is resolvable in api_server.
# --------------------------------------------------------------------------- #
def _invariant_messages():
"""One mock per MEDIA_SOURCES type whose selected object carries a real
file_unique_id/file_id, so find_file_id_in_message can resolve it."""
def base(**extra):
m = SimpleNamespace(id=1, chat=SimpleNamespace(id=-100, username="c"), web_page=None, poll=None)
for a in ("photo", "video", "document", "audio", "voice", "video_note",
"animation", "sticker"):
setattr(m, a, None)
for k, v in extra.items():
setattr(m, k, v)
return m
media = SimpleNamespace(file_unique_id="UID", file_id="FID")
cases = {
MessageMediaType.PHOTO: base(media=MessageMediaType.PHOTO, photo=media),
MessageMediaType.VIDEO: base(media=MessageMediaType.VIDEO, video=media),
MessageMediaType.ANIMATION: base(media=MessageMediaType.ANIMATION, animation=media),
MessageMediaType.VIDEO_NOTE: base(media=MessageMediaType.VIDEO_NOTE, video_note=media),
MessageMediaType.AUDIO: base(media=MessageMediaType.AUDIO, audio=media),
MessageMediaType.VOICE: base(media=MessageMediaType.VOICE, voice=media),
MessageMediaType.DOCUMENT: base(media=MessageMediaType.DOCUMENT,
document=SimpleNamespace(mime_type="image/png", file_unique_id="UID", file_id="FID")),
MessageMediaType.STICKER: base(media=MessageMediaType.STICKER,
sticker=SimpleNamespace(is_video=False, file_unique_id="UID", file_id="FID")),
MessageMediaType.LIVE_PHOTO: base(media=MessageMediaType.LIVE_PHOTO, live_photo=media),
MessageMediaType.STORY: base(media=MessageMediaType.STORY,
story=SimpleNamespace(video=media, photo=None)),
MessageMediaType.POLL: base(media=MessageMediaType.POLL,
poll=SimpleNamespace(description_media=SimpleNamespace(photo=media))),
MessageMediaType.WEB_PAGE: base(media=MessageMediaType.WEB_PAGE,
web_page=SimpleNamespace(photo=media)),
}
return cases
@pytest.mark.parametrize("media_type", list(MEDIA_SOURCES.keys()))
async def test_selector_object_is_resolvable_in_api_server(media_type):
"""The object MEDIA_SOURCES selects is found by api_server.find_file_id_in_message
via its file_unique_id — the /media download path can always resolve a URL the
table produced (spec §5a inter-module invariant)."""
message = _invariant_messages()[media_type]
selected_obj, _kind = MEDIA_SOURCES[media_type](message)
file_unique_id = getattr(selected_obj, "file_unique_id", None)
assert file_unique_id, f"{media_type} selector returned no usable file_unique_id"
resolved = await find_file_id_in_message(message, file_unique_id)
assert resolved == getattr(selected_obj, "file_id", None), \
f"{media_type}: selected object not resolvable in find_file_id_in_message"
def test_media_sources_covers_all_old_ladder_types():
"""Every type the pre-refactor _get_file_unique_id dict handled is in the table."""
old_types = {
MessageMediaType.PHOTO, MessageMediaType.VIDEO, MessageMediaType.DOCUMENT,
MessageMediaType.AUDIO, MessageMediaType.VOICE, MessageMediaType.VIDEO_NOTE,
MessageMediaType.ANIMATION, MessageMediaType.STICKER, MessageMediaType.WEB_PAGE,
MessageMediaType.LIVE_PHOTO, MessageMediaType.STORY, MessageMediaType.POLL,
}
assert old_types <= set(MEDIA_SOURCES.keys())
# --------------------------------------------------------------------------- #
# 6. /flags endpoint constraint: flags.append(...) stays inside _extract_flags so
# inspect.getsource still discovers them.
# --------------------------------------------------------------------------- #
def test_get_all_possible_flags_nonempty_and_known():
flags = PostParser.get_all_possible_flags()
assert flags, "flag introspection returned nothing"
for known in ("video", "audio", "no_image", "sticker", "poll", "fwd"):
assert known in flags, f"known flag {known} not discovered by /flags introspection"