"""aiogram-style composable formatting built on Peyk's RichText IR.
The composition objects are platform-neutral. Rendering happens only when a
target platform is known, so application code does not carry platform names.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Protocol, Sequence, Union, cast
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from peyk.bot.policy import UnsupportedPolicy
from peyk.platform_core.contracts import PlatformCapabilities
from peyk.platform_core.enums import ParseMode
from peyk.platform_core.errors import UnsupportedFeatureError
from peyk.utils.text_formatting import RichText, _Element, _Text, bale_expandable, bold, code, italic, link, mention_user, mention_username, pre, render_for, render_rubika_metadata, spoiler, strikethrough, underline
Content = Union[str, RichText, 'Text']
[docs]
class TextDefaults(Protocol):
"""Minimal defaults contract required by :func:`resolve_text`."""
parse_mode: ParseMode | str | None
def _rich(value: Content) -> RichText:
if isinstance(value, Text):
return value.to_rich_text()
return RichText.from_value(value)
[docs]
@dataclass(frozen=True)
class Text:
"""A composable sequence of text fragments and formatting nodes."""
body: tuple[Content, ...]
def __init__(self, *body: Content, sep: str='') -> None:
if sep:
parts: list[Content] = []
for index, item in enumerate(body):
if index:
parts.append(sep)
parts.append(item)
body = tuple(parts)
object.__setattr__(self, 'body', tuple(body))
[docs]
def to_rich_text(self) -> RichText:
"""Compile this composition into Peyk's platform-neutral IR."""
result = RichText()
for item in self.body:
result = result + _rich(item)
return result
def __add__(self, other: Content) -> 'Text':
"""Return a new composition containing ``other``."""
return Text(*self.body, other)
[docs]
def line(self) -> 'Text':
"""Append a newline and return a new composition."""
return Text(*self.body, "\n")
[docs]
def bold(self, *body: Content) -> 'Text':
"""Append a bold fragment and return a new composition."""
return Text(*self.body, Bold(*body))
[docs]
def italic(self, *body: Content) -> 'Text':
"""Append an italic fragment and return a new composition."""
return Text(*self.body, Italic(*body))
[docs]
def underline(self, *body: Content) -> 'Text':
"""Append an underlined fragment and return a new composition."""
return Text(*self.body, Underline(*body))
[docs]
def strikethrough(self, *body: Content) -> 'Text':
"""Append a strikethrough fragment and return a new composition."""
return Text(*self.body, Strikethrough(*body))
[docs]
def spoiler(self, *body: Content) -> 'Text':
"""Append a spoiler fragment and return a new composition."""
return Text(*self.body, Spoiler(*body))
[docs]
def code(self, *body: Content) -> 'Text':
"""Append an inline-code fragment and return a new composition."""
return Text(*self.body, Code(*body))
[docs]
def pre(self, *body: Content, language: str | None = None) -> 'Text':
"""Append a preformatted fragment and return a new composition."""
return Text(*self.body, Pre(*body, language=language))
[docs]
def link(self, text: Content, url: str) -> 'Text':
"""Append a URL link and return a new composition."""
return Text(*self.body, TextLink(text, url=url))
[docs]
def mention(self, text: Content, user_id: int | str) -> 'Text':
"""Append a Telegram user mention and return a new composition."""
return Text(*self.body, TextMention(text, user_id=user_id))
[docs]
def blockquote(self, *body: Content) -> 'Text':
"""Append a block quote and return a new composition."""
return Text(*self.body, BlockQuote(*body))
[docs]
def expandable_blockquote(self, *body: Content) -> 'Text':
"""Append an expandable block quote and return a new composition."""
return Text(*self.body, ExpandableBlockQuote(*body))
[docs]
def render(self, target: PlatformCapabilities | str) -> str:
"""Render this composition for a platform or audited capability object."""
return render_for(self.to_rich_text(), _target_capabilities(target))
[docs]
def as_kwargs(self, target: PlatformCapabilities | str) -> dict[str, object]:
"""Return the exact text-bearing keyword arguments for ``target``."""
caps = _target_capabilities(target)
if caps.platform == 'rubika':
text, metadata = render_rubika_metadata(self.to_rich_text())
return {'text': text, 'metadata': {'meta_data_parts': metadata}}
text = render_for(self.to_rich_text(), caps)
if caps.platform == 'telegram':
return {'text': text, 'parse_mode': 'HTML'}
return {'text': text}
[docs]
class Bold(Text):
"""Render children in bold."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return bold(_compile_body(self.body))
[docs]
class Italic(Text):
"""Render children in italics."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return italic(_compile_body(self.body))
[docs]
class Underline(Text):
"""Render children with an underline where the target supports it."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return underline(_compile_body(self.body))
[docs]
class Strikethrough(Text):
"""Render children with a strikethrough where the target supports it."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return strikethrough(_compile_body(self.body))
[docs]
class Spoiler(Text):
"""Render children as a spoiler where the target supports it."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return spoiler(_compile_body(self.body))
[docs]
class Code(Text):
"""Render children as inline code."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return code(_compile_body(self.body))
[docs]
class Pre(Text):
"""Render children as a preformatted block."""
def __init__(self, *body: Content, language: str | None=None) -> None:
super().__init__(*body)
object.__setattr__(self, 'language', language)
language: str | None
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return pre(_compile_body(self.body), self.language)
[docs]
class TextLink(Text):
"""Render children as a URL link."""
def __init__(self, *body: Content, url: str) -> None:
super().__init__(*body)
object.__setattr__(self, 'url', url)
url: str
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return link(_compile_body(self.body), self.url)
[docs]
class TextMention(Text):
"""Render children as a Telegram user mention."""
def __init__(self, *body: Content, user_id: int | str) -> None:
super().__init__(*body)
object.__setattr__(self, 'user_id', user_id)
user_id: int | str
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return mention_user(_compile_body(self.body), self.user_id)
[docs]
class BlockQuote(Text):
"""Render children as a block quote."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return _wrap_composition('blockquote', self.body)
[docs]
class ExpandableBlockQuote(Text):
"""Render children as an expandable block quote."""
[docs]
def to_rich_text(self) -> RichText:
"""Provides the to rich text operation for the peyk integration.
Returns:
Result produced by the operation."""
return _wrap_composition('expandable_blockquote', self.body)
def _compile_body(body: Sequence[Content]) -> RichText:
result = RichText()
for item in body:
result = result + _rich(item)
return result
def _wrap_composition(kind: str, body: Sequence[Content]) -> RichText:
return RichText((_Element(kind, _compile_body(body).nodes),))
def _target_capabilities(target: PlatformCapabilities | str) -> PlatformCapabilities:
if isinstance(target, PlatformCapabilities):
return target
from peyk.platform_core.adapters import BALE_CAPABILITIES, RUBIKA_CAPABILITIES, TELEGRAM_CAPABILITIES
try:
return {'telegram': TELEGRAM_CAPABILITIES, 'bale': BALE_CAPABILITIES, 'rubika': RUBIKA_CAPABILITIES}[target]
except KeyError as exc:
raise ValueError(f'unknown formatting target: {target!r}') from exc
[docs]
def as_line(*items: Content, sep: str=' ') -> Text:
"""Join items on one line."""
return Text(*_interleave(items, sep))
[docs]
def as_list(*items: Content | Iterable[Content], sep: str='\n') -> Text:
"""Join items as a list separated by ``sep``; one iterable is also accepted."""
values = _normalize_items(items)
return Text(*_interleave(values, sep))
[docs]
def as_marked_list(*items: Content | Iterable[Content], marker: str='▫️', sep: str='\n') -> Text:
"""Prefix each item with ``marker``."""
values = _normalize_items(items)
return as_list(*(Text(marker, ' ', item) for item in values), sep=sep)
[docs]
def as_numbered_list(*items: Content | Iterable[Content], sep: str='\n') -> Text:
"""Prefix each item with its one-based position."""
values = _normalize_items(items)
return as_list(*(Text(f'{index}. ', item) for index, item in enumerate(values, 1)), sep=sep)
[docs]
def as_section(title: Content, *body: Content, sep: str='\n') -> Text:
"""Return a title followed by body lines."""
return Text(title, *body, sep=sep)
[docs]
def as_marked_section(title: Content, *body: Content, marker: str='▫️', sep: str='\n') -> Text:
"""Return a title followed by a marked list."""
return Text(title, as_marked_list(*body, marker=marker, sep=sep), sep=sep)
[docs]
def as_key_value(key: Content, value: Content, sep: str=': ') -> Text:
"""Join a key and value with ``sep``."""
return Text(key, value, sep=sep)
def _normalize_items(items: Sequence[Content | Iterable[Content]]) -> tuple[Content, ...]:
if len(items) == 1 and (not isinstance(items[0], (str, RichText, Text))):
candidate = items[0]
if isinstance(candidate, Iterable):
return cast(tuple[Content, ...], tuple(candidate))
return cast(tuple[Content, ...], tuple(items))
def _interleave(items: Sequence[Content], sep: str) -> tuple[Content, ...]:
result: list[Content] = []
for index, item in enumerate(items):
if index:
result.append(sep)
result.append(item)
return tuple(result)
[docs]
def resolve_text(content: str | RichText | Text, platform_caps: PlatformCapabilities, policy: 'UnsupportedPolicy', defaults: TextDefaults) -> tuple[str, dict[str, object]]:
"""Resolve text once at send time into target text and native keyword arguments.
Plain strings remain plain unless a default parse mode is configured.
Rich compositions are rendered from the neutral IR; Rubika additionally
receives its structured metadata. Explicit parse-mode handling is applied
by :class:`peyk.Bot` before this resolver is called.
"""
from peyk.bot.policy import UnsupportedPolicy
if isinstance(content, str):
native: dict[str, object] = {}
parse_mode = getattr(defaults, 'parse_mode', None)
if parse_mode is not None:
native['parse_mode'] = parse_mode.value if isinstance(parse_mode, ParseMode) else str(parse_mode)
return (content, native)
rich = content if isinstance(content, RichText) else content.to_rich_text()
text = render_for(rich, platform_caps)
if platform_caps.platform == 'rubika':
rendered, metadata = render_rubika_metadata(rich)
return (rendered, {'metadata': {'meta_data_parts': metadata}})
if platform_caps.platform == 'telegram':
return (text, {'parse_mode': 'HTML'})
return (text, {})
__all__ = ['Text', 'Bold', 'Italic', 'Underline', 'Strikethrough', 'Spoiler', 'Code', 'Pre', 'TextLink', 'TextMention', 'BlockQuote', 'ExpandableBlockQuote', 'as_line', 'as_list', 'as_marked_list', 'as_numbered_list', 'as_section', 'as_marked_section', 'as_key_value', 'resolve_text', 'TextDefaults', 'bold', 'italic', 'underline', 'strikethrough', 'spoiler', 'code', 'pre', 'link', 'mention_user', 'mention_username', 'bale_expandable', 'ParseMode']