100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
import re
|
|
import unicodedata
|
|
|
|
from atproto import models as at_models
|
|
from atproto_client.utils import TextBuilder
|
|
|
|
from .defaults import DM_TEXT_CHUNK_LIMIT, POST_CHAR_LIMIT
|
|
|
|
_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
|
|
_MULTI_NEWLINE_RE = re.compile(r"\n{3,}")
|
|
|
|
|
|
def build_post_with_mentions(
|
|
parts: list[tuple[str, str | None]],
|
|
) -> tuple[str, list[dict] | None]:
|
|
has_mentions = any(did is not None for _, did in parts)
|
|
if not has_mentions:
|
|
text = "".join(content for content, _ in parts)
|
|
return text, None
|
|
|
|
tb = TextBuilder()
|
|
for content, did in parts:
|
|
if did:
|
|
tb.mention(content, did)
|
|
else:
|
|
tb.text(content)
|
|
|
|
return tb.build_text(), tb.build_facets()
|
|
|
|
|
|
def plain_post(text: str) -> tuple[str, None]:
|
|
return text, None
|
|
|
|
|
|
def build_dm_facets(text: str, mentions: list[dict] | None = None) -> list[dict] | None:
|
|
if not mentions:
|
|
return None
|
|
|
|
facets = []
|
|
for m in mentions:
|
|
did = m.get("did", "")
|
|
byte_start = m.get("byte_start")
|
|
byte_end = m.get("byte_end")
|
|
|
|
if byte_start is None or byte_end is None:
|
|
continue
|
|
|
|
facets.append(
|
|
at_models.AppBskyRichtextFacet.Main(
|
|
index=at_models.AppBskyRichtextFacet.ByteSlice(byte_start=byte_start, byte_end=byte_end),
|
|
features=[at_models.AppBskyRichtextFacet.Mention(did=did)],
|
|
)
|
|
)
|
|
|
|
return facets if facets else None
|
|
|
|
|
|
class BlueskyFormatter:
|
|
max_dm_length = DM_TEXT_CHUNK_LIMIT
|
|
max_post_length = POST_CHAR_LIMIT
|
|
|
|
def sanitize_text(self, text: str, payload: object | None = None) -> str:
|
|
text = unicodedata.normalize("NFKC", text)
|
|
text = _CONTROL_RE.sub("", text)
|
|
text = _MULTI_NEWLINE_RE.sub("\n\n", text)
|
|
return text.strip()
|
|
|
|
def markdown_to_native(self, md_text: str) -> str:
|
|
text = md_text
|
|
text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE)
|
|
text = re.sub(r"```[\s\S]*?```", "", text)
|
|
text = re.sub(r"`([^`]+)`", r"\1", text)
|
|
text = re.sub(r"\*\*\*(.+?)\*\*\*", r"_\1_", text)
|
|
text = re.sub(r"\*\*(.+?)\*\*", r"_\1_", text)
|
|
text = re.sub(r"\*(.+?)\*", r"_\1_", text)
|
|
text = re.sub(r"~~(.+?)~~", r"\1", text)
|
|
text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)", text)
|
|
return self.sanitize_text(text)
|
|
|
|
def native_to_markdown(self, native_content: str) -> str:
|
|
return native_content
|
|
|
|
def strip_mentions(self, text: str, ctx: object | None = None) -> str:
|
|
return re.sub(r"@[\w.]+(?:\.bsky\.social)?", "", text).strip()
|
|
|
|
def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]:
|
|
chunks = []
|
|
limit = min(limit, self.max_dm_length)
|
|
while len(text) > limit:
|
|
split_at = text.rfind("\n", 0, limit)
|
|
if split_at < 0:
|
|
split_at = text.rfind(" ", 0, limit)
|
|
if split_at < 0:
|
|
split_at = limit
|
|
chunks.append(text[:split_at])
|
|
text = text[split_at:].lstrip()
|
|
if text:
|
|
chunks.append(text)
|
|
return chunks
|