diff --git a/backend/package/yuxi/channel/extensions/kuaishou/__init__.py b/backend/package/yuxi/channel/extensions/kuaishou/__init__.py new file mode 100644 index 00000000..4858c78d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/__init__.py @@ -0,0 +1,308 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.errors import classify_error as _default_classify_error +from yuxi.channel.errors import is_retryable as _default_is_retryable +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +from .api import KuaishouAPIError +from .config import COMMON_CONFIG_SCHEMA +from .dedupe import KuaishouDedupeStore +from .format import KuaishouFormat +from .gateway import KuaishouGateway +from .media import KuaishouMedia +from .monitor import to_unified_message +from .outbound import KuaishouOutbound +from .pairing import KuaishouPairingStore +from .security import DMPolicyMode, KuaishouSecurity +from .status import KuaishouStatus +from .streaming import KuaishouBlockStreaming +from .types import KuaishouAccount +from .webhook import kwaisign_verify, parse_inbound_message + +logger = logging.getLogger(__name__) + + +class KuaishouPlugin(BaseChannelPlugin): + id = "kuaishou" + name = "快手" + order = 120 + label = "Kuaishou" + aliases = ["kuaishou", "ks", "gifshow"] + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=False, + reply=False, + media=False, + native_commands=False, + polls=False, + group_management=False, + streaming=False, + streaming_mode="block", + block_streaming=False, + experimental=True, + ) + + def __init__(self): + super().__init__() + self._gateway: KuaishouGateway | None = None + self._outbound: KuaishouOutbound | None = None + self._status: KuaishouStatus | None = None + self._media: KuaishouMedia | None = None + self._streaming: KuaishouBlockStreaming | None = None + self._format = KuaishouFormat() + self._dedupe = KuaishouDedupeStore() + self._pairing = KuaishouPairingStore() + self._security = KuaishouSecurity() + + # ── ConfigProtocol ────────────────────────────────── + + def config_schema(self) -> dict: + return COMMON_CONFIG_SCHEMA + + def list_account_ids(self, config: dict) -> list[str]: + kuaishou_config = config.get("kuaishou", {}) + accounts = kuaishou_config.get("accounts", []) + if not accounts: + return ["default"] + return [a.get("account_id", "default") for a in accounts] + + async def resolve_account(self, account_id: str) -> dict: + config = getattr(self, "_current_config", None) or {} + kuaishou_config = config.get("kuaishou", {}) + accounts = kuaishou_config.get("accounts", []) + for acct in accounts: + if acct.get("account_id", "default") == account_id: + return acct + return {"account_id": account_id, "app_id": "", "app_secret": ""} + + def is_configured(self, account: dict) -> bool: + return bool(account.get("app_id") and account.get("app_secret")) + + def is_enabled(self, account: dict) -> bool: + return account.get("enabled", False) and self.is_configured(account) + + def disabled_reason(self, account: dict) -> str: + if not self.is_configured(account): + return "未配置 App ID / App Secret" + if not account.get("enabled", False): + return "快手 IM API 尚未开放，仅作预研占位" + return "快手 IM API 尚未开放 (2026-05 调研结论)，启用后仅作占位" + + # ── GatewayProtocol ───────────────────────────────── + + async def start(self, ctx) -> object: + account_id = getattr(ctx, "account_id", "default") + account_dict = await self.resolve_account(account_id) + + if not self.is_configured(account_dict): + return {"running": False, "reason": "not-configured"} + + account = KuaishouAccount( + account_id=account_id, + app_id=account_dict.get("app_id", ""), + app_secret=account_dict.get("app_secret", ""), + webhook_token=account_dict.get("webhook_token", ""), + dm_policy=account_dict.get("dm_policy", "open"), + streaming=account_dict.get("streaming", True), + max_text_length=account_dict.get("max_text_length", 2000), + http_timeout_ms=account_dict.get("http_timeout_ms", 15000), + redirect_uri=account_dict.get("redirect_uri", ""), + scopes=account_dict.get("scopes", "user_info"), + ) + + self._gateway = KuaishouGateway(account) + result = await self._gateway.start(ctx) + + self._media = KuaishouMedia(self._gateway) + self._outbound = KuaishouOutbound(self._gateway, self._media) + self._status = KuaishouStatus(self._gateway) + + self._format = KuaishouFormat(account.max_text_length) + + self._security = KuaishouSecurity( + policy=DMPolicyMode(account.dm_policy), + ) + + if account.streaming: + self._streaming = KuaishouBlockStreaming(self._outbound) + + return result + + async def stop(self, ctx) -> None: + if self._gateway: + await self._gateway.stop(ctx) + self._gateway = None + self._outbound = None + self._status = None + self._media = None + self._streaming = None + + # ── OutboundProtocol ──────────────────────────────── + + async def send_text(self, target_id, content, *, reply_to_id=None, thread_id=None, account_id=None): + if not content or not self._outbound: + return + await self._outbound.send_text(target_id, content, reply_to_id=reply_to_id, thread_id=thread_id) + + async def send_media(self, target_id, media_url, media_type, *, reply_to_id=None, thread_id=None): + if not self._outbound: + return + if media_type == "image": + media_id = await self._outbound.upload_media(media_url) + if media_id: + await self._outbound.send_image(target_id, media_id) + + # ── StatusProtocol ────────────────────────────────── + + async def probe(self, account) -> bool: + if self._status: + return await self._status.probe() + return False + + def build_summary(self, snapshot) -> dict: + if self._status: + return self._status.build_summary(snapshot) + return { + "status": "unavailable", + "reason": "快手 IM API 尚未开放 (2026-05 调研结论)", + "monitor_url": "https://open.kuaishou.com/platform", + } + + # ── SecurityProtocol ──────────────────────────────── + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + return self._security.check_allow(peer_id) + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_policy() + + # ── PairingProtocol ───────────────────────────────── + + async def generate_code(self, peer_id: str) -> str: + return self._pairing.generate(peer_id) or "" + + async def verify_code(self, peer_id: str, code: str) -> bool: + return self._pairing.verify(peer_id, code) + + # ── FormatProtocol ────────────────────────────────── + + def markdown_to_native(self, md_text: str) -> dict | str: + return self._format.markdown_to_native(md_text) + + def sanitize_text(self, content: str) -> str: + return self._format.sanitize_text(content) + + # ── DedupeProtocol ────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._dedupe.is_duplicate(key) + + def mark_seen(self, key: str) -> None: + pass + + # ── WebhookProtocol ────────────────────────────────── + + async def handle_webhook( + self, + request: object, + account_id: str | None = None, + ) -> object: + from fastapi import Request as FastAPIRequest + + if not isinstance(request, FastAPIRequest): + return None + + body_bytes = await request.body() + kwaisign = request.headers.get("kwaisign", "") + + account_dict = await self.resolve_account(account_id or "default") + app_secret = account_dict.get("app_secret", "") + + if not kwaisign_verify(body_bytes, app_secret, kwaisign): + return None + + import json + + event_data = json.loads(body_bytes.decode("utf-8")) + return event_data + + def verify_signature(self, body: bytes, signature: str, secret: str) -> bool: + return kwaisign_verify(body, secret, signature) + + # ── InboundHandlerProtocol ────────────────────────── + + async def handle_raw_event( + self, + event: dict, + account: dict, + ) -> object | None: + inbound = parse_inbound_message(event) + return inbound + + def parse_to_unified( + self, + raw_event: dict, + account_id: str, + ) -> object | None: + inbound = parse_inbound_message(raw_event) + if inbound is None: + return None + return to_unified_message(inbound) + + # ── MediaProtocol ─────────────────────────────────── + + async def upload_media( + self, + file_path: str, + media_type: str, + *, + account_id: str | None = None, + ) -> dict: + if self._media: + media_id = await self._media.upload(file_path, media_type) + if media_id: + return {"media_id": media_id, "success": True} + return {"success": False} + + # ── StreamingProtocol ─────────────────────────────── + + async def create_draft_stream_session(self, to_user_id: str, content: str) -> list: + if self._streaming: + return await self._streaming.stream_text(to_user_id, content) + return [] + + # ── ErrorHandlingProtocol ─────────────────────────── + + def classify_error(self, error: BaseException) -> object: + from yuxi.channel.errors import ErrorSeverity + + if isinstance(error, KuaishouAPIError): + from yuxi.channel.errors import ChannelError + + code = error.code + if code == 429: + return ChannelError(str(error), severity=ErrorSeverity.RATE_LIMITED) + if code in (401, 403): + return ChannelError(str(error), severity=ErrorSeverity.FORBIDDEN) + return ChannelError(str(error), severity=ErrorSeverity.RETRYABLE, raw=error.raw) + return _default_classify_error(error) + + def is_retryable(self, error: BaseException) -> bool: + if isinstance(error, KuaishouAPIError): + return error.code not in (401, 403) + return _default_is_retryable(error) + + +ChannelPluginRegistry.register(KuaishouPlugin()) diff --git a/backend/package/yuxi/channel/extensions/kuaishou/accounts.py b/backend/package/yuxi/channel/extensions/kuaishou/accounts.py new file mode 100644 index 00000000..9d48db4f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/accounts.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/backend/package/yuxi/channel/extensions/kuaishou/api.py b/backend/package/yuxi/channel/extensions/kuaishou/api.py new file mode 100644 index 00000000..1ac640b7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/api.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import asyncio +import logging +import time +from typing import Any +from urllib.parse import urlencode + +import httpx + +from .types import TokenInfo + +logger = logging.getLogger(__name__) + +KUWAISHOU_OPEN_API_BASE = "https://open.kuaishou.com" + +# 快手官方 access_token 有效期 48h (172800s) +DEFAULT_EXPIRES_IN = 172800 +# refresh_token 有效期 180 天 (15552000s) +DEFAULT_REFRESH_EXPIRES_IN = 15552000 + + +class KuaishouAPIError(Exception): + def __init__(self, message: str, code: int = -1, raw: dict | None = None): + super().__init__(message) + self.code = code + self.raw = raw or {} + + +class KuaishouAPIClient: + def __init__(self, app_id: str, app_secret: str, timeout: float = 15.0): + self._app_id = app_id + self._app_secret = app_secret + self._http = httpx.AsyncClient( + base_url=KUWAISHOU_OPEN_API_BASE, + timeout=timeout, + ) + self._token_info: TokenInfo | None = None + self._token_lock = asyncio.Lock() + + async def close(self): + await self._http.aclose() + + @property + def access_token(self) -> str | None: + if self._token_info and time.time() < self._token_info.expires_at - 300: + return self._token_info.access_token + return None + + def get_authorization_url(self, redirect_uri: str, scope: str = "user_info") -> str: + """生成快手 OAuth 授权 URL（authorization_code 模式）。""" + params = { + "app_id": self._app_id, + "redirect_uri": redirect_uri, + "scope": scope, + "response_type": "code", + } + return f"{KUWAISHOU_OPEN_API_BASE}/oauth2/authorize?{urlencode(params)}" + + async def fetch_access_token(self, code: str, redirect_uri: str) -> TokenInfo: + """使用 authorization_code 换取 access_token。 + + 快手官方格式：GET /oauth2/access_token?grant_type=authorization_code + &code={code}&redirect_uri={redirect_uri}&app_id={app_id}&app_secret={app_secret} + """ + async with self._token_lock: + params = { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": redirect_uri, + "app_id": self._app_id, + "app_secret": self._app_secret, + } + resp = await self._http.get("/oauth2/access_token", params=params) + data = resp.json() + if data.get("result") != 1: + raise KuaishouAPIError( + f"获取 access_token 失败: {data}", + code=data.get("error_code", -1), + raw=data, + ) + self._token_info = TokenInfo( + access_token=data["access_token"], + expires_at=time.time() + data.get("expires_in", DEFAULT_EXPIRES_IN), + refresh_token=data.get("refresh_token", ""), + refresh_expires_at=time.time() + data.get("refresh_expires_in", DEFAULT_REFRESH_EXPIRES_IN), + ) + return self._token_info + + async def refresh_access_token(self) -> TokenInfo: + if not self._token_info or not self._token_info.refresh_token: + raise KuaishouAPIError("无 refresh_token，无法刷新") + + async with self._token_lock: + params = { + "grant_type": "refresh_token", + "refresh_token": self._token_info.refresh_token, + "app_id": self._app_id, + "app_secret": self._app_secret, + } + resp = await self._http.get("/oauth2/refresh_token", params=params) + data = resp.json() + if data.get("result") != 1: + raise KuaishouAPIError( + f"刷新 access_token 失败: {data}", + code=data.get("error_code", -1), + raw=data, + ) + self._token_info = TokenInfo( + access_token=data["access_token"], + expires_at=time.time() + data.get("expires_in", DEFAULT_EXPIRES_IN), + refresh_token=data.get("refresh_token", ""), + refresh_expires_at=time.time() + data.get("refresh_expires_in", DEFAULT_REFRESH_EXPIRES_IN), + ) + return self._token_info + + async def ensure_token(self) -> str: + if token := self.access_token: + return token + if self._token_info and self._token_info.refresh_token: + info = await self.refresh_access_token() + return info.access_token + raise KuaishouAPIError("未获取 access_token，请先完成 OAuth 授权") + + async def post(self, path: str, json: dict | None = None, **kwargs) -> dict[str, Any]: + resp = await self._http.post(path, json=json, **kwargs) + return resp.json() + + async def get(self, path: str, **kwargs) -> dict[str, Any]: + resp = await self._http.get(path, **kwargs) + return resp.json() diff --git a/backend/package/yuxi/channel/extensions/kuaishou/config.py b/backend/package/yuxi/channel/extensions/kuaishou/config.py new file mode 100644 index 00000000..668efae2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/config.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +COMMON_CONFIG_SCHEMA = { + "$schema": "https://json-schema.org/draft-07/schema#", + "type": "object", + "title": "快手渠道配置", + "properties": { + "appId": { + "type": "string", + "title": "App ID", + "description": "快手开放平台应用 ID (https://open.kuaishou.com)", + }, + "appSecret": { + "type": "string", + "title": "App Secret", + "x-ui-password": True, + "description": "快手开放平台应用密钥", + }, + "redirectUri": { + "type": "string", + "title": "OAuth 回调地址", + "description": "快手 OAuth 授权回调地址，用于 authorization_code 模式（API 未开放前无需配置）", + "default": "", + }, + "scopes": { + "type": "string", + "title": "授权范围", + "description": "快手 OAuth 授权范围，默认 user_info（API 未开放前无需配置）", + "default": "user_info", + }, + "webhookToken": { + "type": "string", + "title": "Webhook 验证 Token", + "x-ui-password": True, + "description": "用于验证快手服务器回调的签名 (预留，API 未开放前无需配置)", + }, + "webhookPath": { + "type": "string", + "title": "Webhook 回调路径", + "default": "/webhook/kuaishou/callback", + }, + "dmPolicy": { + "type": "string", + "title": "DM 安全策略", + "enum": ["open", "pairing", "allowlist", "disabled"], + "default": "open", + "description": "私信安全策略（本地逻辑，无需 API 即可生效）", + }, + "streaming": { + "type": "boolean", + "title": "启用流式输出", + "default": True, + "description": "块级流式输出（API 未开放前无实际效果）", + }, + "maxTextLength": { + "type": "integer", + "title": "单条消息最大字符数", + "default": 2000, + "maximum": 2000, + "description": "单条消息最大长度限制（本地逻辑）", + }, + "httpTimeoutMs": { + "type": "integer", + "title": "HTTP 请求超时(毫秒)", + "default": 15000, + }, + "enabled": { + "type": "boolean", + "title": "启用", + "default": False, + "description": "当前快手 API 尚未开放 IM 能力，启用后仅作占位", + }, + }, + "required": ["appId", "appSecret"], +} diff --git a/backend/package/yuxi/channel/extensions/kuaishou/dedupe.py b/backend/package/yuxi/channel/extensions/kuaishou/dedupe.py new file mode 100644 index 00000000..d6fce7fb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/dedupe.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import time +from collections import OrderedDict + + +class KuaishouDedupeStore: + def __init__(self, max_size: int = 10000, ttl_seconds: int = 3600): + self._store: OrderedDict[str, float] = OrderedDict() + self._max_size = max_size + self._ttl = ttl_seconds + + def is_duplicate(self, msg_id: str) -> bool: + self._evict_expired() + if msg_id in self._store: + return True + self._store[msg_id] = time.time() + self._evict_overflow() + return False + + def _evict_expired(self): + now = time.time() + expired = [k for k, v in self._store.items() if now - v > self._ttl] + for k in expired: + del self._store[k] + + def _evict_overflow(self): + while len(self._store) > self._max_size: + self._store.popitem(last=False) diff --git a/backend/package/yuxi/channel/extensions/kuaishou/format.py b/backend/package/yuxi/channel/extensions/kuaishou/format.py new file mode 100644 index 00000000..601eee59 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/format.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import re + + +def sanitize_text(content: str, max_chars: int = 2000) -> str: + if len(content) <= max_chars: + return content + return content[: max_chars - 3] + "..." + + +def markdown_to_native(md_text: str) -> str: + return md_text + + +def strip_html(content: str) -> str: + return re.sub(r"<[^>]+>", "", content or "") + + +class KuaishouFormat: + def __init__(self, max_text_length: int = 2000): + self._max_text_length = max_text_length + + def sanitize_text(self, content: str) -> str: + return sanitize_text(content, self._max_text_length) + + def markdown_to_native(self, md_text: str) -> str: + return markdown_to_native(md_text) + + def strip_html(self, content: str) -> str: + return strip_html(content) diff --git a/backend/package/yuxi/channel/extensions/kuaishou/gateway.py b/backend/package/yuxi/channel/extensions/kuaishou/gateway.py new file mode 100644 index 00000000..7640f0a5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/gateway.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import asyncio +import logging + +from .api import KuaishouAPIClient, KuaishouAPIError +from .types import KuaishouAccount, TokenInfo + +logger = logging.getLogger(__name__) + +# 快手官方 access_token 有效期 48h，刷新周期设为 44h (158400s)，留 4h 缓冲 +TOKEN_REFRESH_INTERVAL = 158400 + + +class KuaishouGateway: + def __init__(self, account: KuaishouAccount): + self._account = account + self._client: KuaishouAPIClient | None = None + self._running = False + self._refresh_task: asyncio.Task | None = None + self._awaiting_authorization = False + + @property + def account_id(self) -> str: + return self._account.account_id + + @property + def client(self) -> KuaishouAPIClient: + if self._client is None: + raise RuntimeError("Gateway 未启动") + return self._client + + @property + def awaiting_authorization(self) -> bool: + return self._awaiting_authorization + + async def start(self, ctx) -> object: + if not self._account.is_configured(): + return {"running": False, "reason": "not-configured"} + + self._client = KuaishouAPIClient( + app_id=self._account.app_id, + app_secret=self._account.app_secret, + timeout=self._account.http_timeout_ms / 1000, + ) + + # TODO(P1-3): API 开放后，如已有持久化 token，尝试 refresh；否则标记为 awaiting_authorization + # 当前阶段：快手 IM API 未开放，不自动获取 token（authorization_code 需用户授权） + self._awaiting_authorization = True + self._running = True + return { + "running": True, + "account_id": self._account.account_id, + "awaiting_authorization": True, + } + + async def authorize(self, code: str, redirect_uri: str) -> TokenInfo: + """OAuth 授权回调后，用 code 换取 token 并启动刷新循环。""" + if self._client is None: + raise RuntimeError("Gateway 未启动") + token_info = await self._client.fetch_access_token(code, redirect_uri) + self._awaiting_authorization = False + self._refresh_task = asyncio.create_task(self._token_refresh_loop()) + logger.info("快手 OAuth 授权成功，access_token 已获取") + return token_info + + async def stop(self, ctx) -> None: + self._running = False + if self._refresh_task and not self._refresh_task.done(): + self._refresh_task.cancel() + try: + await self._refresh_task + except asyncio.CancelledError: + pass + if self._client: + await self._client.close() + self._client = None + self._awaiting_authorization = False + + async def _token_refresh_loop(self) -> None: + while self._running: + await asyncio.sleep(TOKEN_REFRESH_INTERVAL) + if not self._running: + return + try: + await self._client.refresh_access_token() + logger.info("快手 access_token 刷新成功") + except KuaishouAPIError as e: + logger.exception(f"快手 access_token 刷新失败: {e}") + # TODO(P1-3): 刷新失败后标记 awaiting_authorization，提示重新授权 + self._awaiting_authorization = True + except Exception: + logger.exception("快手 access_token 刷新失败") + + async def probe(self) -> bool: + if not self._client: + return False + try: + await self._client.ensure_token() + return True + except Exception: + return False diff --git a/backend/package/yuxi/channel/extensions/kuaishou/media.py b/backend/package/yuxi/channel/extensions/kuaishou/media.py new file mode 100644 index 00000000..890cf1e0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/media.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import logging +from pathlib import Path + +from .gateway import KuaishouGateway + +logger = logging.getLogger(__name__) + +# TODO(P1-1): 当快手开放 IM API 后，替换为官方实际上传端点 +KUAISHOU_IM_UPLOAD_ENDPOINT = "/openapi/im/material/upload" + + +class KuaishouMedia: + def __init__(self, gateway: KuaishouGateway): + self._gateway = gateway + + async def upload(self, file_path: str | Path, media_type: str = "image") -> str | None: + try: + token = await self._gateway.client.ensure_token() + file_path = Path(file_path) + + resp = await self._gateway.client._http.post( + KUAISHOU_IM_UPLOAD_ENDPOINT, + data={"access_token": token, "media_type": media_type}, + files={"media": (file_path.name, file_path.open("rb"))}, + ) + data = resp.json() + if data.get("result") == 1: + return data.get("media_id", "") + return None + except Exception: + logger.exception(f"上传快手素材失败: {file_path}") + return None diff --git a/backend/package/yuxi/channel/extensions/kuaishou/monitor.py b/backend/package/yuxi/channel/extensions/kuaishou/monitor.py new file mode 100644 index 00000000..376bcce5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/monitor.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from .types import InboundKuaishouMessage + + +def to_unified_message(inbound: InboundKuaishouMessage) -> dict: + return { + "message_id": inbound.msg_id, + "channel_type": "kuaishou", + "sender": { + "id": inbound.open_id, + "name": inbound.nickname, + "avatar_url": inbound.avatar_url, + }, + "message_type": inbound.msg_type, + "content": inbound.content, + "timestamp": inbound.create_time, + "chat_type": inbound.chat_type, + "group_id": inbound.group_id, + "raw_event": inbound.raw_event, + } diff --git a/backend/package/yuxi/channel/extensions/kuaishou/outbound.py b/backend/package/yuxi/channel/extensions/kuaishou/outbound.py new file mode 100644 index 00000000..262e4816 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/outbound.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import logging + +from .api import KuaishouAPIError +from .gateway import KuaishouGateway +from .media import KuaishouMedia +from .types import OutboundResult + +logger = logging.getLogger(__name__) + +KUAISHOU_IM_SEND_ENDPOINT = "/openapi/im/message/send" + + +class KuaishouOutbound: + def __init__(self, gateway: KuaishouGateway, media: KuaishouMedia | None = None): + self._gateway = gateway + self._media = media + + async def send_text( + self, + to_user_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + chat_id: str | None = None, + ) -> OutboundResult: + try: + token = await self._gateway.client.ensure_token() + payload = { + "access_token": token, + "open_id": to_user_id, + "msg_type": "text", + "content": content[:2000], + } + if reply_to_id: + payload["reply_to_msg_id"] = reply_to_id + if chat_id: + payload["chat_id"] = chat_id + + resp = await self._gateway.client.post( + KUAISHOU_IM_SEND_ENDPOINT, + json=payload, + ) + success = resp.get("result") == 1 + return OutboundResult( + success=success, + message_id=resp.get("msg_id", ""), + raw_response=resp, + ) + except KuaishouAPIError as e: + logger.exception(f"发送快手文本消息失败: {e}") + return OutboundResult(success=False, error=str(e)) + + async def send_image( + self, + to_user_id: str, + media_id: str, + *, + reply_to_id: str | None = None, + chat_id: str | None = None, + ) -> OutboundResult: + try: + token = await self._gateway.client.ensure_token() + payload = { + "access_token": token, + "open_id": to_user_id, + "msg_type": "image", + "media_id": media_id, + } + if reply_to_id: + payload["reply_to_msg_id"] = reply_to_id + if chat_id: + payload["chat_id"] = chat_id + + resp = await self._gateway.client.post( + KUAISHOU_IM_SEND_ENDPOINT, + json=payload, + ) + return OutboundResult( + success=resp.get("result") == 1, + message_id=resp.get("msg_id", ""), + raw_response=resp, + ) + except KuaishouAPIError as e: + logger.exception(f"发送快手图片消息失败: {e}") + return OutboundResult(success=False, error=str(e)) + + async def upload_media(self, file_path: str, media_type: str = "image") -> str | None: + if self._media: + return await self._media.upload(file_path, media_type) + return None diff --git a/backend/package/yuxi/channel/extensions/kuaishou/pairing.py b/backend/package/yuxi/channel/extensions/kuaishou/pairing.py new file mode 100644 index 00000000..38356922 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/pairing.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import secrets +import time + + +class KuaishouPairingStore: + def __init__(self, code_ttl: int = 300): + self._codes: dict[str, tuple[str, float]] = {} + self._ttl = code_ttl + + def generate(self, open_id: str) -> str: + code = secrets.token_hex(3).upper() + self._codes[code] = (open_id, time.time()) + return code + + def verify(self, open_id: str, code: str) -> bool: + entry = self._codes.get(code) + if entry is None: + return False + stored_open_id, created_at = entry + if time.time() - created_at > self._ttl: + del self._codes[code] + return False + return stored_open_id == open_id diff --git a/backend/package/yuxi/channel/extensions/kuaishou/plugin.json b/backend/package/yuxi/channel/extensions/kuaishou/plugin.json new file mode 100644 index 00000000..ad10e8f0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/plugin.json @@ -0,0 +1,31 @@ +{ + "id": "kuaishou", + "name": "快手", + "version": "1.0.0", + "label": "Kuaishou", + "aliases": ["kuaishou", "ks", "gifshow"], + "description": "快手渠道插件 — 预研占位，待快手开放 IM API 后启用", + "author": "ForcePilot Team", + "order": 120, + "dependencies": ["httpx"], + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text"], + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": false, + "reply": false, + "media": false, + "native_commands": false, + "polls": false, + "group_management": false, + "streaming": false, + "streaming_mode": "block", + "block_streaming": false, + "experimental": true + }, + "enabled": true, + "python_requires": ">=3.12" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/kuaishou/security.py b/backend/package/yuxi/channel/extensions/kuaishou/security.py new file mode 100644 index 00000000..82084550 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/security.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from enum import Enum + + +class DMPolicyMode(Enum): + OPEN = "open" + PAIRING = "pairing" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +class KuaishouSecurity: + def __init__(self, policy: DMPolicyMode = DMPolicyMode.OPEN, allow_from: list[str] | None = None): + self._policy = policy + self._allowlist: set[str] = set(allow_from or []) + + def check_allow(self, open_id: str) -> bool: + if self._policy == DMPolicyMode.OPEN: + return True + if self._policy == DMPolicyMode.DISABLED: + return False + if self._policy == DMPolicyMode.ALLOWLIST: + return open_id in self._allowlist + return True + + def resolve_policy(self) -> dict: + return { + "mode": self._policy.value, + "allow_from": list(self._allowlist), + } diff --git a/backend/package/yuxi/channel/extensions/kuaishou/signature.py b/backend/package/yuxi/channel/extensions/kuaishou/signature.py new file mode 100644 index 00000000..6404562b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/signature.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import hashlib +import hmac + + +def sha1_sign(secret: str, timestamp: str, nonce: str, body: str) -> str: + """SHA1 签名（历史兼容，快手官方 WebHook 不使用此算法）。""" + params = sorted([secret, timestamp, nonce, body]) + sign_str = "".join(params) + return hashlib.sha1(sign_str.encode()).hexdigest() + + +def md5_sign(body: str, secret: str) -> str: + """MD5 签名：MD5(body + secret)。 + + 快手官方 WebHook 签名方式：通过 kwaisign header 传递 MD5(body + appsecret)。 + """ + return hashlib.md5((body + secret).encode()).hexdigest() + + +def kwaisign_verify(body: bytes, app_secret: str, signature: str) -> bool: + """验证快手官方 WebHook 签名（kwaisign header）。 + + 官方算法：MD5(request_body + appsecret)，通过 kwaisign HTTP Header 传递。 + """ + expected = hashlib.md5(body + app_secret.encode()).hexdigest() + return hmac.compare_digest(expected, signature) + + +def verify_sha1(secret: str, timestamp: str, nonce: str, body: str, signature: str) -> bool: + """验证 SHA1 签名（历史兼容）。""" + computed = sha1_sign(secret, timestamp, nonce, body) + return hmac.compare_digest(computed, signature) + + +def verify_md5(body: str, secret: str, signature: str) -> bool: + """验证 MD5 签名。""" + computed = md5_sign(body, secret) + return hmac.compare_digest(computed, signature) diff --git a/backend/package/yuxi/channel/extensions/kuaishou/status.py b/backend/package/yuxi/channel/extensions/kuaishou/status.py new file mode 100644 index 00000000..86391d03 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/status.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import logging + +from .gateway import KuaishouGateway + +logger = logging.getLogger(__name__) + + +class KuaishouStatus: + def __init__(self, gateway: KuaishouGateway): + self._gateway = gateway + + async def probe(self) -> bool: + return await self._gateway.probe() + + def build_summary(self, snapshot) -> dict: + probe_ok = snapshot.get("probe_ok", False) + account_id = self._gateway.account_id + return { + "status": "running" if probe_ok else "error", + "reason": "" if probe_ok else "token 验证失败", + "account_id": account_id, + } diff --git a/backend/package/yuxi/channel/extensions/kuaishou/streaming.py b/backend/package/yuxi/channel/extensions/kuaishou/streaming.py new file mode 100644 index 00000000..e374e08e --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/streaming.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import asyncio +import logging + +from .types import OutboundResult + +logger = logging.getLogger(__name__) + +# 默认块级流式参数（经验值，P2-2 阶段需根据官方频率限制调整） +DEFAULT_MIN_CHARS = 800 +DEFAULT_MAX_CHARS = 1200 +DEFAULT_BLOCK_DELAY_MS = 400 + + +class KuaishouBlockStreaming: + def __init__( + self, + outbound, + min_chars: int = DEFAULT_MIN_CHARS, + max_chars: int = DEFAULT_MAX_CHARS, + block_delay_ms: int = DEFAULT_BLOCK_DELAY_MS, + ): + self._outbound = outbound + self._min_chars = min_chars + self._max_chars = max_chars + self._block_delay_ms = block_delay_ms + + async def stream_text( + self, + to_user_id: str, + content: str, + ) -> list[OutboundResult]: + results: list[OutboundResult] = [] + pos = 0 + + while pos < len(content): + chunk_end = min(pos + self._max_chars, len(content)) + + if chunk_end < len(content): + break_point = content.rfind("\n", pos, chunk_end) + if break_point > pos + self._min_chars: + chunk_end = break_point + + chunk = content[pos:chunk_end] + try: + result = await self._outbound.send_text(to_user_id, chunk) + results.append(result) + except Exception: + logger.exception("流式发送块失败") + # TODO(P2-2): 收到 429 时实现指数退避重试 + raise + + pos = chunk_end + if pos < len(content): + await asyncio.sleep(self._block_delay_ms / 1000) + + return results diff --git a/backend/package/yuxi/channel/extensions/kuaishou/types.py b/backend/package/yuxi/channel/extensions/kuaishou/types.py new file mode 100644 index 00000000..6d67f998 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/types.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class KuaishouAccount: + account_id: str = "default" + app_id: str = "" + app_secret: str = "" + webhook_token: str = "" + dm_policy: str = "open" + streaming: bool = True + max_text_length: int = 2000 + http_timeout_ms: int = 15000 + enabled: bool = False + redirect_uri: str = "" + scopes: str = "user_info" + + def is_configured(self) -> bool: + return bool(self.app_id and self.app_secret) + + def is_enabled(self) -> bool: + return self.enabled and self.is_configured() + + +@dataclass +class InboundKuaishouMessage: + msg_id: str + open_id: str + msg_type: str + content: str + create_time: int + conversation_type: str + raw_event: dict = field(default_factory=dict) + nickname: str = "" + avatar_url: str = "" + # TODO(P2-3): 如官方支持群聊，增加以下字段 + chat_type: str = "direct" + group_id: str = "" + + +@dataclass +class OutboundResult: + success: bool + message_id: str = "" + error: str = "" + raw_response: dict = field(default_factory=dict) + + +@dataclass +class TokenInfo: + access_token: str + expires_at: float + refresh_token: str = "" + refresh_expires_at: float = 0.0 diff --git a/backend/package/yuxi/channel/extensions/kuaishou/webhook.py b/backend/package/yuxi/channel/extensions/kuaishou/webhook.py new file mode 100644 index 00000000..25a3b114 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/kuaishou/webhook.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import json +import logging +import time +from typing import Any + +from fastapi import APIRouter, Request, Response + +from .signature import kwaisign_verify +from .types import InboundKuaishouMessage + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook/kuaishou", tags=["kuaishou"]) + +# TODO(P1-2): 当快手开放 IM API 后，替换为官方实际事件类型 +KUAISHOU_IM_EVENT_TYPE = "im_message_receive" + + +def parse_inbound_message(raw: dict[str, Any]) -> InboundKuaishouMessage | None: + try: + event = raw.get("event", "") + if event != KUAISHOU_IM_EVENT_TYPE: + return None + + msg_data = raw.get("message", raw) + return InboundKuaishouMessage( + msg_id=str(msg_data.get("msg_id", "")), + open_id=str(msg_data.get("open_id", "")), + msg_type=msg_data.get("msg_type", "text"), + content=msg_data.get("content", ""), + create_time=msg_data.get("create_time", int(time.time() * 1000)), + conversation_type=msg_data.get("conversation_type", "1"), + nickname=msg_data.get("nickname", ""), + avatar_url=msg_data.get("avatar", ""), + raw_event=raw, + ) + except Exception: + logger.exception("解析快手入站消息失败") + return None + + +def create_webhook_handler(app_secret: str): + + @router.post("/callback") + async def kuaishou_webhook(request: Request): + body_bytes = await request.body() + + # 快手官方 WebHook 签名通过 kwaisign header 传递，算法：MD5(body + appsecret) + kwaisign = request.headers.get("kwaisign", "") + if not kwaisign_verify(body_bytes, app_secret, kwaisign): + return Response(content='{"status":"signature_invalid"}', status_code=403) + + event_data = json.loads(body_bytes.decode("utf-8")) + + if event_data.get("event") != KUAISHOU_IM_EVENT_TYPE: + return {"status": "ok"} + + inbound = parse_inbound_message(event_data) + if inbound is None: + return {"status": "ok"} + + return {"status": "ok", "msg_id": inbound.msg_id} + + return kuaishou_webhook