From bb85da1ca2bd241d7f7af35721734c8b3950e361 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 10:40:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(bluesky):=20=E6=96=B0=E5=A2=9EBluesky=20(A?= =?UTF-8?q?T=20Protocol)=20=E9=A2=91=E9=81=93=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现了完整的Bluesky频道功能，支持私信轮询、@提及通知、账号会话管理、重复消息过滤、配对授权机制以及完整的API调用能力，包含配置schema和插件元数据定义 --- .../channel/extensions/bluesky/__init__.py | 369 ++++++++++++ .../yuxi/channel/extensions/bluesky/config.py | 28 + .../extensions/bluesky/config_schema.py | 63 ++ .../yuxi/channel/extensions/bluesky/dedupe.py | 29 + .../channel/extensions/bluesky/defaults.py | 19 + .../yuxi/channel/extensions/bluesky/format.py | 99 +++ .../channel/extensions/bluesky/gateway.py | 387 ++++++++++++ .../channel/extensions/bluesky/outbound.py | 568 ++++++++++++++++++ .../channel/extensions/bluesky/pairing.py | 39 ++ .../channel/extensions/bluesky/plugin.json | 27 + .../channel/extensions/bluesky/security.py | 41 ++ .../extensions/bluesky/session_store.py | 36 ++ .../yuxi/channel/extensions/bluesky/status.py | 71 +++ .../yuxi/channel/extensions/bluesky/types.py | 51 ++ 14 files changed, 1827 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/bluesky/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/config.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/config_schema.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/dedupe.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/defaults.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/format.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/bluesky/security.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/session_store.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/status.py create mode 100644 backend/package/yuxi/channel/extensions/bluesky/types.py diff --git a/backend/package/yuxi/channel/extensions/bluesky/__init__.py b/backend/package/yuxi/channel/extensions/bluesky/__init__.py new file mode 100644 index 00000000..8188c570 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/__init__.py @@ -0,0 +1,369 @@ +import logging +import uuid + +from atproto import models as at_models + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.context import ChannelContext +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.bluesky.config import ( + list_bluesky_account_ids, + resolve_bluesky_account, +) +from yuxi.channel.extensions.bluesky.config_schema import BLUESKY_CONFIG_SCHEMA +from yuxi.channel.extensions.bluesky.dedupe import DedupeTracker +from yuxi.channel.extensions.bluesky.format import BlueskyFormatter +from yuxi.channel.extensions.bluesky.gateway import BlueskyGateway, BlueskyGatewayError +from yuxi.channel.extensions.bluesky.outbound import BlueskyOutbound +from yuxi.channel.extensions.bluesky.security import BlueskySecurity +from yuxi.channel.extensions.bluesky.pairing import BlueskyPairing +from yuxi.channel.extensions.bluesky.status import BlueskyStatus +from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.routing.models import PeerKind + +logger = logging.getLogger(__name__) + + +class BlueskyPlugin(BaseChannelPlugin): + id = "bluesky" + name = "Bluesky" + order = 101 + label = "Bluesky" + aliases = ["bsky", "atproto"] + resolve_reply_to_mode = "off" + + def __init__(self): + super().__init__() + self._gateway = BlueskyGateway() + self._outbound = BlueskyOutbound(self._gateway) + self._security = BlueskySecurity() + self._pairing = BlueskyPairing() + self._status = BlueskyStatus(self._gateway) + self._formatter = BlueskyFormatter() + self._dedupe = DedupeTracker() + self._config: dict = {} + self._dynamic_allowlist: list[str] = [] + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text"], + reactions=True, + typing_indicator=False, + threads=False, + edit=False, + unsend=True, + reply=True, + media=True, + native_commands=False, + polls=False, + streaming=True, + streaming_mode="block_streaming", + block_streaming=True, + block_streaming_chunk_min_chars=800, + block_streaming_chunk_max_chars=2000, + ) + + def list_account_ids(self, config: dict) -> list[str]: + return list_bluesky_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + config = getattr(self, "_config", None) or {} + account = resolve_bluesky_account(config, account_id) + return { + "account_id": account.account_id, + "name": account.name, + "handle": account.handle, + "configured": account.is_configured, + "dm_policy": account.dm_policy, + "allow_from": account.allow_from, + "enable_notifications": account.enable_notifications, + "enable_firehose": account.enable_firehose, + } + + def is_configured(self, account: dict) -> bool: + return bool(account.get("handle") and account.get("configured")) + + def is_enabled(self, account: dict) -> bool: + return True + + def disabled_reason(self, account: dict) -> str: + if not account.get("handle"): + return "handle not configured" + if not account.get("configured"): + return "appPassword not configured" + return "" + + def describe_account(self, account: dict) -> dict: + return { + "account_id": account.get("account_id", "default"), + "name": account.get("name", ""), + "handle": account.get("handle", ""), + } + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + static = self._security.resolve_allow_from(config, account_id) or [] + return static + self._dynamic_allowlist + + def config_schema(self) -> dict: + return BLUESKY_CONFIG_SCHEMA + + async def start(self, ctx: ChannelContext) -> object: + account = resolve_bluesky_account(ctx.config, ctx.account_id) + + async def on_dm(dm): + unified = UnifiedMessage( + msg_id=uuid.uuid4().hex, + channel_type="bluesky", + account_id=ctx.account_id, + content=dm.text, + sender=PeerInfo( + kind=PeerKind.DIRECT, + id=dm.sender_did, + display_name=dm.sender_handle, + ), + message_type=MessageType.TEXT, + reply_to_id=dm.message_id, + raw_payload={ + "convo_id": dm.convo_id, + "sender_did": dm.sender_did, + "sender_handle": dm.sender_handle, + "rev": dm.rev, + }, + ) + if ctx.queue is not None: + await ctx.queue.put(unified) + + async def on_mention(mention): + unified = UnifiedMessage( + msg_id=uuid.uuid4().hex, + channel_type="bluesky", + account_id=ctx.account_id, + content=mention.text, + sender=PeerInfo( + kind=PeerKind.DIRECT, + id=mention.author_did, + display_name=mention.author_handle, + ), + message_type=MessageType.TEXT, + reply_to_id=mention.uri, + raw_payload={ + "uri": mention.uri, + "cid": mention.cid, + "author_did": mention.author_did, + "author_handle": mention.author_handle, + "reason": mention.reason, + "root_uri": mention.root_uri, + "root_cid": mention.root_cid, + "parent_uri": mention.parent_uri, + "parent_cid": mention.parent_cid, + "source": "mention", + }, + ) + if ctx.queue is not None: + await ctx.queue.put(unified) + + async def authorize_sender(sender_did: str, _text: str) -> bool: + dm_policy = await self._security.resolve_dm_policy(ctx.config, ctx.account_id) + allow_from = self._security.resolve_allow_from(ctx.config, ctx.account_id) + allow_from = allow_from + self._dynamic_allowlist + result = await self._security.authorize(sender_did, dm_policy, allow_from) + if result == "pairing": + logger.info("Bluesky pairing challenge for %s", sender_did) + return False + return result == "allow" + + try: + handle = await self._gateway.start( + account_id=ctx.account_id, + account=account, + on_dm=on_dm, + on_mention=on_mention, + authorize_sender=authorize_sender, + ) + return handle + except BlueskyGatewayError as e: + logger.error("Bluesky start failed for account %s: %s", ctx.account_id, e) + raise + + async def stop(self, ctx: ChannelContext) -> None: + await self._gateway.stop(ctx.account_id) + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + target = target_id.replace("did:", "").strip() + aid = account_id or "default" + return await self._outbound.send_dm(target, content, account_id=aid) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + if media_type.startswith("image/"): + upload_result = await self._outbound.upload_blob(media_url, account_id=account_id or "default") + + image_embed = at_models.AppBskyEmbedImages.Main( + images=[ + at_models.AppBskyEmbedImages.Image( + image=upload_result["blob"], + alt="", + ) + ] + ) + target = target_id.replace("did:", "").strip() + await self._outbound.send_dm( + target, + "", + account_id=account_id or "default", + embed=image_embed, + ) + else: + logger.warning("Bluesky media type '%s' not yet supported", media_type) + + async def probe(self, account: dict | None = None) -> bool: + aid = account.get("account_id", "default") if account else "default" + snapshot = await self._status.probe(aid) + return snapshot.connected and snapshot.session_valid + + def build_summary(self, snapshot: object) -> dict: + return self._status.build_summary(snapshot) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + config = getattr(self, "_config", None) or {} + allow_from = self._security.resolve_allow_from(config, "default") + if not allow_from: + return False + normalized = self._pairing.normalize_allow_entry(peer_id) + for entry in allow_from: + if self._pairing.normalize_allow_entry(entry) == normalized: + return True + return False + + def resolve_dm_policy(self) -> dict: + config = getattr(self, "_config", None) or {} + account = resolve_bluesky_account(config, "default") + return { + "mode": account.dm_policy, + "allow_from": account.allow_from, + } + + async def generate_code(self, peer_id: str) -> str: + return self._pairing.generate_code(peer_id, "") + + async def verify_code(self, peer_id: str, code: str) -> bool: + entry = self._pairing.verify_code(code) + if entry is None: + return False + expected = self._pairing.normalize_allow_entry(peer_id) + actual = self._pairing.normalize_allow_entry(entry["did"]) + return actual == expected + + def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]: + warnings = [] + aid = account_id or "default" + acct = resolve_bluesky_account(config, aid) + if not acct.handle: + warnings.append("Bluesky handle is not configured") + if not acct.app_password: + warnings.append("Bluesky appPassword is not configured") + return warnings + + @property + def config_prefixes(self) -> list[str]: + return ["channels.bluesky"] + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + self._config = next_cfg + + async def on_account_removed(self, account_id: str) -> None: + await self._gateway.stop(account_id) + + async def on_retire(self) -> None: + await self._gateway.stop_all() + + async def build_reply_payload( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + session_key: str | None = None, + agent_config_id: str | None = None, + **kwargs, + ): + from yuxi.channel.protocols import BoundReplyPayload + + return BoundReplyPayload( + target_id=target_id.replace("did:", "").strip(), + content=content, + reply_to_id=reply_to_id, + thread_id=thread_id, + ) + + def sanitize_text(self, text: str, payload: object | None = None) -> str: + return self._formatter.sanitize_text(text, payload) + + def markdown_to_native(self, md_text: str) -> str: + return self._formatter.markdown_to_native(md_text) + + def is_duplicate(self, key: str) -> bool: + return self._dedupe.has(key) + + def mark_seen(self, key: str) -> None: + self._dedupe.has(key) + + def reset(self) -> None: + self._dedupe._store.clear() + + @property + def ttl_seconds(self) -> int: + return self._dedupe.ttl_s + + @property + def max_entries(self) -> int: + return self._dedupe.max_entries + + async def list_allow_entries(self, config: dict, scope: str) -> list[str]: + static = self._security.resolve_allow_from(config, "default") or [] + return static + self._dynamic_allowlist + + async def add_allow_entry( + self, + entry: str, + *, + account_id: str | None = None, + ) -> None: + did = self._pairing.normalize_allow_entry(entry) + if did not in self._dynamic_allowlist: + self._dynamic_allowlist.append(did) + logger.info("Bluesky allowlist entry added: %s", did) + + async def remove_allow_entry( + self, + entry: str, + *, + account_id: str | None = None, + ) -> None: + did = self._pairing.normalize_allow_entry(entry) + if did in self._dynamic_allowlist: + self._dynamic_allowlist.remove(did) + logger.info("Bluesky allowlist entry removed: %s", did) + + +ChannelPluginRegistry.register(BlueskyPlugin()) diff --git a/backend/package/yuxi/channel/extensions/bluesky/config.py b/backend/package/yuxi/channel/extensions/bluesky/config.py new file mode 100644 index 00000000..098e80a4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/config.py @@ -0,0 +1,28 @@ +from .types import BlueskyAccount + + +def list_bluesky_account_ids(config: dict) -> list[str]: + accounts = config.get("channels", {}).get("bluesky", {}).get("accounts", {}) + if accounts: + return list(accounts.keys()) + return ["default"] + + +def resolve_bluesky_account(config: dict, account_id: str = "default") -> BlueskyAccount: + bluesky_cfg = config.get("channels", {}).get("bluesky", {}) + account_cfg = bluesky_cfg.get("accounts", {}).get(account_id, bluesky_cfg) + + handle = account_cfg.get("handle", "") + app_password = account_cfg.get("appPassword", "") + + return BlueskyAccount( + account_id=account_id, + handle=handle, + app_password=app_password, + session_string=account_cfg.get("sessionString", ""), + dm_policy=account_cfg.get("dmPolicy", "pairing"), + allow_from=account_cfg.get("allowFrom", []), + enable_notifications=account_cfg.get("enableNotifications", False), + enable_firehose=account_cfg.get("enableFirehose", False), + name=account_cfg.get("name", handle), + ) diff --git a/backend/package/yuxi/channel/extensions/bluesky/config_schema.py b/backend/package/yuxi/channel/extensions/bluesky/config_schema.py new file mode 100644 index 00000000..d3d47e2a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/config_schema.py @@ -0,0 +1,63 @@ +from pydantic import BaseModel, Field + + +class BlueskyAccountConfig(BaseModel): + handle: str = "" + app_password: str = Field(default="", alias="appPassword") + session_string: str = Field(default="", alias="sessionString") + dm_policy: str = Field(default="pairing", alias="dmPolicy") + allow_from: list[str] = Field(default_factory=list, alias="allowFrom") + enable_notifications: bool = Field(default=False, alias="enableNotifications") + enable_firehose: bool = Field(default=False, alias="enableFirehose") + name: str = "" + + model_config = {"populate_by_name": True} + + +class BlueskyConfig(BaseModel): + accounts: dict[str, BlueskyAccountConfig] = Field(default_factory=dict) + + +BLUESKY_CONFIG_SCHEMA = { + "type": "object", + "properties": { + "handle": { + "type": "string", + "description": "Bluesky handle (e.g. my-bot.bsky.social)", + }, + "appPassword": { + "type": "string", + "description": "Bluesky App Password with DM permission", + "sensitive": True, + }, + "sessionString": { + "type": "string", + "description": "Persisted session token (auto-managed)", + "sensitive": True, + }, + "dmPolicy": { + "type": "string", + "enum": ["pairing", "allowlist", "open", "disabled"], + "default": "pairing", + }, + "allowFrom": { + "type": "array", + "items": {"type": "string"}, + "description": "Allowed DIDs (did:plc:xxx) for allowlist/pairing policies", + }, + "enableNotifications": { + "type": "boolean", + "default": False, + "description": "Enable @mention notification polling", + }, + "enableFirehose": { + "type": "boolean", + "default": False, + "description": "Enable Firehose WebSocket (P2 feature)", + }, + "name": { + "type": "string", + "description": "Display name for this account", + }, + }, +} diff --git a/backend/package/yuxi/channel/extensions/bluesky/dedupe.py b/backend/package/yuxi/channel/extensions/bluesky/dedupe.py new file mode 100644 index 00000000..c3542a93 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/dedupe.py @@ -0,0 +1,29 @@ +import time +from collections import OrderedDict + + +class DedupeTracker: + def __init__(self, max_entries: int = 50_000, ttl_s: int = 3600): + self.max_entries = max_entries + self.ttl_s = ttl_s + self._store: OrderedDict[str, float] = OrderedDict() + self._last_prune = time.monotonic() + + def has(self, event_id: str) -> bool: + self._maybe_prune() + if event_id in self._store: + self._store.move_to_end(event_id) + return True + self._store[event_id] = time.monotonic() + if len(self._store) > self.max_entries: + self._store.popitem(last=False) + return False + + def _maybe_prune(self): + now = time.monotonic() + if now - self._last_prune < 600: + return + self._last_prune = now + expired = [k for k, ts in self._store.items() if now - ts > self.ttl_s] + for k in expired: + del self._store[k] diff --git a/backend/package/yuxi/channel/extensions/bluesky/defaults.py b/backend/package/yuxi/channel/extensions/bluesky/defaults.py new file mode 100644 index 00000000..d421de4b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/defaults.py @@ -0,0 +1,19 @@ +DEFAULT_POLL_INTERVAL = 5.0 +DEFAULT_NOTIF_INTERVAL = 15.0 +DEFAULT_DEDUPE_MAX = 50_000 +DEFAULT_DEDUPE_TTL = 3600 +DEFAULT_DM_POLICY = "pairing" +PAIRING_CODE_LENGTH = 6 +PAIRING_CODE_TTL = 300 +DM_TEXT_CHUNK_LIMIT = 2000 +POST_CHAR_LIMIT = 300 + +CREATE_SESSION_LIMIT_5MIN = 30 +CREATE_SESSION_LIMIT_DAY = 300 +REFRESH_SESSION_LIMIT_5MIN = 30 +REFRESH_SESSION_LIMIT_DAY = 300 + +WRITE_POINTS_CREATE = 3 +WRITE_POINTS_UPDATE = 2 +WRITE_POINTS_DELETE = 1 +WRITE_POINTS_DAILY_LIMIT = 5000 diff --git a/backend/package/yuxi/channel/extensions/bluesky/format.py b/backend/package/yuxi/channel/extensions/bluesky/format.py new file mode 100644 index 00000000..9cc55e50 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/format.py @@ -0,0 +1,99 @@ +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 diff --git a/backend/package/yuxi/channel/extensions/bluesky/gateway.py b/backend/package/yuxi/channel/extensions/bluesky/gateway.py new file mode 100644 index 00000000..8ae8c963 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/gateway.py @@ -0,0 +1,387 @@ +import asyncio +import logging +import random +from dataclasses import dataclass + +from atproto import Client, SessionEvent +from atproto import models as at_models + +from .dedupe import DedupeTracker +from .session_store import load_session, save_session +from .types import BlueskyAccount, InboundDM, InboundMention + +logger = logging.getLogger(__name__) + + +@dataclass +class BlueskyClientHandle: + client: Client + dm_client: Client + me_did: str + me_handle: str + account_id: str + + +@dataclass +class BlueskyPollHandle: + task: asyncio.Task + dedupe: DedupeTracker + + +class BlueskyGatewayError(Exception): + pass + + +class BlueskyGateway: + def __init__(self): + self._active_clients: dict[str, BlueskyClientHandle] = {} + self._poll_handles: dict[str, BlueskyPollHandle] = {} + self._notif_tasks: dict[str, asyncio.Task] = {} + + async def _create_or_restore_client(self, account: BlueskyAccount, account_id: str) -> BlueskyClientHandle: + client = Client() + + session_str = account.session_string or load_session(account_id) + if session_str: + try: + client.login(session_string=session_str) + me = client.me + logger.info( + "Bluesky session restored for %s (%s)", + account_id, + me.handle, + ) + + client.on_session_change = lambda event, session: ( + save_session(account_id, client.export_session_string()) + if event in (SessionEvent.CREATE, SessionEvent.REFRESH) + else None + ) + + self._label_as_bot(client, me.did) + + return BlueskyClientHandle( + client=client, + dm_client=client.with_bsky_chat_proxy(), + me_did=me.did, + me_handle=me.handle, + account_id=account_id, + ) + except Exception as e: + logger.warning( + "Bluesky session restore failed for %s: %s, will re-login", + account_id, + e, + ) + + client.login(account.handle, account.app_password) + me = client.me + logger.info("Bluesky login success for %s (%s)", account_id, me.handle) + + session_str = client.export_session_string() + save_session(account_id, session_str) + + client.on_session_change = lambda event, session: ( + save_session(account_id, client.export_session_string()) + if event in (SessionEvent.CREATE, SessionEvent.REFRESH) + else None + ) + + self._label_as_bot(client, me.did) + + return BlueskyClientHandle( + client=client, + dm_client=client.with_bsky_chat_proxy(), + me_did=me.did, + me_handle=me.handle, + account_id=account_id, + ) + + def _label_as_bot(self, client: Client, did: str): + try: + profile = client.get_profile(actor=did) + existing_labels = [] + if profile.labels: + existing_labels = [ + at_models.ComAtprotoLabelDefs.SelfLabel(val=label.val) + for label in profile.labels + if hasattr(label, "val") + ] + if any(label.val == "bot" for label in existing_labels): + return + + new_labels = existing_labels + [at_models.ComAtprotoLabelDefs.SelfLabel(val="bot")] + client.app.bsky.actor.profile.put( + did, + at_models.AppBskyActorProfile.Record( + display_name=profile.display_name or "", + description=profile.description or "", + avatar=profile.avatar, + banner=profile.banner, + labels=at_models.ComAtprotoLabelDefs.SelfLabels(values=new_labels), + ), + ) + logger.info("Bluesky bot self-label applied for %s", did) + except Exception as e: + logger.warning("Failed to apply bot self-label: %s", e) + + async def start( + self, + account_id: str, + account: BlueskyAccount, + on_dm: callable, + on_mention: callable, + authorize_sender: callable, + ) -> BlueskyClientHandle: + if not account.is_configured: + raise BlueskyGatewayError(f"Bluesky account '{account_id}' not configured") + + handle = await self._create_or_restore_client(account, account_id) + self._active_clients[account_id] = handle + + dedupe = DedupeTracker() + task = asyncio.create_task( + self._poll_dm_loop( + account_id, + handle, + on_dm, + authorize_sender, + dedupe, + ) + ) + self._poll_handles[account_id] = BlueskyPollHandle(task=task, dedupe=dedupe) + + if account.enable_notifications: + notif_task = asyncio.create_task( + self._poll_notifications_loop( + account_id, + handle, + on_mention, + dedupe, + ) + ) + self._notif_tasks[account_id] = notif_task + + return handle + + async def stop(self, account_id: str): + handle = self._poll_handles.pop(account_id, None) + if handle: + handle.task.cancel() + try: + await handle.task + except asyncio.CancelledError: + pass + + notif_task = self._notif_tasks.pop(account_id, None) + if notif_task: + notif_task.cancel() + try: + await notif_task + except asyncio.CancelledError: + pass + + self._active_clients.pop(account_id, None) + logger.info("Bluesky gateway stopped for account %s", account_id) + + def get_client(self, account_id: str) -> BlueskyClientHandle | None: + return self._active_clients.get(account_id) + + async def stop_all(self): + for account_id in list(self._active_clients.keys()): + await self.stop(account_id) + + async def _poll_dm_loop( + self, + account_id: str, + handle: BlueskyClientHandle, + on_dm: callable, + authorize_sender: callable, + dedupe: DedupeTracker, + interval: float = 5.0, + ): + dm_client = handle.dm_client + cursor: str | None = None + consecutive_errors = 0 + max_interval = 120.0 + + while True: + try: + params = {} + if cursor: + params["cursor"] = cursor + log = dm_client.chat.bsky.convo.get_log(params=params) + cursor = log.cursor + for item in log.logs: + self._process_log_item( + item, + handle, + on_dm, + authorize_sender, + dedupe, + ) + consecutive_errors = 0 + except asyncio.CancelledError: + raise + except Exception as e: + consecutive_errors += 1 + delay = min(interval * (2**consecutive_errors), max_interval) + delay *= random.uniform(0.8, 1.2) + logger.error( + "Bluesky DM poll error for %s (attempt %d, retry in %.1fs): %s", + account_id, + consecutive_errors, + delay, + e, + ) + await asyncio.sleep(delay) + continue + await asyncio.sleep(interval) + + def _process_log_item( + self, + item, + handle: BlueskyClientHandle, + on_dm: callable, + authorize_sender: callable, + dedupe: DedupeTracker, + ): + is_create_message = isinstance(item, at_models.ChatBskyConvoGetLog.LogCreateMessage) + is_add_reaction = isinstance(item, at_models.ChatBskyConvoGetLog.LogAddReaction) + is_remove_reaction = isinstance(item, at_models.ChatBskyConvoGetLog.LogRemoveReaction) + + if is_create_message: + msg = item.message + mid = msg.id + + if dedupe.has(mid): + return + + if msg.sender.did == handle.me_did: + return + + dm = InboundDM( + message_id=mid, + convo_id=item.convo_id, + sender_did=msg.sender.did, + sender_handle=msg.sender.handle or msg.sender.did, + text=msg.text, + sent_at=msg.sent_at, + rev=msg.rev, + ) + + asyncio.create_task(self._dispatch_dm(dm, handle, on_dm, authorize_sender)) + elif is_add_reaction or is_remove_reaction: + mid = f"{item.convo_id}:{item.message_id}:{item.reaction.value}" + if dedupe.has(mid): + return + if item.reaction.sender.did == handle.me_did: + return + asyncio.create_task(self._dispatch_reaction(item, on_dm, is_add_reaction)) + + async def _dispatch_dm( + self, + dm: InboundDM, + handle: BlueskyClientHandle, + on_dm: callable, + authorize_sender: callable, + ): + try: + if not await authorize_sender(dm.sender_did, dm.text): + return + await on_dm(dm) + try: + handle.dm_client.chat.bsky.convo.update_read( + at_models.ChatBskyConvoUpdateRead.Data( + convo_id=dm.convo_id, + message_id=dm.message_id, + ) + ) + except Exception: + pass + except Exception as e: + logger.error("DM dispatch error: %s", e) + + async def _dispatch_reaction(self, item, on_dm, is_add): + + reaction_dm = InboundDM( + message_id=item.message_id, + convo_id=item.convo_id, + sender_did=item.reaction.sender.did, + sender_handle=item.reaction.sender.handle or item.reaction.sender.did, + text=f"[reaction:{'add' if is_add else 'remove'}:{item.reaction.value}]", + sent_at=item.reaction.created_at, + rev="", + ) + try: + await on_dm(reaction_dm) + except Exception as e: + logger.error("Reaction dispatch error: %s", e) + + async def _poll_notifications_loop( + self, + account_id: str, + handle: BlueskyClientHandle, + on_mention: callable, + dedupe: DedupeTracker, + interval: float = 15.0, + ): + client = handle.client + consecutive_errors = 0 + max_interval = 120.0 + + while True: + try: + resp = client.app.bsky.notification.list_notifications(params={"limit": 50}) + for notif in resp.notifications: + if notif.reason not in ("mention", "reply", "like", "repost", "follow", "quote"): + continue + + if dedupe.has(notif.uri): + continue + + mention = InboundMention( + uri=notif.uri, + cid=notif.cid, + author_did=notif.author.did, + author_handle=notif.author.handle or notif.author.did, + text=getattr(notif.record, "text", f"[{notif.reason}]"), + reason=notif.reason, + indexed_at=notif.indexed_at, + ) + + try: + thread = client.get_post_thread(uri=notif.uri) + if thread.thread.reply: + mention.root_uri = thread.thread.reply.root.uri + mention.root_cid = thread.thread.reply.root.cid + mention.parent_uri = thread.thread.reply.parent.uri + mention.parent_cid = thread.thread.reply.parent.cid + except Exception: + pass + + asyncio.create_task(self._dispatch_mention(mention, on_mention)) + + consecutive_errors = 0 + except asyncio.CancelledError: + raise + except Exception as e: + consecutive_errors += 1 + delay = min(interval * (2**consecutive_errors), max_interval) + delay *= random.uniform(0.8, 1.2) + logger.error( + "Bluesky notification poll error for %s (attempt %d, retry in %.1fs): %s", + account_id, + consecutive_errors, + delay, + e, + ) + await asyncio.sleep(delay) + continue + await asyncio.sleep(interval) + + async def _dispatch_mention(self, mention: InboundMention, on_mention: callable): + try: + await on_mention(mention) + except Exception as e: + logger.error("Mention dispatch error: %s", e) diff --git a/backend/package/yuxi/channel/extensions/bluesky/outbound.py b/backend/package/yuxi/channel/extensions/bluesky/outbound.py new file mode 100644 index 00000000..b8960a04 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/outbound.py @@ -0,0 +1,568 @@ +import mimetypes +from pathlib import Path + +from atproto import models as at_models + +from .gateway import BlueskyGateway +from .types import PostRef + + +class BlueskyOutbound: + delivery_mode = "direct" + text_chunk_limit = 2000 + + def __init__(self, gateway: BlueskyGateway): + self.gateway = gateway + + async def send_dm( + self, + target_did: str, + text: str, + account_id: str = "default", + facets: list[dict] | None = None, + embed: dict | None = None, + ) -> str: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + dm = handle.dm_client + + convo = dm.chat.bsky.convo.get_convo_for_members( + at_models.ChatBskyConvoGetConvoForMembers.Params(members=[target_did]) + ).convo + + msg_input = at_models.ChatBskyConvoDefs.MessageInput(text=text) + if facets: + msg_input.facets = facets + if embed: + msg_input.embed = embed + + result = dm.chat.bsky.convo.send_message( + at_models.ChatBskyConvoSendMessage.Data( + convo_id=convo.id, + message=msg_input, + ) + ) + return result.id + + async def send_reply( + self, + parent: PostRef, + text: str, + account_id: str = "default", + ) -> PostRef: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + client = handle.client + + post_view = client.get_posts([parent.uri]).posts[0] + + root_uri = parent.uri + root_cid = parent.cid + if post_view.record.reply: + root_uri = post_view.record.reply.root.uri + root_cid = post_view.record.reply.root.cid + + reply_ref = at_models.AppBskyFeedPost.ReplyRef( + root=at_models.ComAtprotoRepoStrongRef.Main(uri=root_uri, cid=root_cid), + parent=at_models.ComAtprotoRepoStrongRef.Main( + uri=parent.uri, + cid=parent.cid, + ), + ) + + result = client.send_post(text=text, reply_to=reply_ref) + return PostRef(uri=result.uri, cid=result.cid) + + async def send_post( + self, + text: str, + account_id: str = "default", + ) -> PostRef: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + result = handle.client.send_post(text=text) + return PostRef(uri=result.uri, cid=result.cid) + + async def upload_blob( + self, + file_path: str, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + path = Path(file_path) + mime_type, _ = mimetypes.guess_type(path.name) + mime_type = mime_type or "application/octet-stream" + + with open(path, "rb") as f: + blob_data = f.read() + + blob_ref = handle.client.com.atproto.repo.upload_blob(blob_data) + return {"blob": blob_ref.blob, "mime_type": mime_type} + + async def send_post_with_image( + self, + text: str, + image_path: str, + *, + alt_text: str = "", + account_id: str = "default", + ) -> PostRef: + upload_result = await self.upload_blob(image_path, account_id) + + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + image_embed = at_models.AppBskyEmbedImages.Main( + images=[ + at_models.AppBskyEmbedImages.Image( + image=upload_result["blob"], + alt=alt_text or "", + ) + ] + ) + + result = handle.client.send_post(text=text, embed=image_embed) + return PostRef(uri=result.uri, cid=result.cid) + + async def send_post_with_link( + self, + text: str, + uri: str, + *, + title: str = "", + description: str = "", + thumb_blob: dict | None = None, + account_id: str = "default", + ) -> PostRef: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + external_embed = at_models.AppBskyEmbedExternal.Main( + external=at_models.AppBskyEmbedExternal.External( + uri=uri, + title=title, + description=description, + thumb=thumb_blob, + ) + ) + + result = handle.client.send_post(text=text, embed=external_embed) + return PostRef(uri=result.uri, cid=result.cid) + + async def send_reaction( + self, + convo_id: str, + message_id: str, + emoji: str, + account_id: str = "default", + ) -> None: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + dm = handle.dm_client + dm.chat.bsky.convo.add_reaction( + at_models.ChatBskyConvoAddReaction.Data( + convo_id=convo_id, + message_id=message_id, + value=emoji, + ) + ) + + async def remove_reaction( + self, + convo_id: str, + message_id: str, + emoji: str, + account_id: str = "default", + ) -> None: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + dm = handle.dm_client + dm.chat.bsky.convo.remove_reaction( + at_models.ChatBskyConvoRemoveReaction.Data( + convo_id=convo_id, + message_id=message_id, + value=emoji, + ) + ) + + async def get_messages( + self, + convo_id: str, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + dm = handle.dm_client + params = {"convo_id": convo_id, "limit": min(limit, 100)} + if cursor: + params["cursor"] = cursor + + resp = dm.chat.bsky.convo.get_messages(params=params) + messages = [] + for msg in resp.messages: + messages.append( + { + "id": msg.id, + "text": msg.text, + "sender_did": msg.sender.did, + "sender_handle": msg.sender.handle, + "sent_at": str(msg.sent_at) if msg.sent_at else "", + "rev": msg.rev, + } + ) + + return { + "messages": messages, + "cursor": resp.cursor, + } + + async def delete_message( + self, + convo_id: str, + message_id: str, + account_id: str = "default", + ) -> None: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + dm = handle.dm_client + dm.chat.bsky.convo.delete_message_for_self( + at_models.ChatBskyConvoDeleteMessageForSelf.Data( + convo_id=convo_id, + message_id=message_id, + ) + ) + + async def list_convos( + self, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + dm = handle.dm_client + params = {"limit": min(limit, 100)} + if cursor: + params["cursor"] = cursor + resp = dm.chat.bsky.convo.list_convos(params=params) + convos = [] + for c in resp.convos: + convos.append( + { + "id": c.id, + "rev": c.rev, + "members": [{"did": m.did, "handle": m.handle} for m in c.members], + "last_message": { + "id": c.last_message.id if c.last_message else None, + "text": c.last_message.text if c.last_message else "", + "sent_at": str(c.last_message.sent_at) if c.last_message and c.last_message.sent_at else "", + } + if c.last_message + else None, + "unread_count": getattr(c, "unread_count", 0), + "opened": c.opened, + } + ) + return {"convos": convos, "cursor": resp.cursor} + + async def delete_post( + self, + uri: str, + account_id: str = "default", + ) -> None: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + handle.client.com.atproto.repo.delete_record( + at_models.ComAtprotoRepoDeleteRecord.Data( + collection="app.bsky.feed.post", + repo=handle.me_did, + rkey=uri.split("/")[-1], + ) + ) + + async def like_post( + self, + uri: str, + cid: str, + account_id: str = "default", + ) -> str: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + result = handle.client.like(uri=uri, cid=cid) + return result.uri + + async def repost( + self, + uri: str, + cid: str, + account_id: str = "default", + ) -> str: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + result = handle.client.repost(uri=uri, cid=cid) + return result.uri + + async def search_actors( + self, + query: str, + *, + limit: int = 25, + typeahead: bool = False, + account_id: str = "default", + ) -> list[dict]: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + params = {"q": query, "limit": min(limit, 100)} + if typeahead: + resp = handle.client.app.bsky.actor.search_actors_typeahead(params=params) + else: + resp = handle.client.app.bsky.actor.search_actors(params=params) + + return [ + { + "did": a.did, + "handle": a.handle, + "display_name": a.display_name, + "description": a.description, + } + for a in resp.actors + ] + + async def get_follows( + self, + actor: str, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + resp = handle.client.get_follows(actor=actor, limit=min(limit, 100), cursor=cursor) + follows = [{"did": f.did, "handle": f.handle, "display_name": f.display_name} for f in resp.follows] + return {"follows": follows, "cursor": resp.cursor} + + async def get_followers( + self, + actor: str, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + resp = handle.client.get_followers(actor=actor, limit=min(limit, 100), cursor=cursor) + followers = [{"did": f.did, "handle": f.handle, "display_name": f.display_name} for f in resp.followers] + return {"followers": followers, "cursor": resp.cursor} + + async def get_profile( + self, + actor: str, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + profile = handle.client.get_profile(actor=actor) + return { + "did": profile.did, + "handle": profile.handle, + "display_name": profile.display_name, + "description": profile.description, + "avatar": profile.avatar, + "banner": profile.banner, + "followers_count": profile.followers_count, + "follows_count": profile.follows_count, + "posts_count": profile.posts_count, + } + + async def get_preferences( + self, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + resp = handle.client.app.bsky.actor.get_preferences() + return {"preferences": [p.model_dump() for p in resp.preferences]} + + async def follow_user( + self, + did: str, + account_id: str = "default", + ) -> str: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + result = handle.client.follow(did) + return result.uri + + async def unfollow_user( + self, + follow_uri: str, + account_id: str = "default", + ) -> None: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + handle.client.com.atproto.repo.delete_record( + at_models.ComAtprotoRepoDeleteRecord.Data( + collection="app.bsky.graph.follow", + repo=handle.me_did, + rkey=follow_uri.split("/")[-1], + ) + ) + + async def get_timeline( + self, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + params = {"limit": min(limit, 100)} + if cursor: + params["cursor"] = cursor + resp = handle.client.get_timeline(params=params) + posts = [] + for feed_view in resp.feed: + post = feed_view.post + posts.append( + { + "uri": post.uri, + "cid": post.cid, + "author_did": post.author.did, + "author_handle": post.author.handle, + "text": getattr(post.record, "text", ""), + "created_at": getattr(post.record, "created_at", ""), + } + ) + return {"posts": posts, "cursor": resp.cursor} + + async def get_author_feed( + self, + actor: str, + *, + limit: int = 50, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + params = {"actor": actor, "limit": min(limit, 100)} + if cursor: + params["cursor"] = cursor + resp = handle.client.get_author_feed(params=params) + posts = [] + for feed_view in resp.feed: + post = feed_view.post + posts.append( + { + "uri": post.uri, + "cid": post.cid, + "author_did": post.author.did, + "author_handle": post.author.handle, + "text": getattr(post.record, "text", ""), + "created_at": getattr(post.record, "created_at", ""), + } + ) + return {"posts": posts, "cursor": resp.cursor} + + async def search_posts( + self, + query: str, + *, + limit: int = 25, + cursor: str | None = None, + account_id: str = "default", + ) -> dict: + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + params = {"q": query, "limit": min(limit, 100)} + if cursor: + params["cursor"] = cursor + resp = handle.client.app.bsky.feed.search_posts(params=params) + posts = [] + for post in resp.posts: + posts.append( + { + "uri": post.uri, + "cid": post.cid, + "author_did": post.author.did, + "author_handle": post.author.handle, + "text": getattr(post.record, "text", ""), + "created_at": getattr(post.record, "created_at", ""), + } + ) + return {"posts": posts, "cursor": resp.cursor} + + async def send_post_with_video( + self, + text: str, + video_path: str, + *, + alt_text: str = "", + account_id: str = "default", + ) -> PostRef: + upload_result = await self.upload_blob(video_path, account_id) + + handle = self.gateway.get_client(account_id) + if not handle: + raise RuntimeError(f"No active Bluesky client for account '{account_id}'") + + video_embed = at_models.AppBskyEmbedVideo.Main( + video=upload_result["blob"], + alt=alt_text or "", + ) + + result = handle.client.send_post(text=text, embed=video_embed) + return PostRef(uri=result.uri, cid=result.cid) + + def chunker(self, text: str, limit: int, ctx=None) -> list[str]: + chunks = [] + 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 diff --git a/backend/package/yuxi/channel/extensions/bluesky/pairing.py b/backend/package/yuxi/channel/extensions/bluesky/pairing.py new file mode 100644 index 00000000..f2f2f815 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/pairing.py @@ -0,0 +1,39 @@ +import secrets +import time + +from .defaults import PAIRING_CODE_LENGTH, PAIRING_CODE_TTL + + +class BlueskyPairing: + id_label = "DID" + + def __init__(self): + self._pending: dict[str, dict] = {} + + def generate_code(self, sender_did: str, sender_handle: str = "") -> str: + self._cleanup_expired() + + code = secrets.token_hex(PAIRING_CODE_LENGTH // 2)[:PAIRING_CODE_LENGTH].upper() + self._pending[code] = { + "did": sender_did, + "handle": sender_handle, + "created_at": time.time(), + } + return code + + def verify_code(self, code: str) -> dict | None: + self._cleanup_expired() + return self._pending.pop(code, None) + + def normalize_allow_entry(self, entry: str) -> str: + return entry.replace("did:", "").strip().lower() + + def get_pending(self) -> list[dict]: + self._cleanup_expired() + return [{"code": code, **entry} for code, entry in self._pending.items()] + + def _cleanup_expired(self): + now = time.time() + expired = [code for code, entry in self._pending.items() if now - entry["created_at"] > PAIRING_CODE_TTL] + for code in expired: + del self._pending[code] diff --git a/backend/package/yuxi/channel/extensions/bluesky/plugin.json b/backend/package/yuxi/channel/extensions/bluesky/plugin.json new file mode 100644 index 00000000..b53b51d3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/plugin.json @@ -0,0 +1,27 @@ +{ + "id": "bluesky", + "name": "Bluesky", + "version": "1.0.0", + "description": "Bluesky (AT Protocol) channel plugin for ForcePilot. Supports DM via chat.bsky.convo polling and public @mention interactions.", + "author": "ForcePilot", + "order": 101, + "enabled": true, + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text"], + "reactions": true, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": true, + "reply": true, + "media": true, + "native_commands": false, + "polls": false, + "streaming": true, + "streaming_mode": "block_streaming", + "block_streaming": true, + "block_streaming_chunk_min_chars": 800, + "block_streaming_chunk_max_chars": 2000 + } +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/bluesky/security.py b/backend/package/yuxi/channel/extensions/bluesky/security.py new file mode 100644 index 00000000..89c62880 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/security.py @@ -0,0 +1,41 @@ +DM_POLICY_OPEN = "open" +DM_POLICY_PAIRING = "pairing" +DM_POLICY_ALLOWLIST = "allowlist" +DM_POLICY_DISABLED = "disabled" + +VALID_DM_POLICIES = { + DM_POLICY_OPEN, + DM_POLICY_PAIRING, + DM_POLICY_ALLOWLIST, + DM_POLICY_DISABLED, +} + + +class BlueskySecurity: + def __init__(self): + self.default_policy = DM_POLICY_PAIRING + + async def resolve_dm_policy(self, config: dict, account_id: str) -> str: + bluesky_cfg = config.get("channels", {}).get("bluesky", {}) + account_cfg = bluesky_cfg.get("accounts", {}).get(account_id, bluesky_cfg) + policy = account_cfg.get("dmPolicy", self.default_policy) + return policy if policy in VALID_DM_POLICIES else self.default_policy + + def resolve_allow_from(self, config: dict, account_id: str) -> list[str]: + bluesky_cfg = config.get("channels", {}).get("bluesky", {}) + account_cfg = bluesky_cfg.get("accounts", {}).get(account_id, bluesky_cfg) + raw = account_cfg.get("allowFrom", []) + return [entry.replace("did:", "").strip().lower() for entry in raw if entry.strip()] + + async def authorize(self, sender_did: str, dm_policy: str, allow_from: list[str]) -> str: + normalized = sender_did.strip().lower() + + if dm_policy == DM_POLICY_OPEN: + return "allow" + if dm_policy == DM_POLICY_DISABLED: + return "block" + if dm_policy == DM_POLICY_ALLOWLIST: + return "allow" if normalized in allow_from else "block" + if dm_policy == DM_POLICY_PAIRING: + return "allow" if normalized in allow_from else "pairing" + return "block" diff --git a/backend/package/yuxi/channel/extensions/bluesky/session_store.py b/backend/package/yuxi/channel/extensions/bluesky/session_store.py new file mode 100644 index 00000000..b5303335 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/session_store.py @@ -0,0 +1,36 @@ +import os +from pathlib import Path + +SESSION_DIR = Path.home() / ".forcepilot" / "bluesky" + + +def _session_dir() -> Path: + env_dir = os.environ.get("BLUESKY_SESSION_DIR") + if env_dir: + return Path(env_dir) + return SESSION_DIR + + +def save_session(account_id: str, session_string: str) -> None: + session_dir = _session_dir() + session_dir.mkdir(parents=True, exist_ok=True) + path = session_dir / f"{account_id}.session" + tmp = session_dir / f"{account_id}.session.tmp" + tmp.write_text(session_string) + tmp.chmod(0o600) + if path.exists(): + path.unlink() + tmp.rename(path) + + +def load_session(account_id: str) -> str | None: + path = _session_dir() / f"{account_id}.session" + if path.exists(): + return path.read_text().strip() + return None + + +def delete_session(account_id: str) -> None: + path = _session_dir() / f"{account_id}.session" + if path.exists(): + path.unlink() diff --git a/backend/package/yuxi/channel/extensions/bluesky/status.py b/backend/package/yuxi/channel/extensions/bluesky/status.py new file mode 100644 index 00000000..f6d8310f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/status.py @@ -0,0 +1,71 @@ +import datetime +from dataclasses import dataclass + + +@dataclass +class BlueskyStatusSnapshot: + account_id: str + connected: bool + session_valid: bool + dm_service_reachable: bool + me_did: str = "" + me_handle: str = "" + agent_count: int = 0 + last_dm_poll_at: datetime.datetime | None = None + last_dm_poll_error: str = "" + + +class BlueskyStatus: + def __init__(self, gateway): + self._gateway = gateway + + async def probe(self, account_id: str) -> BlueskyStatusSnapshot: + handle = self._gateway.get_client(account_id) + if not handle: + return BlueskyStatusSnapshot( + account_id=account_id, + connected=False, + session_valid=False, + dm_service_reachable=False, + ) + + session_valid = False + me_did = "" + me_handle = "" + try: + me = handle.client.me + if me: + session_valid = True + me_did = getattr(me, "did", "") + me_handle = getattr(me, "handle", "") + except Exception: + pass + + dm_reachable = False + try: + handle.dm_client.chat.bsky.convo.list_convos(params={"limit": 1}) + dm_reachable = True + except Exception: + pass + + return BlueskyStatusSnapshot( + account_id=account_id, + connected=True, + session_valid=session_valid, + dm_service_reachable=dm_reachable, + me_did=me_did, + me_handle=me_handle, + agent_count=len(self._gateway._active_clients), + ) + + def build_summary(self, snapshot: BlueskyStatusSnapshot) -> dict: + return { + "channel": "bluesky", + "account_id": snapshot.account_id, + "connected": snapshot.connected, + "session_valid": snapshot.session_valid, + "dm_service_reachable": snapshot.dm_service_reachable, + "handle": snapshot.me_handle, + "did": snapshot.me_did, + "agent_count": snapshot.agent_count, + } diff --git a/backend/package/yuxi/channel/extensions/bluesky/types.py b/backend/package/yuxi/channel/extensions/bluesky/types.py new file mode 100644 index 00000000..43e08988 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/bluesky/types.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass, field +from datetime import datetime + + +@dataclass +class BlueskyAccount: + account_id: str = "default" + handle: str = "" + app_password: str = "" + session_string: str = "" + dm_policy: str = "pairing" + allow_from: list[str] = field(default_factory=list) + enable_notifications: bool = False + enable_firehose: bool = False + name: str = "" + + @property + def is_configured(self) -> bool: + return bool(self.handle and self.app_password) + + +@dataclass +class InboundDM: + message_id: str + convo_id: str + sender_did: str + sender_handle: str + text: str + sent_at: datetime + rev: str = "" + + +@dataclass +class InboundMention: + uri: str + cid: str + author_did: str + author_handle: str + text: str + reason: str + indexed_at: datetime + root_uri: str | None = None + root_cid: str | None = None + parent_uri: str | None = None + parent_cid: str | None = None + + +@dataclass +class PostRef: + uri: str + cid: str