from __future__ import annotations
import base64
import html
import io
import re
from PIL import Image
BBOX_IMAGE_PATTERN = re.compile(
r'',
flags=re.IGNORECASE,
)
UNMATERIALIZED_BBOX_IMAGE_PATTERN = re.compile(
r'
]*\bsrc=["\']images/bbox_[^"\']+["\'][^>]*>',
flags=re.IGNORECASE,
)
def clean_truncated_repeats(
text: str,
min_text_len: int = 8000,
max_period: int = 200,
min_period: int = 1,
min_repeat_chars: int = 100,
min_repeat_times: int = 5,
) -> str:
"""Remove a repeated suffix created when generation reaches its token ceiling."""
n = len(text)
if n < min_text_len:
return text
max_period = min(max_period, n - 1)
for unit_len in range(min_period, max_period + 1):
if text[n - 1] != text[n - 1 - unit_len]:
continue
match_len = 1
idx = n - 2
while idx >= unit_len and text[idx] == text[idx - unit_len]:
match_len += 1
idx -= 1
total_len = match_len + unit_len
repeat_times = total_len // unit_len
tail_len = total_len % unit_len
if repeat_times >= min_repeat_times and total_len >= min_repeat_chars:
return text[: n - total_len + unit_len] + text[n - tail_len :]
return text
def neutralize_unmaterialized_bbox_images(markdown: str) -> str:
"""Render placeholder examples as code instead of issuing broken requests."""
def replace(match: re.Match[str]) -> str:
escaped = html.escape(match.group(0), quote=False)
return f'
{escaped}'
return UNMATERIALIZED_BBOX_IMAGE_PATTERN.sub(replace, markdown)
def stream_safe_markdown(markdown: str) -> str:
"""Avoid broken image requests until a page's bbox crops are materialized."""
return neutralize_unmaterialized_bbox_images(
BBOX_IMAGE_PATTERN.sub(
'