dheraingoud's picture
feat: synchronize proxy with upstream commits up to 5305bd7 and integrate NIM key rotator
0a54372
Raw
History Blame Contribute Delete
1.89 kB
"""Render and truncate ordered transcript segments."""
from collections import deque
from collections.abc import Iterable
from .context import RenderCtx
from .segments import Segment
def render_segments(
segments: Iterable[Segment],
ctx: RenderCtx,
*,
limit_chars: int,
status: str | None,
) -> str:
rendered: list[str] = []
for segment in segments:
try:
output = segment.render(ctx)
except Exception:
continue
if output:
rendered.append(output)
status_text = f"\n\n{status}" if status else ""
prefix_marker = ctx.escape_text("... (truncated)\n")
def _join(parts: Iterable[str], add_marker: bool) -> str:
body = "\n".join(parts)
if add_marker and body:
body = prefix_marker + body
return body + status_text if (body or status_text) else status_text
candidate = _join(rendered, add_marker=False)
if len(candidate) <= limit_chars:
return candidate
parts: deque[str] = deque(rendered)
dropped = False
last_part: str | None = None
while parts:
candidate = _join(parts, add_marker=True)
if len(candidate) <= limit_chars:
return candidate
last_part = parts.popleft()
dropped = True
if dropped and last_part:
budget = limit_chars - len(prefix_marker) - len(status_text)
if budget > 20:
tail = (
"..." + last_part[-(budget - 3) :]
if len(last_part) > budget
else last_part
)
candidate = prefix_marker + tail + status_text
if len(candidate) <= limit_chars:
return candidate
if dropped:
minimal = prefix_marker + status_text.lstrip("\n")
if len(minimal) <= limit_chars:
return minimal
return status or ""