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