diff --git a/backend/package/yuxi/channel/extensions/confluence/__init__.py b/backend/package/yuxi/channel/extensions/confluence/__init__.py new file mode 100644 index 00000000..1d033e14 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/__init__.py @@ -0,0 +1,359 @@ +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.context import ChannelContext +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.confluence.attachments import ConfluenceAttachmentManager +from yuxi.channel.extensions.confluence.config import list_confluence_account_ids, resolve_confluence_account +from yuxi.channel.extensions.confluence.content_properties import ContentPropertyStore +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.knowledge import ConfluenceKnowledge +from yuxi.channel.extensions.confluence.labels import ConfluenceLabelManager +from yuxi.channel.extensions.confluence.mentions import ConfluenceMentions +from yuxi.channel.extensions.confluence.outbound import ConfluenceOutbound +from yuxi.channel.extensions.confluence.page_writer import ConfluencePageWriter +from yuxi.channel.extensions.confluence.security import ConfluenceCommentPolicy +from yuxi.channel.extensions.confluence.status import ConfluenceStatus +from yuxi.channel.extensions.confluence.streaming import ConfluenceStreaming +from yuxi.channel.extensions.confluence.types import InboundConfluenceComment, PageContext +from yuxi.channel.extensions.confluence.webhook import ConfluenceWebhookReceiver +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +logger = logging.getLogger(__name__) + + +class ConfluencePlugin(BaseChannelPlugin): + id = "confluence" + name = "Confluence" + order = 90 + label = "Confluence" + aliases = ["confluence-cloud", "atlassian-confluence"] + resolve_reply_to_mode = "off" + + def __init__(self): + super().__init__() + self._gateway: ConfluenceGateway | None = None + self._outbound: ConfluenceOutbound | None = None + self._status: ConfluenceStatus | None = None + self._security: ConfluenceCommentPolicy | None = None + self._streaming = ConfluenceStreaming() + self._page_writer: ConfluencePageWriter | None = None + self._knowledge: ConfluenceKnowledge | None = None + self._webhook: ConfluenceWebhookReceiver | None = None + self._attachment_manager: ConfluenceAttachmentManager | None = None + self._label_manager: ConfluenceLabelManager | None = None + self._content_property_store: ContentPropertyStore | None = None + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "url": { + "type": "string", + "title": "Confluence 站点 URL", + "description": "例如 https://your-domain.atlassian.net", + }, + "email": { + "type": "string", + "title": "Bot 账号邮箱", + }, + "api_token": { + "type": "string", + "title": "API Token", + "format": "password", + }, + "auth_mode": { + "type": "string", + "title": "认证方式", + "enum": ["api_token", "pat", "oauth"], + "default": "api_token", + }, + "space_keys": { + "type": "string", + "title": "空间 Key（逗号分隔）", + "description": "留空则监听所有空间", + }, + "comment_poll_enabled": { + "type": "boolean", + "title": "启用评论轮询", + "default": True, + }, + "poll_interval": { + "type": "number", + "title": "轮询间隔（秒）", + "default": 30, + }, + "comment_policy": { + "type": "string", + "title": "评论策略", + "enum": ["open", "allowlist", "pairing", "disabled"], + "default": "open", + }, + "allow_from": { + "type": "string", + "title": "允许评论的用户 ID（逗号分隔）", + "description": "仅 allowlist 模式下生效", + }, + "enable_page_writer": { + "type": "boolean", + "title": "启用 AI 页面编写", + "default": False, + }, + "enable_knowledge_retrieval": { + "type": "boolean", + "title": "启用知识库检索", + "default": False, + }, + "enable_webhook": { + "type": "boolean", + "title": "启用 Webhook 接收", + "default": False, + }, + "name": { + "type": "string", + "title": "Bot 显示名称", + "default": "Confluence Bot", + }, + }, + "required": ["url", "email", "api_token"], + } + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["comment"], + message_types=["text"], + interactions=[], + reactions=False, + typing_indicator=False, + threads=True, + edit=True, + unsend=True, + reply=True, + media=False, + effects=False, + native_commands=False, + polls=False, + group_management=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=self._streaming.MIN_CHARS, + block_streaming_chunk_max_chars=self._streaming.MAX_CHARS, + block_streaming_chunk_break_preference=self._streaming.CHUNK_BREAK, + block_streaming_coalesce_min_chars=self._streaming.COALESCE_MIN, + block_streaming_coalesce_max_chars=self._streaming.COALESCE_MAX, + block_streaming_coalesce_idle_ms=self._streaming.COALESCE_IDLE_MS, + sse_support=False, + sse_max_reasoning_chars=0, + polling_fallback=True, + tts=None, + ) + + def list_account_ids(self, config: dict | None = None) -> list[str]: + return list_confluence_account_ids(config or {}) + + def resolve_account(self, account_id: str = "default", config: dict | None = None) -> dict: + account = resolve_confluence_account(config or {}, account_id) + return { + "account_id": account.account_id, + "url": account.url, + "email": account.email, + "space_keys": account.space_keys, + "comment_policy": account.comment_policy, + "allow_from": account.allow_from, + "comment_poll_enabled": account.comment_poll_enabled, + "poll_interval": account.poll_interval, + "auth_mode": account.auth_mode, + "deployment_type": account.deployment_type, + "enable_page_writer": account.enable_page_writer, + "enable_knowledge_retrieval": account.enable_knowledge_retrieval, + "enable_webhook": account.enable_webhook, + } + + def is_configured(self, account: dict | None = None) -> bool: + if account is None: + account = self.resolve_account() + return bool(account.get("url") and account.get("email") and account.get("api_token")) + + async def start(self, ctx) -> object: + config = getattr(ctx, "config", ctx) if hasattr(ctx, "config") else (ctx if isinstance(ctx, dict) else {}) + account_id = getattr(ctx, "account_id", "default") if hasattr(ctx, "account_id") else "default" + account = resolve_confluence_account(config, account_id) + + self._gateway = ConfluenceGateway() + self._status = ConfluenceStatus() + self._security = ConfluenceCommentPolicy( + policy=account.comment_policy, + allow_from=account.allow_from, + ) + + async def _authorize_sender(author_id: str, _raw: dict) -> bool: + return self._security.is_allowed(author_id) if self._security else True + + async def _on_comment(comment: InboundConfluenceComment) -> None: + logger.info( + "Confluence comment received: %s from %s on page %s", + comment.comment_id, comment.author_name, + comment.page_context.page_title if comment.page_context else "?" + ) + + client = await self._gateway.start( + account_id=account_id, + account=account, + on_comment=_on_comment, + authorize_sender=_authorize_sender, + ) + self._outbound = ConfluenceOutbound(self._gateway) + + if account.enable_page_writer: + self._page_writer = ConfluencePageWriter(self._gateway) + + if account.enable_knowledge_retrieval: + self._knowledge = ConfluenceKnowledge(self._gateway) + + if account.enable_webhook: + self._webhook = ConfluenceWebhookReceiver( + on_page_changed=None, + on_comment=_on_comment, + ) + await self._auto_register_webhooks(client, account) + + self._attachment_manager = ConfluenceAttachmentManager(self._gateway) + self._label_manager = ConfluenceLabelManager(self._gateway) + self._content_property_store = ContentPropertyStore(self._gateway) + + return client + + async def _auto_register_webhooks(self, client, account) -> None: + try: + existing = await client._request("GET", "/rest/webhooks/1.0/webhook") + hooks = existing if isinstance(existing, list) else existing.get("results", []) + target_url = f"{account.url.rstrip('/')}/webhooks/confluence" + for hook in hooks: + if hook.get("url") == target_url: + logger.info("Webhook already registered: %s", hook.get("id")) + return + result = await client._request( + "POST", + "/rest/webhooks/1.0/webhook", + json={ + "name": "ForcePilot Confluence Integration", + "url": target_url, + "events": list({ + "page_created", "page_updated", "page_removed", + "blog_created", "blog_updated", + "comment_created", "comment_updated", "comment_removed", + }), + }, + ) + logger.info("Webhook registered: %s", result.get("id", "unknown")) + except Exception as e: + logger.warning("Failed to auto-register webhook: %s", e) + + async def stop(self, ctx) -> None: + if self._gateway: + await self._gateway.stop_all() + self._gateway = None + self._outbound = None + self._status = None + self._security = None + self._page_writer = None + self._knowledge = None + self._webhook = None + self._attachment_manager = None + self._label_manager = None + self._content_property_store = None + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + if prev_cfg != next_cfg: + logger.info("Confluence config changed, reloading...") + ctx = ChannelContext(channel_type="confluence", account_id=account_id, config=next_cfg) + await self.stop(ctx) + await self.start(ctx) + + 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, + ) -> dict: + if not content or self._outbound is None: + return {"success": False, "error": "Not ready"} + + result = await self._outbound.send_text( + target_id=target_id, + content=content, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id or "default", + ) + return {"success": result.success, "comment_id": result.comment_id, "error": result.error} + + async def edit_message( + self, + message_id: str, + content: str, + *, + account_id: str | None = None, + ) -> dict: + if not content or self._outbound is None: + return {"success": False, "error": "Not ready"} + result = await self._outbound.edit_comment( + comment_id=message_id, + text=content, + account_id=account_id or "default", + ) + return {"success": result.success, "comment_id": result.comment_id, "error": result.error} + + async def delete_message( + self, + message_id: str, + *, + account_id: str | None = None, + ) -> dict: + if self._outbound is None: + return {"success": False, "error": "Not ready"} + result = await self._outbound.delete_comment( + comment_id=message_id, + account_id=account_id or "default", + ) + return {"success": result.success, "comment_id": result.comment_id, "error": result.error} + + async def probe(self, account: dict | None = None) -> bool: + if self._gateway is None or self._status is None: + return False + cfg = account or {} + aid = cfg.get("account_id", "default") + confluence_account = resolve_confluence_account(cfg, aid) + client = self._gateway.get_client(aid) + if client is None: + return False + result = await self._status.probe(confluence_account, client) + return result.get("success", False) + + def resolve_dm_policy(self) -> dict: + if self._security: + return {"mode": self._security.policy, "allow_from": list(self._security.allowlist)} + return {"mode": "open", "allow_from": []} + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + if self._security: + return self._security.is_allowed(peer_id) + return True + + def extract_mentions(self, raw_message: dict) -> list[dict]: + return ConfluenceMentions.extract_mentions(raw_message) + + def build_context_note(self, context) -> str: + if context is None: + return "" + if isinstance(context, PageContext): + return context.context_label + return str(context) + + +ChannelPluginRegistry.register(ConfluencePlugin()) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/attachments.py b/backend/package/yuxi/channel/extensions/confluence/attachments.py new file mode 100644 index 00000000..f598e4ee --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/attachments.py @@ -0,0 +1,106 @@ +import logging +from pathlib import Path + +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.types import OutboundResult + +logger = logging.getLogger(__name__) + + +class ConfluenceAttachmentManager: + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def list_attachments( + self, page_id: str, account_id: str = "default", + ) -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + try: + result = await client._request( + "GET", f"/wiki/api/v2/pages/{page_id}/attachments", + ) + return result.get("results", []) + except Exception as e: + logger.error("Failed to list attachments for page %s: %s", page_id, e) + return [] + + async def get_attachment( + self, attachment_id: str, account_id: str = "default", + ) -> dict | None: + client = self._gateway.get_client(account_id) + if client is None: + return None + try: + return await client._request( + "GET", f"/wiki/api/v2/attachments/{attachment_id}", + ) + except Exception as e: + logger.error("Failed to get attachment %s: %s", attachment_id, e) + return None + + async def upload_attachment( + self, + page_id: str, + file_path: str, + filename: str | None = None, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + path = Path(file_path) + if not path.exists(): + return OutboundResult(success=False, error=f"File not found: {file_path}") + + fname = filename or path.name + try: + result = await client._request( + "POST", + f"/wiki/api/v2/pages/{page_id}/attachments", + files={"file": (fname, path.read_bytes(), "application/octet-stream")}, + ) + return OutboundResult(success=True, comment_id=result.get("id", "")) + except Exception as e: + logger.error("Failed to upload attachment to page %s: %s", page_id, e) + return OutboundResult(success=False, error=str(e)) + + async def download_attachment( + self, + page_id: str, + filename: str, + save_path: str, + account_id: str = "default", + ) -> str | None: + client = self._gateway.get_client(account_id) + if client is None: + return None + + try: + url = f"/download/attachments/{page_id}/{filename}" + if client._http is None: + raise RuntimeError("Client not initialized") + resp = await client._http.get(url) + resp.raise_for_status() + Path(save_path).write_bytes(resp.content) + return save_path + except Exception as e: + logger.error("Failed to download attachment %s: %s", filename, e) + return None + + async def delete_attachment( + self, attachment_id: str, account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + try: + await client._request( + "DELETE", f"/wiki/api/v2/attachments/{attachment_id}", + ) + return OutboundResult(success=True, comment_id=attachment_id) + except Exception as e: + logger.error("Failed to delete attachment %s: %s", attachment_id, e) + return OutboundResult(success=False, error=str(e)) diff --git a/backend/package/yuxi/channel/extensions/confluence/auth.py b/backend/package/yuxi/channel/extensions/confluence/auth.py new file mode 100644 index 00000000..0cb0f462 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/auth.py @@ -0,0 +1,49 @@ +import logging + +import httpx + +logger = logging.getLogger(__name__) + +ATLASSIAN_TOKEN_URL = "https://auth.atlassian.com/oauth/token" + + +class ConfluenceOAuth2Flow: + @staticmethod + async def get_access_token( + client_id: str, + client_secret: str, + code: str, + redirect_uri: str, + ) -> dict: + async with httpx.AsyncClient() as c: + resp = await c.post( + ATLASSIAN_TOKEN_URL, + json={ + "grant_type": "authorization_code", + "client_id": client_id, + "client_secret": client_secret, + "code": code, + "redirect_uri": redirect_uri, + }, + ) + resp.raise_for_status() + return resp.json() + + @staticmethod + async def refresh_token( + client_id: str, + client_secret: str, + refresh_token: str, + ) -> dict: + async with httpx.AsyncClient() as c: + resp = await c.post( + ATLASSIAN_TOKEN_URL, + json={ + "grant_type": "refresh_token", + "client_id": client_id, + "client_secret": client_secret, + "refresh_token": refresh_token, + }, + ) + resp.raise_for_status() + return resp.json() diff --git a/backend/package/yuxi/channel/extensions/confluence/config.py b/backend/package/yuxi/channel/extensions/confluence/config.py new file mode 100644 index 00000000..df25df37 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/config.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from yuxi.channel.extensions.confluence.types import ConfluenceAccount + +ENV_MAP = { + "url": "CONFLUENCE_URL", + "email": "CONFLUENCE_EMAIL", + "api_token": "CONFLUENCE_API_TOKEN", + "oauth_client_id": "CONFLUENCE_OAUTH_CLIENT_ID", + "oauth_client_secret": "CONFLUENCE_OAUTH_CLIENT_SECRET", + "auth_mode": "CONFLUENCE_AUTH_MODE", + "deployment_type": "CONFLUENCE_DEPLOYMENT_TYPE", + "space_keys": "CONFLUENCE_SPACE_KEYS", + "comment_poll_enabled": "CONFLUENCE_COMMENT_POLL_ENABLED", + "poll_interval": "CONFLUENCE_POLL_INTERVAL", + "comment_policy": "CONFLUENCE_COMMENT_POLICY", + "allow_from": "CONFLUENCE_ALLOW_FROM", +} + + +def _env_or_config(key: str) -> str | None: + env_key = ENV_MAP.get(key, "") + if env_key: + return os.getenv(env_key) + return None + + +def list_confluence_account_ids(config: dict) -> list[str]: + accounts = config.get("channels", {}).get("confluence", {}).get("accounts", {}) + if accounts: + return list(accounts.keys()) + return ["default"] + + +def resolve_confluence_account(config: dict, account_id: str = "default") -> ConfluenceAccount: + from yuxi.channel.extensions.confluence.types import ConfluenceAccount + + confluence_cfg = config.get("channels", {}).get("confluence", {}) + account_cfg = confluence_cfg.get("accounts", {}).get(account_id, confluence_cfg) + + def _get(key: str) -> str | None: + val = account_cfg.get(key) + if val is not None: + return str(val) + return _env_or_config(key) + + space_keys_str = _get("space_keys") or "" + allow_from_str = _get("allow_from") or "" + poll_enabled_str = _get("comment_poll_enabled") or "true" + + return ConfluenceAccount( + account_id=account_id, + url=_get("url") or "", + email=_get("email") or "", + api_token=_get("api_token") or "", + oauth_client_id=_get("oauth_client_id") or "", + oauth_client_secret=_get("oauth_client_secret") or "", + auth_mode=_get("auth_mode") or "api_token", + deployment_type=_get("deployment_type") or "cloud", + space_keys=[k.strip() for k in space_keys_str.split(",") if k.strip()], + comment_poll_enabled=poll_enabled_str.lower() != "false", + poll_interval=float(_get("poll_interval") or "30"), + comment_policy=_get("comment_policy") or "open", + allow_from=[a.strip() for a in allow_from_str.split(",") if a.strip()], + name=account_cfg.get("name", "Confluence Bot"), + ) diff --git a/backend/package/yuxi/channel/extensions/confluence/content_properties.py b/backend/package/yuxi/channel/extensions/confluence/content_properties.py new file mode 100644 index 00000000..6a13ef5a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/content_properties.py @@ -0,0 +1,91 @@ +import logging +from datetime import datetime, UTC + +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway + +logger = logging.getLogger(__name__) + + +class ContentPropertyStore: + NAMESPACE = "forcepilot_ai" + + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def _get_properties(self, page_id: str, account_id: str = "default") -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + try: + result = await client._request("GET", f"/wiki/api/v2/pages/{page_id}/properties") + return result.get("results", []) + except Exception as e: + logger.error("Failed to get properties for page %s: %s", page_id, e) + return [] + + async def _set( + self, page_id: str, key: str, value: dict, account_id: str = "default", + ) -> dict | None: + client = self._gateway.get_client(account_id) + if client is None: + return None + try: + return await client._request( + "POST", + f"/wiki/api/v2/pages/{page_id}/properties", + json={"key": key, "value": value}, + ) + except Exception as e: + logger.error("Failed to set property '%s' for page %s: %s", key, page_id, e) + return None + + async def _delete( + self, page_id: str, property_id: str, account_id: str = "default", + ) -> bool: + client = self._gateway.get_client(account_id) + if client is None: + return False + try: + await client._request( + "DELETE", f"/wiki/api/v2/pages/{page_id}/properties/{property_id}", + ) + return True + except Exception as e: + logger.error("Failed to delete property %s from page %s: %s", property_id, page_id, e) + return False + + async def mark_processed(self, page_id: str, processor: str, account_id: str = "default") -> dict | None: + return await self._set( + page_id, + f"{self.NAMESPACE}.{processor}.processed_at", + {"timestamp": datetime.now(tz=UTC).isoformat()}, + account_id, + ) + + async def store_summary(self, page_id: str, summary: str, account_id: str = "default") -> dict | None: + return await self._set( + page_id, + f"{self.NAMESPACE}.summary", + {"text": summary}, + account_id, + ) + + async def get_ai_metadata(self, page_id: str, account_id: str = "default") -> dict: + properties = await self._get_properties(page_id, account_id) + metadata = {} + for prop in properties: + key = prop.get("key", "") + if key.startswith(self.NAMESPACE): + metadata[key] = prop.get("value", {}) + return metadata + + async def clear_ai_metadata(self, page_id: str, account_id: str = "default") -> bool: + properties = await self._get_properties(page_id, account_id) + success = True + for prop in properties: + key = prop.get("key", "") + prop_id = prop.get("id") + if key.startswith(self.NAMESPACE) and prop_id: + if not await self._delete(page_id, prop_id, account_id): + success = False + return success diff --git a/backend/package/yuxi/channel/extensions/confluence/dedupe.py b/backend/package/yuxi/channel/extensions/confluence/dedupe.py new file mode 100644 index 00000000..32b796cf --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/dedupe.py @@ -0,0 +1,34 @@ +import time + + +class CommentDeduplicator: + def __init__(self, max_size: int = 50_000, ttl_seconds: int = 600): + self._max_size = max_size + self._ttl = ttl_seconds + self._cache: dict[str, float] = {} + + def is_duplicate(self, comment_id: str) -> bool: + self._evict_expired() + return comment_id in self._cache + + def mark(self, comment_id: str): + self._evict_expired() + self._cache[comment_id] = time.time() + self._evict_oldest_if_needed() + + def _evict_expired(self): + now = time.time() + expired = [cid for cid, ts in self._cache.items() if now - ts > self._ttl] + for cid in expired: + del self._cache[cid] + + def _evict_oldest_if_needed(self): + while len(self._cache) > self._max_size: + oldest = min(self._cache, key=self._cache.get) + del self._cache[oldest] + + def clear(self): + self._cache.clear() + + def __len__(self) -> int: + return len(self._cache) diff --git a/backend/package/yuxi/channel/extensions/confluence/format.py b/backend/package/yuxi/channel/extensions/confluence/format.py new file mode 100644 index 00000000..980ddbe2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/format.py @@ -0,0 +1,426 @@ +import json +import re +from datetime import datetime + + +class ADFBuilder: + PANEL_TYPES = {"info", "note", "success", "warning", "error", "tip"} + STATUS_COLORS = {"neutral", "purple", "blue", "red", "yellow", "green"} + + @staticmethod + def doc(blocks: list[dict]) -> str: + return json.dumps( + { + "version": 1, + "type": "doc", + "content": blocks, + }, + ensure_ascii=False, + ) + + @classmethod + def paragraph(cls, text: str) -> dict: + return { + "type": "paragraph", + "content": cls._parse_inline(text), + } + + @classmethod + def heading(cls, text: str, level: int = 3) -> dict: + return { + "type": "heading", + "attrs": {"level": min(max(level, 1), 6)}, + "content": cls._parse_inline(text), + } + + @staticmethod + def code_block(code: str, language: str = "python") -> dict: + return { + "type": "codeBlock", + "attrs": {"language": language}, + "content": [{"type": "text", "text": code}], + } + + @classmethod + def bullet_list(cls, items: list[str]) -> dict: + return { + "type": "bulletList", + "content": [ + { + "type": "listItem", + "content": [cls.paragraph(item)], + } + for item in items + ], + } + + @classmethod + def ordered_list(cls, items: list[str]) -> dict: + return { + "type": "orderedList", + "content": [ + { + "type": "listItem", + "content": [cls.paragraph(item)], + } + for item in items + ], + } + + @staticmethod + def mention(account_id: str, display_name: str) -> dict: + return { + "type": "mention", + "attrs": { + "id": account_id, + "text": f"@{display_name}", + "accessLevel": "", + "userType": "DEFAULT", + }, + } + + @staticmethod + def link(url: str, text: str | None = None) -> dict: + return { + "type": "text", + "text": text or url, + "marks": [{"type": "link", "attrs": {"href": url}}], + } + + @staticmethod + def text_with_marks(text: str, marks: list[dict] | None = None) -> dict: + node: dict = {"type": "text", "text": text} + if marks: + node["marks"] = marks + return node + + @staticmethod + def table(headers: list[str], rows: list[list[str]]) -> dict: + return { + "type": "table", + "attrs": {"isNumberColumnEnabled": False, "layout": "default"}, + "content": [ + { + "type": "tableRow", + "content": [ + { + "type": "tableHeader", + "attrs": {}, + "content": [{"type": "paragraph", "content": [{"type": "text", "text": h}]}], + } + for h in headers + ], + } + ] + + [ + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {}, + "content": [{"type": "paragraph", "content": [{"type": "text", "text": cell}]}], + } + for cell in row + ], + } + for row in rows + ], + } + + @classmethod + def panel(cls, content: list[dict], panel_type: str = "info") -> dict: + return { + "type": "panel", + "attrs": {"panelType": panel_type if panel_type in cls.PANEL_TYPES else "info"}, + "content": content, + } + + @classmethod + def info_panel(cls, text: str) -> dict: + return cls.panel( + [{"type": "paragraph", "content": [{"type": "text", "text": text}]}], + "info", + ) + + @staticmethod + def expand(title: str, content: list[dict]) -> dict: + return { + "type": "expand", + "attrs": {"title": title}, + "content": content, + } + + @staticmethod + def task_list(tasks: list[dict]) -> dict: + return { + "type": "taskList", + "attrs": {}, + "content": [ + { + "type": "taskItem", + "attrs": { + "localId": task.get("id", f"task-{i}"), + "state": task.get("state", "TODO"), + }, + "content": [ + {"type": "text", "text": task.get("text", "")}, + ], + } + for i, task in enumerate(tasks) + ], + } + + @staticmethod + def blockquote(content: list[dict]) -> dict: + return {"type": "blockquote", "content": content} + + @staticmethod + def rule() -> dict: + return {"type": "rule"} + + @classmethod + def status(cls, text: str, color: str = "neutral") -> dict: + return { + "type": "status", + "attrs": { + "text": text, + "color": color if color in cls.STATUS_COLORS else "neutral", + }, + } + + @staticmethod + def date(iso_date: str) -> dict: + ts = int(datetime.fromisoformat(iso_date).timestamp() * 1000) + return { + "type": "date", + "attrs": {"timestamp": ts}, + } + + @staticmethod + def media(media_id: str, media_type: str = "file", collection: str = "") -> dict: + return { + "type": "mediaSingle", + "attrs": {"layout": "center"}, + "content": [ + { + "type": "media", + "attrs": { + "type": media_type, + "id": media_id, + "collection": collection, + }, + } + ], + } + + @staticmethod + def emoji(short_name: str, id: str | None = None, text: str | None = None) -> dict: + attrs: dict = {"shortName": short_name} + if id: + attrs["id"] = id + if text: + attrs["text"] = text + return { + "type": "emoji", + "attrs": attrs, + } + + @staticmethod + def macro(name: str, params: dict | None = None) -> dict: + macro_params: dict = {} + if params: + for key, value in params.items(): + macro_params[key] = {"value": value} + return { + "type": "extension", + "attrs": { + "extensionType": "com.atlassian.confluence.macro.core", + "extensionKey": name, + "parameters": {"macroParams": macro_params}, + }, + } + + @staticmethod + def layout_section(columns: list[list[dict]], column_widths: list[int] | None = None) -> dict: + if column_widths is None: + column_widths = [100 // len(columns)] * len(columns) + return { + "type": "layoutSection", + "attrs": {"layoutType": "two_equal" if len(columns) == 2 else "default"}, + "content": [ + { + "type": "layoutColumn", + "attrs": {"width": column_widths[i]}, + "content": col, + } + for i, col in enumerate(columns) + ], + } + + @staticmethod + def _parse_inline(text: str) -> list[dict]: + pattern = re.compile( + r'(\*\*(.+?)\*\*|\*(.+?)\*|`(.+?)`|\[(.+?)\]\((.+?)\))' + ) + nodes = [] + last_end = 0 + for m in pattern.finditer(text): + if m.start() > last_end: + plain = text[last_end:m.start()] + if plain: + nodes.append({"type": "text", "text": plain}) + last_end = m.end() + + if m.group(2): + nodes.append({"type": "text", "text": m.group(2), "marks": [{"type": "strong"}]}) + elif m.group(3): + nodes.append({"type": "text", "text": m.group(3), "marks": [{"type": "em"}]}) + elif m.group(4): + nodes.append({"type": "text", "text": m.group(4), "marks": [{"type": "code"}]}) + elif m.group(5) and m.group(6): + nodes.append({ + "type": "text", + "text": m.group(5), + "marks": [{"type": "link", "attrs": {"href": m.group(6)}}], + }) + + if last_end < len(text): + nodes.append({"type": "text", "text": text[last_end:]}) + + return nodes or [{"type": "text", "text": text}] + + @classmethod + def from_markdown(cls, md_text: str) -> str: + blocks = [] + lines = md_text.split("\n") + i = 0 + + while i < len(lines): + stripped = lines[i].strip() + + if not stripped: + i += 1 + continue + + heading_match = re.match(r"^(#{1,6})\s+(.+)", stripped) + if heading_match: + level = len(heading_match.group(1)) + blocks.append(cls.heading(heading_match.group(2), level)) + i += 1 + continue + + if stripped.startswith("```"): + lang = stripped[3:].strip() or "text" + code_lines = [] + i += 1 + while i < len(lines) and not lines[i].strip().startswith("```"): + code_lines.append(lines[i]) + i += 1 + i += 1 + blocks.append(cls.code_block("\n".join(code_lines), lang)) + continue + + if stripped.startswith("- ") or stripped.startswith("* "): + items = [] + while i < len(lines) and (lines[i].strip().startswith("- ") or lines[i].strip().startswith("* ")): + items.append(lines[i].strip()[2:]) + i += 1 + blocks.append(cls.bullet_list(items)) + continue + + ordered_match = re.match(r"^(\d+)\.\s+(.+)", stripped) + if ordered_match: + items = [] + while i < len(lines): + m = re.match(r"^(\d+)\.\s+(.+)", lines[i].strip()) + if not m: + break + items.append(m.group(2)) + i += 1 + blocks.append(cls.ordered_list(items)) + continue + + if stripped.startswith("> "): + quote_lines = [] + while i < len(lines) and lines[i].strip().startswith("> "): + quote_lines.append(lines[i].strip()[2:]) + i += 1 + blocks.append(cls.blockquote([cls.paragraph("\n".join(quote_lines))])) + continue + + if stripped.startswith("---") or stripped.startswith("***"): + blocks.append(cls.rule()) + i += 1 + continue + + table_match = re.match(r"^\|(.+)\|$", stripped) + if table_match and i + 1 < len(lines) and re.match(r"^\|[\s\-:|]+\|$", lines[i + 1].strip()): + headers = [h.strip() for h in table_match.group(1).split("|") if h.strip() or h.strip() == ""] + rows = [] + i += 2 + while i < len(lines): + row_match = re.match(r"^\|(.+)\|$", lines[i].strip()) + if not row_match: + break + cells = [c.strip() for c in row_match.group(1).split("|")] + rows.append(cells) + i += 1 + blocks.append(cls.table(headers, rows)) + continue + + blocks.append(cls.paragraph(stripped)) + i += 1 + + return cls.doc(blocks) + + +def extract_text_from_adf(adf_json: str) -> str: + try: + doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json + except (json.JSONDecodeError, TypeError): + return "" + + parts = [] + + def _walk(node): + if isinstance(node, dict): + if node.get("type") == "text": + parts.append(node.get("text", "")) + elif node.get("type") == "mention": + parts.append(node.get("attrs", {}).get("text", "@unknown")) + elif node.get("type") == "hardBreak": + parts.append("\n") + for child in node.get("content", []): + _walk(child) + elif isinstance(node, list): + for item in node: + _walk(item) + + _walk(doc) + return "".join(parts) + + +ADF_REQUIRED_KEYS = {"version", "type", "content"} + + +def validate_adf(adf_json: str) -> tuple[bool, str]: + try: + doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json + except (json.JSONDecodeError, TypeError) as e: + return False, f"Invalid JSON: {e}" + + if not isinstance(doc, dict): + return False, "ADF root must be a dict" + + missing = ADF_REQUIRED_KEYS - set(doc.keys()) + if missing: + return False, f"Missing required keys: {missing}" + + if doc.get("type") != "doc": + return False, f"Root type must be 'doc', got '{doc.get('type')}'" + + if doc.get("version") != 1: + return False, f"ADF version must be 1, got {doc.get('version')}" + + return True, "OK" diff --git a/backend/package/yuxi/channel/extensions/confluence/gateway.py b/backend/package/yuxi/channel/extensions/confluence/gateway.py new file mode 100644 index 00000000..84c4164e --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/gateway.py @@ -0,0 +1,492 @@ +import asyncio +import logging +import re +import time +from collections.abc import Callable +from datetime import datetime, UTC + +import httpx + +from yuxi.channel.extensions.confluence.dedupe import CommentDeduplicator +from yuxi.channel.extensions.confluence.search import CQLBuilder +from yuxi.channel.extensions.confluence.types import ( + ConfluenceAccount, + InboundConfluenceComment, + PageContext, +) + +logger = logging.getLogger(__name__) + + +class ConfluenceClient: + def __init__(self, account: ConfluenceAccount): + self._account = account + self._http: httpx.AsyncClient | None = None + self._username = account.email + self._rate_limit_remaining = 65_000 + + async def __aenter__(self): + auth_header = self._account.auth_header + self._http = httpx.AsyncClient( + base_url=self._account.base_url, + headers={ + **auth_header, + "Accept": "application/json", + "Content-Type": "application/json", + }, + timeout=httpx.Timeout(30.0), + ) + return self + + async def __aexit__(self, *args): + if self._http: + await self._http.aclose() + + async def _request(self, method: str, path: str, **kwargs) -> dict: + if self._http is None: + raise RuntimeError("Client not initialized") + + if self._rate_limit_remaining < 100: + await asyncio.sleep(0.5) + elif self._rate_limit_remaining < 500: + await asyncio.sleep(0.1) + + resp = await self._http.request(method, path, **kwargs) + self._update_rate_limit(resp.headers) + + if resp.status_code == 429: + retry_after = int(resp.headers.get("Retry-After", "5")) + logger.warning("Confluence rate limited, retry after %ds", retry_after) + await asyncio.sleep(retry_after) + resp = await self._http.request(method, path, **kwargs) + self._update_rate_limit(resp.headers) + + resp.raise_for_status() + return resp.json() + + def _update_rate_limit(self, headers: dict): + remaining = headers.get("X-RateLimit-Remaining") + if remaining is not None: + self._rate_limit_remaining = int(remaining) + + async def cql_search(self, cql: str, limit: int = 50, expand: str | None = None) -> dict: + params = {"cql": cql, "limit": limit} + if expand: + params["expand"] = expand + return await self._request("GET", "/rest/api/search", params=params) + + async def create_footer_comment(self, payload: dict) -> dict: + return await self._request("POST", "/wiki/api/v2/footer-comments", json=payload) + + async def get_comment_children(self, comment_id: str, limit: int = 25) -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/footer-comments/{comment_id}/children", + params={"limit": limit}, + ) + + async def get_page(self, page_id: str, body_format: str = "atlas_doc_format") -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/pages/{page_id}", + params={"body-format": body_format}, + ) + + async def update_page(self, page_id: str, payload: dict) -> dict: + return await self._request("PUT", f"/wiki/api/v2/pages/{page_id}", json=payload) + + async def get_space(self, space_id: str) -> dict: + return await self._request("GET", f"/wiki/api/v2/spaces/{space_id}") + + async def create_page(self, payload: dict) -> dict: + return await self._request("POST", "/wiki/api/v2/pages", json=payload) + + async def update_footer_comment(self, comment_id: str, adf_body: str) -> dict: + return await self._request( + "PUT", + f"/wiki/api/v2/footer-comments/{comment_id}", + json={ + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + }, + ) + + async def delete_footer_comment(self, comment_id: str) -> None: + await self._request("DELETE", f"/wiki/api/v2/footer-comments/{comment_id}") + + async def get_footer_comments( + self, + page_id: str, + cursor: str | None = None, + limit: int = 25, + sort: str = "created-date", + ) -> dict: + params = {"limit": limit, "sort": sort} + if cursor: + params["cursor"] = cursor + return await self._request( + "GET", + f"/wiki/api/v2/pages/{page_id}/footer-comments", + params=params, + ) + + async def get_comment(self, comment_id: str) -> dict: + return await self._request("GET", f"/wiki/api/v2/footer-comments/{comment_id}") + + async def delete_page(self, page_id: str, purge: bool = False) -> None: + params = {"purge": str(purge).lower()} + await self._request("DELETE", f"/wiki/api/v2/pages/{page_id}", params=params) + + async def get_space_pages( + self, + space_id: str, + cursor: str | None = None, + limit: int = 25, + depth: str = "all", + ) -> dict: + params = {"limit": limit, "depth": depth} + if cursor: + params["cursor"] = cursor + return await self._request( + "GET", + f"/wiki/api/v2/spaces/{space_id}/pages", + params=params, + ) + + async def get_spaces(self, limit: int = 10, cursor: str | None = None) -> dict: + params = {"limit": limit} + if cursor: + params["cursor"] = cursor + return await self._request("GET", "/wiki/api/v2/spaces", params=params) + + async def get_users( + self, query: str | None = None, limit: int = 25, + ) -> dict: + params = {"limit": limit} + if query: + params["query"] = query + return await self._request("GET", "/wiki/api/v2/users", params=params) + + async def create_space(self, key: str, name: str, description: str = "") -> dict: + payload = { + "key": key, + "name": name, + } + if description: + payload["description"] = {"plain": {"value": description}} + return await self._request("POST", "/wiki/api/v2/spaces", json=payload) + + async def delete_space(self, space_id: str) -> None: + await self._request("DELETE", f"/wiki/api/v2/spaces/{space_id}") + + async def get_page_children(self, page_id: str, limit: int = 25) -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/pages/{page_id}/children", + params={"limit": limit}, + ) + + async def get_page_ancestors(self, page_id: str) -> dict: + return await self._request("GET", f"/wiki/api/v2/pages/{page_id}/ancestors") + + async def get_page_versions(self, page_id: str, limit: int = 25) -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/pages/{page_id}/versions", + params={"limit": limit}, + ) + + async def list_all_footer_comments( + self, + cursor: str | None = None, + limit: int = 25, + sort: str = "created-date", + ) -> dict: + params = {"limit": limit, "sort": sort} + if cursor: + params["cursor"] = cursor + return await self._request("GET", "/wiki/api/v2/footer-comments", params=params) + + async def update_page_title(self, page_id: str, title: str) -> dict: + return await self._request( + "PUT", + f"/wiki/api/v2/pages/{page_id}/title", + json={"title": title}, + ) + + async def update_space(self, space_id: str, name: str, description: str = "") -> dict: + payload: dict = {"name": name} + if description: + payload["description"] = {"plain": {"value": description}} + return await self._request( + "PUT", f"/wiki/api/v2/spaces/{space_id}", json=payload, + ) + + async def get_blogpost(self, blogpost_id: str, body_format: str = "atlas_doc_format") -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/blogposts/{blogpost_id}", + params={"body-format": body_format}, + ) + + async def create_blogpost(self, payload: dict) -> dict: + return await self._request("POST", "/wiki/api/v2/blogposts", json=payload) + + async def update_blogpost(self, blogpost_id: str, payload: dict) -> dict: + return await self._request("PUT", f"/wiki/api/v2/blogposts/{blogpost_id}", json=payload) + + async def delete_blogpost(self, blogpost_id: str) -> None: + await self._request("DELETE", f"/wiki/api/v2/blogposts/{blogpost_id}") + + async def get_tasks(self, page_id: str, limit: int = 25) -> dict: + return await self._request( + "GET", + f"/wiki/api/v2/pages/{page_id}/tasks", + params={"limit": limit}, + ) + + async def get_task(self, task_id: str) -> dict: + return await self._request("GET", f"/wiki/api/v2/tasks/{task_id}") + + async def update_task(self, task_id: str, state: str = "COMPLETE") -> dict: + return await self._request( + "PUT", + f"/wiki/api/v2/tasks/{task_id}", + json={"state": state}, + ) + + async def _paginated_request( + self, + method: str, + path: str, + params: dict | None = None, + max_results: int = 250, + ) -> list[dict]: + cursor = None + all_results: list[dict] = [] + params = params or {} + params.setdefault("limit", 25) + + while len(all_results) < max_results: + if cursor: + params["cursor"] = cursor + + if self._http is None: + raise RuntimeError("Client not initialized") + resp = await self._http.request(method, path, params=params) + self._update_rate_limit(resp.headers) + resp.raise_for_status() + + page = resp.json() + all_results.extend(page.get("results", [])) + + links = resp.headers.get("Link", "") + cursor = self._extract_next_cursor(links) + if not cursor: + break + + return all_results[:max_results] + + @staticmethod + def _extract_next_cursor(link_header: str) -> str | None: + match = re.search(r'<[^>]*cursor=([^>&]+)[^>]*>\s*;\s*rel="next"', link_header) + if match: + return match.group(1) + return None + + +class ConfluenceGateway: + MAX_RETRY_BACKOFF = 300.0 + MAX_SEEN_COMMENTS = 50_000 + + def __init__(self): + self._active_clients: dict[str, ConfluenceClient] = {} + self._poll_tasks: dict[str, asyncio.Task] = {} + self._last_poll_time: dict[str, float] = {} + self._deduplicator = CommentDeduplicator(max_size=50_000, ttl_seconds=600) + self._cancel_events: dict[str, asyncio.Event] = {} + + async def start( + self, + account_id: str, + account: ConfluenceAccount, + on_comment: Callable[[InboundConfluenceComment], None], + authorize_sender: Callable[[str, dict], bool], + ) -> ConfluenceClient: + client = ConfluenceClient(account) + await client.__aenter__() + self._active_clients[account_id] = client + + if account.comment_poll_enabled: + cancel = asyncio.Event() + self._cancel_events[account_id] = cancel + task = asyncio.create_task( + self._poll_comments_loop(account_id, account, client, on_comment, authorize_sender, cancel) + ) + self._poll_tasks[account_id] = task + + return client + + async def stop(self, account_id: str): + cancel = self._cancel_events.pop(account_id, None) + if cancel: + cancel.set() + task = self._poll_tasks.pop(account_id, None) + if task: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + client = self._active_clients.pop(account_id, None) + if client: + await client.__aexit__(None, None, None) + + async def stop_all(self): + for account_id in list(self._active_clients.keys()): + await self.stop(account_id) + + def get_client(self, account_id: str = "default") -> ConfluenceClient | None: + return self._active_clients.get(account_id) + + async def _poll_comments_loop( + self, + account_id: str, + account: ConfluenceAccount, + client: ConfluenceClient, + on_comment: Callable, + authorize_sender: Callable, + cancel: asyncio.Event, + ): + error_count = 0 + last_poll = self._last_poll_time.get(account_id, time.time() - 3600) + seen_in_session: set[str] = set() + + while not cancel.is_set(): + try: + since_iso = datetime.fromtimestamp(last_poll, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z") + + cql = CQLBuilder.comment_poll(since_iso, account.space_keys) + results = await client.cql_search(cql, limit=50, expand="body.view,history.lastUpdated,container") + new_max_time = last_poll + + for result in results.get("results", []): + content = result.get("content", {}) + comment_id = content.get("id", "") + + if not comment_id: + continue + if comment_id in seen_in_session: + continue + if self._deduplicator.is_duplicate(comment_id): + continue + seen_in_session.add(comment_id) + self._deduplicator.mark(comment_id) + + author_info = content.get("history", {}).get("createdBy", {}) + author_id = author_info.get("accountId", "") + + if self._is_self_comment(client, author_info): + continue + + if not authorize_sender(author_id, result): + continue + + page_ctx = self._extract_page_context(result) + if page_ctx is None: + continue + + comment_text = self._extract_comment_text(result) + + inbound = InboundConfluenceComment( + comment_id=comment_id, + author_id=author_id, + author_name=author_info.get("displayName", ""), + body_text=comment_text, + container_id=page_ctx.page_id, + parent_comment_id=self._extract_parent_comment_id(result), + page_context=page_ctx, + created_at=self._parse_datetime(content.get("history", {}).get("createdDate")), + ) + + on_comment(inbound) + + updated = content.get("history", {}).get("lastUpdated", {}).get("when", "") + updated_ts = self._parse_timestamp(updated) + if updated_ts and updated_ts > new_max_time: + new_max_time = updated_ts + + last_poll = new_max_time + self._last_poll_time[account_id] = new_max_time + error_count = 0 + + except asyncio.CancelledError: + break + except Exception as e: + error_count += 1 + backoff = min(2.0 * (2**error_count), self.MAX_RETRY_BACKOFF) + logger.error( + "Confluence poll error (attempt %d, retry in %.1fs): %s", + error_count, + backoff, + e, + ) + + try: + await asyncio.wait_for(cancel.wait(), timeout=account.poll_interval) + break + except TimeoutError: + pass + + def _is_self_comment(self, client: ConfluenceClient, author: dict) -> bool: + author_email = author.get("email", "") + return author_email and author_email.lower() == client._username.lower() + + def _extract_page_context(self, result: dict) -> PageContext | None: + container = result.get("resultParentContainer") + if not container: + return None + return PageContext( + page_id=container.get("id", ""), + page_title=container.get("title", ""), + space_key=container.get("space", {}).get("key", ""), + ) + + def _extract_comment_text(self, result: dict) -> str: + body = result.get("content", {}).get("body", {}) + view_value = body.get("view", {}).get("value", "") + return self._strip_html(view_value) + + def _extract_parent_comment_id(self, result: dict) -> str | None: + ancestors = result.get("content", {}).get("ancestors", []) + for ancestor in reversed(ancestors): + if ancestor.get("type") == "comment": + return ancestor.get("id") + return None + + @staticmethod + def _strip_html(html: str) -> str: + clean = re.sub(r"<[^>]+>", "", html) + clean = re.sub(r"&[a-z]+;", " ", clean) + return re.sub(r"\s+", " ", clean).strip() + + @staticmethod + def _parse_timestamp(iso_str: str | None) -> float: + if not iso_str: + return 0.0 + try: + dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + return dt.timestamp() + except (ValueError, TypeError): + return 0.0 + + @staticmethod + def _parse_datetime(iso_str: str | None) -> datetime | None: + if not iso_str: + return None + try: + return datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + except (ValueError, TypeError): + return None diff --git a/backend/package/yuxi/channel/extensions/confluence/knowledge.py b/backend/package/yuxi/channel/extensions/confluence/knowledge.py new file mode 100644 index 00000000..af2bd8f4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/knowledge.py @@ -0,0 +1,68 @@ +import logging + +from yuxi.channel.extensions.confluence.format import extract_text_from_adf +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.search import CQLBuilder + +logger = logging.getLogger(__name__) + + +class ConfluenceKnowledge: + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def retrieve_context( + self, + query: str, + account_id: str = "default", + space_keys: list[str] | None = None, + max_pages: int = 5, + ) -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + + cql = CQLBuilder.search_pages(query, space_keys) + + try: + results = await client.cql_search(cql, limit=max_pages) + except Exception as e: + logger.error("CQL search failed for context retrieval '%s': %s", query, e) + return [] + + contexts = [] + for item in results.get("results", []): + content = item.get("content", {}) + page_id = content.get("id") + title = content.get("title", "") + body = content.get("body", {}) + adf_value = body.get("view", {}).get("value", "") + text = extract_text_from_adf(adf_value) if adf_value else "" + + contexts.append({ + "page_id": page_id, + "title": title, + "text": text[:2000], + "url": content.get("_links", {}).get("base", ""), + }) + + return contexts + + async def build_context_note( + self, + page_context, + account_id: str = "default", + ) -> str: + if page_context is None: + return "" + + client = self._gateway.get_client(account_id) + if client is None: + return "" + + try: + page = await client.get_page(page_context.page_id) + title = page.get("title", "") + return f"当前页面: [{page_context.space_key}] {title}" + except Exception: + return f"当前页面: {page_context.context_label}" \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/labels.py b/backend/package/yuxi/channel/extensions/confluence/labels.py new file mode 100644 index 00000000..6d9da4a1 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/labels.py @@ -0,0 +1,69 @@ +import logging + +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.types import OutboundResult + +logger = logging.getLogger(__name__) + + +class ConfluenceLabelManager: + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def get_page_labels(self, page_id: str, account_id: str = "default") -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + try: + result = await client._request("GET", f"/wiki/api/v2/pages/{page_id}/labels") + return result.get("results", []) + except Exception as e: + logger.error("Failed to get labels for page %s: %s", page_id, e) + return [] + + async def add_page_label( + self, page_id: str, label: str, prefix: str = "global", account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + try: + result = await client._request( + "POST", + f"/wiki/api/v2/pages/{page_id}/labels", + json={"prefix": prefix, "name": label}, + ) + return OutboundResult(success=True, comment_id=result.get("id", "")) + except Exception as e: + logger.error("Failed to add label '%s' to page %s: %s", label, page_id, e) + return OutboundResult(success=False, error=str(e)) + + async def delete_page_label( + self, page_id: str, label_id: str, account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + try: + await client._request( + "DELETE", f"/wiki/api/v2/pages/{page_id}/labels/{label_id}", + ) + return OutboundResult(success=True, comment_id=label_id) + except Exception as e: + logger.error("Failed to delete label %s from page %s: %s", label_id, page_id, e) + return OutboundResult(success=False, error=str(e)) + + async def get_label_pages( + self, label_id: str, limit: int = 25, account_id: str = "default", + ) -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + try: + result = await client._request( + "GET", f"/wiki/api/v2/labels/{label_id}/pages", params={"limit": limit}, + ) + return result.get("results", []) + except Exception as e: + logger.error("Failed to get pages for label %s: %s", label_id, e) + return [] diff --git a/backend/package/yuxi/channel/extensions/confluence/mentions.py b/backend/package/yuxi/channel/extensions/confluence/mentions.py new file mode 100644 index 00000000..65a19975 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/mentions.py @@ -0,0 +1,38 @@ +import json + + +class ConfluenceMentions: + @staticmethod + def extract_mentions(raw_message: dict) -> list[dict]: + body = raw_message.get("body", {}) + adf_raw = body.get("atlas_doc_format", {}).get("value", "") + + try: + doc = json.loads(adf_raw) if isinstance(adf_raw, str) else adf_raw + except (json.JSONDecodeError, TypeError): + return [] + + mentions = [] + + def _walk(node): + if isinstance(node, dict): + if node.get("type") == "mention": + attrs = node.get("attrs", {}) + mentions.append({ + "account_id": attrs.get("id", ""), + "display_name": attrs.get("text", "").lstrip("@"), + }) + for child in node.get("content", []): + _walk(child) + elif isinstance(node, list): + for item in node: + _walk(item) + + _walk(doc) + return mentions + + @staticmethod + def build_mention(account_id: str, display_name: str) -> dict: + from yuxi.channel.extensions.confluence.format import ADFBuilder + + return ADFBuilder.mention(account_id, display_name) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/outbound.py b/backend/package/yuxi/channel/extensions/confluence/outbound.py new file mode 100644 index 00000000..a665cdac --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/outbound.py @@ -0,0 +1,280 @@ +import asyncio +import logging + +from yuxi.channel.extensions.confluence.format import ADFBuilder +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.search import CQLBuilder +from yuxi.channel.extensions.confluence.types import OutboundResult + +logger = logging.getLogger(__name__) + +MAX_ADF_BLOCK_CHARS = 2000 +MAX_RETRY = 3 +RETRY_BASE_DELAY = 1.0 + + +class ConfluenceOutbound: + delivery_mode = "comment" + + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def reply_to_comment( + self, + page_id: str, + parent_comment_id: str, + text: str, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + adf_body = ADFBuilder.from_markdown(text) + + payload = { + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + "containerId": page_id, + "parentCommentId": parent_comment_id, + } + + return await self._send_with_retry(client, "POST", "/wiki/api/v2/footer-comments", payload) + + async def create_comment( + self, + page_id: str, + text: str, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + adf_body = ADFBuilder.from_markdown(text) + + payload = { + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + "containerId": page_id, + } + + return await self._send_with_retry(client, "POST", "/wiki/api/v2/footer-comments", payload) + + async def edit_comment( + self, + comment_id: str, + text: str, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + adf_body = ADFBuilder.from_markdown(text) + return await self._send_with_retry( + client, "PUT", f"/wiki/api/v2/footer-comments/{comment_id}", + {"body": {"representation": "atlas_doc_format", "value": adf_body}}, + ) + + async def delete_comment( + self, + comment_id: str, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + try: + await client.delete_footer_comment(comment_id) + return OutboundResult(success=True, comment_id=comment_id) + except Exception as e: + return OutboundResult(success=False, error=str(e)) + + async def get_comment_thread( + self, + comment_id: str, + account_id: str = "default", + max_depth: int = 3, + ) -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + + results = [] + children = await client.get_comment_children(comment_id, limit=25) + for child in children.get("results", []): + results.append(child) + if max_depth > 1: + deeper = await self.get_comment_thread( + child["id"], account_id, max_depth - 1, + ) + results.extend(deeper) + return results + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str = "default", + ) -> OutboundResult: + page_id = target_id + parent_id = reply_to_id or thread_id + + if parent_id: + return await self.reply_to_comment(page_id, parent_id, content, account_id) + return await self.create_comment(page_id, content, account_id) + + async def read_page( + self, page_id: str, account_id: str = "default" + ) -> dict | None: + client = self._gateway.get_client(account_id) + if client is None: + return None + return await client.get_page(page_id) + + async def update_page( + self, + page_id: str, + title: str, + content: str, + space_id: str, + current_version: int, + version_comment: str = "AI 自动更新", + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + adf_body = ADFBuilder.from_markdown(content) + + payload = { + "id": page_id, + "status": "current", + "title": title, + "spaceId": space_id, + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + "version": { + "number": current_version + 1, + "message": version_comment, + }, + } + + return await self._send_with_retry(client, "PUT", f"/wiki/api/v2/pages/{page_id}", payload) + + async def update_page_with_conflict_retry( + self, + page_id: str, + title: str, + content: str, + space_id: str, + version_comment: str = "AI 自动更新", + account_id: str = "default", + max_retries: int = 3, + ) -> OutboundResult: + for attempt in range(max_retries): + page = await self.read_page(page_id, account_id) + if page is None: + return OutboundResult(success=False, error="Page not found") + + current_version = page.get("version", {}).get("number", 1) + current_title = page.get("title", title) + + result = await self.update_page( + page_id, current_title, content, space_id, + current_version, version_comment, account_id, + ) + + if result.success: + return result + + if "409" in result.error or "conflict" in result.error.lower(): + delay = RETRY_BASE_DELAY * (2 ** attempt) + logger.info( + "Page update conflict, retry %d/%d after %.1fs", + attempt + 1, max_retries, delay, + ) + await asyncio.sleep(delay) + continue + + return result + + return OutboundResult(success=False, error="Max retries exceeded for update_page") + + async def search_pages( + self, + query: str, + space_keys: list[str] | None = None, + limit: int = 10, + account_id: str = "default", + ) -> list[dict]: + client = self._gateway.get_client(account_id) + if client is None: + return [] + + cql = CQLBuilder.search_pages(query, space_keys) + + try: + results = await client.cql_search(cql, limit=limit, expand="body.view") + return results.get("results", []) + except Exception as e: + logger.error("CQL search failed for query '%s': %s", query, e) + return [] + + async def _send_with_retry( + self, + client, + method: str, + path: str, + payload: dict, + ) -> OutboundResult: + for attempt in range(MAX_RETRY): + try: + result = await client._request(method, path, json=payload) + comment_id = result.get("id", "") + return OutboundResult(success=True, comment_id=comment_id) + except Exception as e: + status_code = getattr(getattr(e, "response", None), "status_code", 0) + + if status_code == 409: + return OutboundResult(success=False, error=f"409 Conflict: {e}") + if status_code == 403: + return OutboundResult(success=False, error=f"403 Forbidden (permissions): {e}") + if attempt == MAX_RETRY - 1: + return OutboundResult(success=False, error=str(e)) + + delay = RETRY_BASE_DELAY * (2 ** attempt) + logger.warning( + "Outbound retry %d/%d after %.1fs: %s", + attempt + 1, MAX_RETRY, delay, e, + ) + await asyncio.sleep(delay) + + return OutboundResult(success=False, error="Unknown retry failure") + + def chunker(self, text: str, limit: int = 2000, ctx=None) -> list[str]: + chunks = [] + paragraphs = text.split("\n\n") + current = "" + + for para in paragraphs: + if len(current) + len(para) + 2 > limit and current: + chunks.append(current.rstrip()) + current = para + else: + current = f"{current}\n\n{para}" if current else para + + if current: + chunks.append(current.rstrip()) + return chunks \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/page_writer.py b/backend/package/yuxi/channel/extensions/confluence/page_writer.py new file mode 100644 index 00000000..bfbbbd8b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/page_writer.py @@ -0,0 +1,120 @@ +import asyncio +import logging + +from yuxi.channel.extensions.confluence.format import ADFBuilder +from yuxi.channel.extensions.confluence.gateway import ConfluenceGateway +from yuxi.channel.extensions.confluence.types import OutboundResult + +logger = logging.getLogger(__name__) + +MAX_VERSION_RETRY = 3 +RETRY_BASE_DELAY = 1.0 + + +class ConfluencePageWriter: + def __init__(self, gateway: ConfluenceGateway): + self._gateway = gateway + + async def create_page( + self, + space_id: str, + title: str, + content: str, + parent_page_id: str | None = None, + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + adf_body = ADFBuilder.from_markdown(content) + payload = { + "spaceId": space_id, + "status": "current", + "title": title, + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + } + if parent_page_id: + payload["parentId"] = parent_page_id + + try: + result = await client._request("POST", "/wiki/api/v2/pages", json=payload) + return OutboundResult(success=True, comment_id=result.get("id", "")) + except Exception as e: + return OutboundResult(success=False, error=str(e)) + + async def update_page_with_conflict_retry( + self, + page_id: str, + title: str, + content: str, + space_id: str, + version_comment: str = "AI 自动更新", + account_id: str = "default", + ) -> OutboundResult: + client = self._gateway.get_client(account_id) + if client is None: + return OutboundResult(success=False, error="Client not found") + + for attempt in range(MAX_VERSION_RETRY): + try: + page = await client.get_page(page_id) + except Exception as e: + return OutboundResult(success=False, error=f"Read page failed: {e}") + + current_version = page.get("version", {}).get("number", 1) + + adf_body = ADFBuilder.from_markdown(content) + payload = { + "id": page_id, + "status": "current", + "title": title, + "spaceId": space_id, + "body": { + "representation": "atlas_doc_format", + "value": adf_body, + }, + "version": { + "number": current_version + 1, + "message": version_comment, + }, + } + + try: + result = await client._request( + "PUT", f"/wiki/api/v2/pages/{page_id}", json=payload + ) + return OutboundResult(success=True, comment_id=result.get("id", "")) + except Exception as e: + status = getattr(getattr(e, "response", None), "status_code", 0) + if status == 409 and attempt < MAX_VERSION_RETRY - 1: + delay = RETRY_BASE_DELAY * (2 ** attempt) + logger.info( + "Page version conflict, retry %d/%d after %.1fs", + attempt + 1, MAX_VERSION_RETRY, delay, + ) + await asyncio.sleep(delay) + continue + return OutboundResult(success=False, error=str(e)) + + return OutboundResult(success=False, error="Max retries exceeded") + + async def generate_meeting_notes( + self, + space_id: str, + title: str, + transcript: str, + account_id: str = "default", + ) -> OutboundResult: + prompt = f"""请根据以下会议记录生成结构化的会议纪要： + +{transcript} + +格式要求： +- 标题使用会议纪要标题 +- 分节：参会人员、会议摘要、讨论要点、决议事项、待办事项 +- 每个待办事项标注负责人和截止时间""" + return await self.create_page(space_id, title, prompt, account_id=account_id) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/plugin.json b/backend/package/yuxi/channel/extensions/confluence/plugin.json new file mode 100644 index 00000000..f8c72b2a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/plugin.json @@ -0,0 +1,28 @@ +{ + "id": "confluence", + "name": "Confluence", + "version": "0.1.0", + "description": "Confluence 知识管理渠道插件，支持页面评论交互、知识库检索与 AI 内容生成", + "author": "ForcePilot Team", + "order": 90, + "dependencies": ["httpx"], + "capabilities": { + "chat_types": ["comment"], + "message_types": ["text"], + "reactions": false, + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "block_streaming_chunk_min_chars": 200, + "block_streaming_chunk_max_chars": 2000, + "block_streaming_chunk_break_preference": "paragraph", + "threads": true, + "reply": true, + "edit": true, + "unsend": true, + "media": false, + "polling_fallback": true + }, + "enabled": false, + "python_requires": ">=3.12" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/search.py b/backend/package/yuxi/channel/extensions/confluence/search.py new file mode 100644 index 00000000..f5d82a31 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/search.py @@ -0,0 +1,48 @@ +from datetime import datetime, UTC + + +class CQLBuilder: + @staticmethod + def comment_poll( + since_iso: str, + space_keys: list[str] | None = None, + ) -> str: + parts = ['type = "comment"'] + if space_keys: + space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys) + parts.append(f"({space_clause})") + parts.append(f'lastModified > "{since_iso}"') + parts.append("ORDER BY lastModified ASC") + return " AND ".join(parts) + + @staticmethod + def search_pages( + query: str, + space_keys: list[str] | None = None, + ) -> str: + parts = [ + f'(title ~ "{query}" OR text ~ "{query}")', + 'type = "page"', + ] + if space_keys: + space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys) + parts.append(f"({space_clause})") + return " AND ".join(parts) + + @staticmethod + def recent_pages( + space_keys: list[str] | None = None, + hours: int = 24, + ) -> str: + parts = [ + 'type = "page"', + f'lastModified > now("-{hours}h")', + ] + if space_keys: + space_clause = " OR ".join(f'space = "{sk}"' for sk in space_keys) + parts.append(f"({space_clause})") + return " AND ".join(parts) + + @staticmethod + def format_timestamp(ts: float) -> str: + return datetime.fromtimestamp(ts, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z") diff --git a/backend/package/yuxi/channel/extensions/confluence/security.py b/backend/package/yuxi/channel/extensions/confluence/security.py new file mode 100644 index 00000000..1374be4f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/security.py @@ -0,0 +1,37 @@ +import logging + +logger = logging.getLogger(__name__) + + +class ConfluenceCommentPolicy: + VALID_POLICIES = frozenset({"open", "pairing", "allowlist", "disabled"}) + + def __init__(self, policy: str = "open", allow_from: list[str] | None = None): + self._policy = policy if policy in self.VALID_POLICIES else "open" + self._allowlist: set[str] = set(allow_from or []) + + def is_allowed(self, author_id: str) -> bool: + if self._policy == "disabled": + return False + if self._policy == "open": + return True + if self._policy == "allowlist": + if not self._allowlist: + logger.warning("Comment policy is 'allowlist' but allowlist is empty") + return False + return author_id in self._allowlist + return True + + def add_to_allowlist(self, account_id: str): + self._allowlist.add(account_id) + + def remove_from_allowlist(self, account_id: str): + self._allowlist.discard(account_id) + + @property + def policy(self) -> str: + return self._policy + + @property + def allowlist(self) -> frozenset[str]: + return frozenset(self._allowlist) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/status.py b/backend/package/yuxi/channel/extensions/confluence/status.py new file mode 100644 index 00000000..64715d34 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/status.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from yuxi.channel.extensions.confluence.gateway import ConfluenceClient + from yuxi.channel.extensions.confluence.types import ConfluenceAccount + + +@dataclass +class ChannelAccountSnapshot: + account_id: str + configured: bool + status_state: str + running: bool + connected: bool + health_state: str + dm_policy: str + + +class ConfluenceStatus: + async def probe( + self, + account: ConfluenceAccount, + client: ConfluenceClient, + ) -> dict: + if not account.is_configured: + return {"success": False, "error": "Not configured"} + + try: + spaces = await client.get_spaces(limit=1) + total = spaces.get("meta", {}).get("total_count", -1) if "meta" in spaces else -1 + return {"success": True, "total_spaces": total} + except Exception as e: + return {"success": False, "error": str(e)} + + async def check_space_permissions( + self, + account: ConfluenceAccount, + client: ConfluenceClient, + ) -> list[dict]: + results = [] + for space_key in account.space_keys: + try: + cql = f'space = "{space_key}" AND type = "page"' + search = await client.cql_search(cql, limit=1) + results.append( + { + "space_key": space_key, + "accessible": len(search.get("results", [])) > 0, + "error": None, + } + ) + except Exception as e: + results.append( + { + "space_key": space_key, + "accessible": False, + "error": str(e), + } + ) + return results + + def build_account_snapshot( + self, + account: ConfluenceAccount, + running: bool, + probe_result: dict, + ) -> ChannelAccountSnapshot: + return ChannelAccountSnapshot( + account_id=account.account_id, + configured=account.is_configured, + status_state="linked" if running else "stopped", + running=running, + connected=probe_result.get("success", False), + health_state="ok" if probe_result.get("success") else "unknown", + dm_policy=account.comment_policy, + ) + + def collect_status_issues( + self, + account: ConfluenceAccount, + probe_result: dict, + running: bool, + ) -> list[str]: + issues = [] + if not account.is_configured: + issues.append("未配置：请设置 CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN") + if not probe_result.get("success"): + issues.append(f"API 连通性检查失败: {probe_result.get('error', 'Unknown error')}") + if not running: + issues.append("Gateway 未启动") + if account.space_keys and probe_result.get("success"): + space_list = ", ".join(account.space_keys) + issues.append(f"监听空间: {space_list}，请确认 Bot 用户对各空间有 View + Add Comments 权限") + return issues diff --git a/backend/package/yuxi/channel/extensions/confluence/streaming.py b/backend/package/yuxi/channel/extensions/confluence/streaming.py new file mode 100644 index 00000000..d7d3c396 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/streaming.py @@ -0,0 +1,30 @@ +class ConfluenceStreaming: + streaming_mode = "block" + block_streaming_enabled = True + MIN_CHARS = 200 + MAX_CHARS = 2000 + CHUNK_BREAK = "paragraph" + COALESCE_MIN = 80 + COALESCE_MAX = 400 + COALESCE_IDLE_MS = 500 + MAX_BLOCKS = 5 + FINISH_MARK = "..." + + @classmethod + def should_stream(cls, estimated_length: int) -> bool: + return estimated_length > cls.MIN_CHARS + + @classmethod + def build_chunk(cls, text: str, is_final: bool) -> str: + if is_final: + return text + return f"{text}\n\n{cls.FINISH_MARK}" + + @classmethod + def create_chunker(cls): + return { + "mode": "paragraph", + "min_chars": cls.COALESCE_MIN, + "max_chars": cls.MAX_CHARS, + "break_on": cls.CHUNK_BREAK, + } \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/confluence/types.py b/backend/package/yuxi/channel/extensions/confluence/types.py new file mode 100644 index 00000000..5fbf4966 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/types.py @@ -0,0 +1,94 @@ +from dataclasses import dataclass, field +from datetime import datetime + + +@dataclass +class ConfluenceAccount: + account_id: str = "default" + url: str = "" + email: str = "" + api_token: str = "" + oauth_client_id: str = "" + oauth_client_secret: str = "" + auth_mode: str = "api_token" + deployment_type: str = "cloud" + space_keys: list[str] = field(default_factory=list) + comment_poll_enabled: bool = True + poll_interval: float = 30.0 + comment_policy: str = "open" + allow_from: list[str] = field(default_factory=list) + enable_page_writer: bool = False + enable_knowledge_retrieval: bool = False + enable_webhook: bool = False + name: str = "" + + @property + def is_configured(self) -> bool: + if self.auth_mode == "api_token": + return bool(self.url and self.email and self.api_token) + return bool(self.url and self.oauth_client_id and self.oauth_client_secret) + + @property + def base_url(self) -> str: + return self.url.rstrip("/") + + @property + def api_v2_url(self) -> str: + return f"{self.base_url}/wiki/api/v2" + + @property + def rest_url(self) -> str: + return f"{self.base_url}/rest/api" + + @property + def auth_header(self) -> dict: + if self.auth_mode == "api_token": + import base64 + + credentials = base64.b64encode(f"{self.email}:{self.api_token}".encode()).decode() + return {"Authorization": f"Basic {credentials}"} + if self.auth_mode == "pat": + return {"Authorization": f"Bearer {self.api_token}"} + return {"Authorization": f"Bearer {self.oauth_client_secret}"} + + +@dataclass +class PageContext: + page_id: str + page_title: str = "" + space_key: str = "" + space_id: str = "" + page_url: str = "" + + @property + def context_label(self) -> str: + return f"[{self.space_key}] {self.page_title}" + + +@dataclass +class InboundConfluenceComment: + comment_id: str + author_id: str + author_name: str + body_text: str + body_adf: str | None = None + body_storage: str | None = None + container_type: str = "page" + container_id: str = "" + parent_comment_id: str | None = None + page_context: PageContext | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + version_number: int = 1 + + @property + def thread_id(self) -> str: + return self.parent_comment_id or self.comment_id + + +@dataclass +class OutboundResult: + success: bool + comment_id: str = "" + error: str = "" + detail: str = "" diff --git a/backend/package/yuxi/channel/extensions/confluence/webhook.py b/backend/package/yuxi/channel/extensions/confluence/webhook.py new file mode 100644 index 00000000..48d0c326 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/confluence/webhook.py @@ -0,0 +1,146 @@ +import json +import logging + +from yuxi.channel.extensions.confluence.types import InboundConfluenceComment, PageContext + +logger = logging.getLogger(__name__) + +SUPPORTED_EVENTS = { + "page_created", "page_updated", "page_removed", + "blog_created", "blog_updated", + "comment_created", "comment_updated", "comment_removed", + "attachment_created", "attachment_updated", "attachment_removed", + "label_added", "label_removed", + "page_moved", "page_restored", "page_trashed", + "content_created", "content_updated", "content_trashed", "content_restored", + "space_created", "space_updated", "space_removed", +} + + +class ConfluenceWebhookReceiver: + def __init__(self, on_page_changed=None, on_comment=None): + self._on_page_changed = on_page_changed + self._on_comment = on_comment + + def parse_webhook(self, body: bytes, headers: dict) -> dict | None: + try: + payload = json.loads(body) + except json.JSONDecodeError: + logger.warning("Invalid webhook JSON payload") + return None + + event_type = payload.get("eventType", "") + if event_type not in SUPPORTED_EVENTS: + logger.debug("Ignored webhook event type: %s", event_type) + return None + + if event_type.startswith("comment_"): + return self._parse_comment_event(payload) + + if event_type.startswith(("attachment_", "label_")): + return self._parse_resource_event(payload, event_type) + + page_id = payload.get("page", {}).get("id") + page_title = payload.get("page", {}).get("title", "") + space_key = payload.get("space", {}).get("key", "") + + if not page_id: + logger.warning("Webhook payload missing page.id") + return None + + return { + "event_type": event_type, + "page_id": page_id, + "page_title": page_title, + "space_key": space_key, + "user_account_id": payload.get("userAccountId", ""), + "timestamp": payload.get("timestamp", ""), + } + + def _parse_comment_event(self, payload: dict) -> dict | None: + comment = payload.get("comment", {}) + comment_id = comment.get("id") + if not comment_id: + return None + + container = payload.get("container") or payload.get("page") or {} + author = comment.get("createdBy", {}) or comment.get("lastUpdatedBy", {}) + + page_ctx = PageContext( + page_id=container.get("id", ""), + page_title=container.get("title", ""), + space_key=payload.get("space", {}).get("key", ""), + ) + + inbound = InboundConfluenceComment( + comment_id=comment_id, + author_id=author.get("accountId", ""), + author_name=author.get("displayName", ""), + body_text=self._extract_comment_body_text(comment), + container_id=container.get("id", ""), + parent_comment_id=comment.get("parentCommentId"), + page_context=page_ctx, + ) + + return { + "event_type": payload.get("eventType", ""), + "comment": inbound, + "timestamp": payload.get("timestamp", ""), + } + + @staticmethod + def _extract_comment_body_text(comment: dict) -> str: + body = comment.get("body", {}) + view_value = body.get("view", {}).get("value", "") + if view_value: + import re + clean = re.sub(r"<[^>]+>", "", view_value) + clean = re.sub(r"&[a-z]+;", " ", clean) + return re.sub(r"\s+", " ", clean).strip() + return "" + + def _parse_resource_event(self, payload: dict, event_type: str) -> dict | None: + if event_type.startswith("attachment_"): + resource = payload.get("attachment", {}) + return { + "event_type": event_type, + "attachment_id": resource.get("id"), + "filename": resource.get("title"), + "page_id": payload.get("page", {}).get("id"), + "timestamp": payload.get("timestamp", ""), + } + + if event_type.startswith("label_"): + return { + "event_type": event_type, + "label_name": payload.get("label", {}).get("name"), + "page_id": payload.get("page", {}).get("id"), + "timestamp": payload.get("timestamp", ""), + } + + return None + + async def handle_page_event(self, event: dict, account_id: str = "default"): + if self._on_page_changed is None: + return + + page_ctx = PageContext( + page_id=event["page_id"], + page_title=event["page_title"], + space_key=event["space_key"], + ) + + await self._on_page_changed( + event_type=event["event_type"], + page_context=page_ctx, + user_account_id=event["user_account_id"], + account_id=account_id, + ) + + async def handle_comment_event(self, event: dict, account_id: str = "default"): + if self._on_comment is None: + return + comment = event.get("comment") + if comment is None: + return + await self._on_comment(comment, account_id=account_id)