feat: 新增 Misskey 平台适配器 #2774
Merged
feat: 新增 Misskey 平台适配器 #2774
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- The default config uses a "Misskey" section name but the platform manager matches on lowercase "misskey", please align the section key with the adapter id for consistency.
- The text formatting logic is duplicated in send_by_session and MisskeyPlatformEvent.send; consider extracting a shared helper to avoid code duplication.
- The custom ClientSession wrapper may leak aiohttp sessions if not closed in every path—ensure sessions are properly closed on adapter termination or errors.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The default config uses a "Misskey" section name but the platform manager matches on lowercase "misskey", please align the section key with the adapter id for consistency.
- The text formatting logic is duplicated in send_by_session and MisskeyPlatformEvent.send; consider extracting a shared helper to avoid code duplication.
- The custom ClientSession wrapper may leak aiohttp sessions if not closed in every path—ensure sessions are properly closed on adapter termination or errors.
## Individual Comments
### Comment 1
<location> `astrbot/core/platform/sources/misskey/misskey_adapter.py:84-93` </location>
<code_context>
+ while self._running:
</code_context>
<issue_to_address>
**suggestion (performance):** Polling loop does not implement exponential backoff on repeated errors.
Implementing exponential backoff will help prevent excessive API requests during persistent errors and improve system resilience.
Suggested implementation:
```python
# Exponential backoff parameters
initial_backoff = 1 # seconds
max_backoff = 60 # seconds
backoff_multiplier = 2
current_backoff = initial_backoff
while self._running:
if not self.api:
logger.error("[Misskey] API 客户端在轮询过程中变为 None")
break
try:
notifications = await self.api.get_mentions(
limit=20, since_id=self.last_notification_id
)
# Reset backoff on success
current_backoff = initial_backoff
if notifications:
```
```python
try:
notifications = await self.api.get_mentions(
limit=20, since_id=self.last_notification_id
)
# Reset backoff on success
current_backoff = initial_backoff
if notifications:
```
```python
except Exception as e:
logger.warning(f"[Misskey] 获取通知失败: {e}")
logger.info(f"[Misskey] 轮询将在 {current_backoff} 秒后重试（指数退避）")
await asyncio.sleep(current_backoff)
current_backoff = min(current_backoff * backoff_multiplier, max_backoff)
continue
```
</issue_to_address>
### Comment 2
<location> `astrbot/core/platform/sources/misskey/misskey_api.py:237-243` </location>
<code_context>
+
+ result = await self._make_request("i/notifications", data)
+ # Misskey API 返回通知列表
+ if isinstance(result, list):
+ return result
+ elif isinstance(result, dict) and "notifications" in result:
+ return result["notifications"]
+ else:
+ logger.warning(f"获取提及通知响应格式异常: {type(result)}")
+ return []
</code_context>
<issue_to_address>
**suggestion:** get_mentions response format handling may be brittle.
Log the entire response when the format is unexpected to simplify future debugging.
</issue_to_address>
### Comment 3
<location> `astrbot/core/platform/sources/misskey/misskey_adapter.py:161` </location>
<code_context>
+
+ return False
+
+ async def send_by_session(
+ self, session: MessageSesion, message_chain: MessageChain
+ ) -> Awaitable[Any]:
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring by extracting helpers for session parsing, component serialization, and visibility resolution to simplify and modularize the code.
```markdown
You can dramatically simplify both `send_by_session` and `convert_message` by extracting three focused helpers:
1) parsing the session‐id,
2) serializing a chain of components to text,
3) resolving visibility settings.
---
### 1. Parse session info
```python
def _parse_session(self, session_id: str) -> Tuple[Optional[str], Optional[str]]:
if not session_id:
return None, None
parts = session_id.split("|", 1)
user_id = parts[0]
orig_id = parts[1] if len(parts) > 1 else None
return user_id, orig_id
```
### 2. Serialize components
```python
def _components_to_text(self, comps: Iterable[Any]) -> Tuple[str, bool]:
text_parts = []
has_at = False
for c in comps:
if isinstance(c, Comp.Plain):
text_parts.append(c.text)
elif isinstance(c, Comp.Image):
text_parts.append("[图片]")
elif isinstance(c, Comp.At):
has_at = True
text_parts.append(f"@{c.qq}")
elif isinstance(c, Comp.Node):
sub, sub_at = self._components_to_text(c.content or [])
has_at |= sub_at
text_parts.append(sub)
else:
text_parts.append(str(c))
return "".join(text_parts), has_at
```
### 3. Resolve visibility
```python
def _resolve_visibility(self, user_id: Optional[str]) -> Tuple[str, Optional[List[str]]]:
vis = "public"
vis_ids = None
if user_id and (info := self._user_cache.get(user_id)):
original = info.get("visibility", "public")
if original == "specified":
vis = "specified"
vis_ids = list({*info.get("visible_user_ids", []), user_id, self.client_self_id})
else:
vis = original
return vis, vis_ids
```
---
Then your `send_by_session` shrinks to:
```python
async def send_by_session(self, session, chain):
user_id, orig_id = self._parse_session(session.session_id)
text, has_at = self._components_to_text(chain.chain)
if not has_at and orig_id and user_id:
# prepend @username logic…
if not text.strip():
return await super().send_by_session(session, chain)
text = text[: self.max_message_length]
visibility, vis_ids = self._resolve_visibility(user_id)
if orig_id:
await self.api.create_note(text, visibility=visibility,
reply_id=orig_id, visible_user_ids=vis_ids)
elif user_id and self._is_user_session(user_id):
await self.api.send_message(user_id, text)
else:
await self.api.create_note(text, visibility=visibility, visible_user_ids=vis_ids)
return await super().send_by_session(session, chain)
```
And `convert_message` can reuse `_components_to_text` for files and plain text, extracting the “@bot” prefix logic into another small helper. This approach preserves all behavior but flattens nesting, removes duplication, and makes each piece independently testable.```
</issue_to_address>
### Comment 4
<location> `astrbot/core/platform/sources/misskey/misskey_api.py:52` </location>
<code_context>
+
+
+# HTTP Client Session Manager
+class ClientSession:
+ session: aiohttp.ClientSession | None = None
+ _token: str | None = None
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring to use a per-instance aiohttp session and Tenacity for retries to simplify resource management and error handling.
Here are two small focused refactorings that remove the global `ClientSession` manager and custom retry logic, while keeping all behavior:
1. **Move session to the instance**
Drop the `ClientSession` class and create an `aiohttp.ClientSession` per `MisskeyAPI` in `__init__`.
```python
class MisskeyAPI:
def __init__(
self,
instance_url: str,
access_token: str,
session: aiohttp.ClientSession | None = None,
):
self.instance_url = instance_url.rstrip("/")
self.access_token = access_token
# create or use injected session
self.session = session or aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.access_token}"}
)
async def close(self) -> None:
await self.session.close()
logger.debug("Misskey API 客户端已关闭")
```
2. **Use Tenacity for retry**
Replace the custom `retry_async` decorator with [tenacity](https://github.com/jd/tenacity).
```python
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
@retry(
stop=stop_after_attempt(API_MAX_RETRIES),
wait=wait_fixed(1),
retry=retry_if_exception_type((APIConnectionError, APIRateLimitError)),
)
async def _make_request(
self, endpoint: str, data: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
url = f"{self.instance_url}/api/{endpoint}"
payload = {"i": self.access_token, **(data or {})}
try:
async with self.session.post(url, json=payload) as resp:
return await self._process_response(resp, endpoint)
except (aiohttp.ClientError, json.JSONDecodeError) as e:
logger.error(f"HTTP 请求错误: {e}")
raise APIConnectionError() from e
```
This keeps all existing functionality but:
- Eliminates the global/shared session boilerplate.
- Replaces the homemade retry loop with a battle-tested library.
- Simplifies setup/teardown to straightforward `__init__/close`.
</issue_to_address>
### Comment 5
<location> `astrbot/core/platform/sources/misskey/misskey_event.py:21` </location>
<code_context>
+ self.client = client
+
+ async def send(self, message: MessageChain):
+ content = ""
+ for item in message.chain:
+ if isinstance(item, Plain):
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the message serialization and note argument construction into shared helper functions to simplify the send method and enable reuse.
```markdown
Consider extracting the flattening/serialization logic (and the “@reply” prefix + note‐args builder) into small shared helpers. That will collapse your nested loops/ifs into a few calls and remove duplication across adapters.
For example, in a new file `astrbot/core/message/utils.py`:
```python
# astrbot/core/message/utils.py
from astrbot.api.message_components import Plain
from astrbot.core.message.components import Node
def serialize_chain(chain) -> str:
parts = []
for item in chain:
if isinstance(item, Plain):
parts.append(item.text)
elif isinstance(item, Node) and item.content:
for sub in item.content:
parts.append(getattr(sub, "text", ""))
else:
parts.append(getattr(item, "text", ""))
return "".join(parts)
def prefix_reply(content: str, raw_message: dict) -> str:
user = raw_message.get("user", {}).get("username", "")
if user and not content.startswith(f"@{user}"):
return f"@{user} {content}"
return content
def build_note_kwargs(reply_id, content: str, raw_message: dict) -> dict:
visibility = raw_message.get("visibility", "public")
kwargs = {"content": content, "reply_id": reply_id, "visibility": visibility}
if visibility == "specified":
users = raw_message.get("visibleUserIds", [])
sender = raw_message.get("userId")
# ensure unique, non-empty IDs
kwargs["visible_user_ids"] = list({*users, sender} - {None, ""})
return kwargs
```
Then your `send` becomes:
```python
from astrbot import logger
from astrbot.api.event import AstrMessageEvent
from astrbot.core.message.utils import serialize_chain, prefix_reply, build_note_kwargs
class MisskeyPlatformEvent(AstrMessageEvent):
...
async def send(self, message):
content = serialize_chain(message.chain)
if not content:
logger.debug("[MisskeyEvent] 内容为空，跳过发送")
return
raw = getattr(self.message_obj, "raw_message", {}) or {}
content = prefix_reply(content, raw)
try:
orig_id = getattr(self.message_obj, "message_id", None)
if orig_id and hasattr(self.client, "create_note"):
await self.client.create_note(**build_note_kwargs(orig_id, content, raw))
return
sid = str(self.session_id or "")
if hasattr(self.client, "send_message") and 5 <= len(sid) <= 64 and " " not in sid:
logger.debug(f"[MisskeyEvent] 发送私信: {sid}")
await self.client.send_message(sid, content)
return
if hasattr(self.client, "create_note"):
logger.debug("[MisskeyEvent] 创建新帖子")
await self.client.create_note(content)
return
await super().send(message)
except Exception as e:
logger.error(f"[MisskeyEvent] 发送失败: {e}")
```
This:
- Flattens your nested loops/conditionals into three small helpers
- Keeps all existing behavior
- Can be reused by other adapters to remove duplication
- Makes the `send` body much easier to read/maintain.
</issue_to_address>
### Comment 6
<location> `astrbot/core/platform/sources/misskey/misskey_adapter.py:94` </location>
<code_context>
async def _start_polling(self):
if not self.api:
logger.error("[Misskey] API 客户端未初始化，无法开始轮询")
return
is_first_poll = True
try:
latest_notifications = await self.api.get_mentions(limit=1)
if latest_notifications:
self.last_notification_id = latest_notifications[0].get("id")
logger.debug(f"[Misskey] 起始通知 ID: {self.last_notification_id}")
except Exception as e:
logger.warning(f"[Misskey] 获取起始通知失败: {e}")
while self._running:
if not self.api:
logger.error("[Misskey] API 客户端在轮询过程中变为 None")
break
try:
notifications = await self.api.get_mentions(
limit=20, since_id=self.last_notification_id
)
if notifications:
if is_first_poll:
logger.debug(f"[Misskey] 跳过 {len(notifications)} 条历史通知")
is_first_poll = False
self.last_notification_id = notifications[0].get("id")
else:
notifications.reverse()
for notification in notifications:
await self._process_notification(notification)
self.last_notification_id = notifications[0].get("id")
else:
if is_first_poll:
is_first_poll = False
logger.info("[Misskey] 开始监听新消息")
await asyncio.sleep(self.poll_interval)
except Exception as e:
logger.error(f"[Misskey] 轮询错误: {e}")
await asyncio.sleep(5)
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Merge else clause's nested if statement into elif ([`merge-else-if-into-elif`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-else-if-into-elif/))
- Hoist repeated code outside conditional statement ([`hoist-statement-from-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/hoist-statement-from-if/))
</issue_to_address>
### Comment 7
<location> `astrbot/core/platform/sources/misskey/misskey_adapter.py:161` </location>
<code_context>
async def send_by_session(
self, session: MessageSesion, message_chain: MessageChain
) -> Awaitable[Any]:
if not self.api:
logger.error("[Misskey] API 客户端未初始化")
return super().send_by_session(session, message_chain)
try:
# 首先解析session信息
session_id = session.session_id
user_id = None
original_message_id = None
if session_id and "|" in session_id:
parts = session_id.split("|", 1)
user_id = parts[0]
original_message_id = parts[1] if len(parts) > 1 else None
elif session_id:
user_id = session_id
text = ""
has_at_user = False
for component in message_chain.chain:
if isinstance(component, Comp.Plain):
text += component.text
elif isinstance(component, Comp.Image):
text += "[图片]"
elif isinstance(component, Comp.Node):
if component.content:
for node_comp in component.content:
if isinstance(node_comp, Comp.Plain):
text += node_comp.text
elif isinstance(node_comp, Comp.Image):
text += "[图片]"
else:
text += str(node_comp)
elif isinstance(component, Comp.At):
has_at_user = True
text += f"@{component.qq}"
else:
text += str(component)
if not has_at_user and original_message_id and user_id:
user_info = self._user_cache.get(user_id)
if user_info:
username = user_info.get("username")
nickname = user_info.get("nickname")
if username:
text = f"@{username} {text}".strip()
elif nickname:
text = f"@{nickname} {text}".strip()
if not text or not text.strip():
logger.warning("[Misskey] 消息内容为空，跳过发送")
return await super().send_by_session(session, message_chain)
if len(text) > self.max_message_length:
text = text[: self.max_message_length] + "..."
visibility = "public"
visible_user_ids = None
if user_id:
user_info = self._user_cache.get(user_id)
if user_info:
original_visibility = user_info.get("visibility", "public")
if original_visibility == "specified":
visibility = "specified"
original_visible_users = user_info.get("visible_user_ids", [])
users_to_include = [user_id]
if self.client_self_id:
users_to_include.append(self.client_self_id)
visible_user_ids = list(
set(original_visible_users + users_to_include)
)
visible_user_ids = [uid for uid in visible_user_ids if uid]
else:
visibility = original_visibility
# 发送消息
if original_message_id:
await self.api.create_note(
text,
visibility=visibility,
reply_id=original_message_id,
visible_user_ids=visible_user_ids,
)
elif user_id and self._is_user_session(user_id):
await self.api.send_message(user_id, text)
else:
await self.api.create_note(
text, visibility=visibility, visible_user_ids=visible_user_ids
)
except Exception as e:
logger.error(f"[Misskey] 发送消息失败: {e}")
return await super().send_by_session(session, message_chain)
</code_context>
<issue_to_address>
**issue (code-quality):** Low code quality found in MisskeyPlatformAdapter.send\_by\_session - 12% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>
### Comment 8
<location> `astrbot/core/platform/sources/misskey/misskey_adapter.py:264` </location>
<code_context>
async def convert_message(self, raw_data: Dict[str, Any]) -> AstrBotMessage:
message = AstrBotMessage()
message.raw_message = raw_data
message.message_str = raw_data.get("text", "")
message.message = []
sender = raw_data.get("user", {})
sender_username = sender.get("username", "")
message.sender = MessageMember(
user_id=str(sender.get("id", "")),
nickname=sender.get("name", sender.get("username", "")),
)
user_id = message.sender.user_id
message_id = str(raw_data.get("id", ""))
message.session_id = f"{user_id}|{message_id}"
message.message_id = message_id
message.self_id = self.client_self_id
message.type = MessageType.FRIEND_MESSAGE
self._user_cache[user_id] = {
"username": sender_username,
"nickname": message.sender.nickname,
"visibility": raw_data.get("visibility", "public"),
"visible_user_ids": raw_data.get("visibleUserIds", []),
}
raw_text = message.message_str
if raw_text:
if self._bot_username:
at_mention = f"@{self._bot_username}"
if raw_text.startswith(at_mention):
message.message.append(Comp.At(qq=self.client_self_id))
remaining_text = raw_text[len(at_mention) :].strip()
message.message_str = remaining_text
if remaining_text:
message.message.append(Comp.Plain(remaining_text))
return message
message.message.append(Comp.Plain(raw_text))
# 处理文件附件
files = raw_data.get("files", [])
if files:
for file_info in files:
file_type = file_info.get("type", "").lower()
file_url = file_info.get("url", "")
file_name = file_info.get("name", "未知文件")
if file_type.startswith("image/"):
# 图片文件
message.message.append(Comp.Image(file_url))
else:
# 其他文件类型，作为纯文本描述
message.message.append(Comp.Plain(f"[文件: {file_name}]"))
return message
</code_context>
<issue_to_address>
**issue (code-quality):** Use named expression to simplify assignment and conditional [×2] ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
</issue_to_address>
### Comment 9
<location> `astrbot/core/platform/sources/misskey/misskey_api.py:4-9` </location>
<code_context>
import json
from typing import Any, Optional
try:
import aiohttp
except ImportError:
raise ImportError(
"aiohttp is required for Misskey API. Please install it with: pip install aiohttp"
)
try:
from loguru import logger # type: ignore
except ImportError:
try:
from astrbot import logger
except ImportError:
import logging
logger = logging.getLogger(__name__)
# Constants
API_MAX_RETRIES = 3
HTTP_OK = 200
HTTP_BAD_REQUEST = 400
HTTP_UNAUTHORIZED = 401
HTTP_FORBIDDEN = 403
HTTP_TOO_MANY_REQUESTS = 429
# Exceptions
class APIError(Exception):
pass
class APIBadRequestError(APIError):
pass
class APIConnectionError(APIError):
pass
class APIRateLimitError(APIError):
pass
class AuthenticationError(APIError):
pass
</code_context>
<issue_to_address>
**issue (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
</issue_to_address>
### Comment 10
<location> `astrbot/core/platform/sources/misskey/misskey_api.py:160` </location>
<code_context>
async def _process_response(self, response, endpoint: str):
if response.status == HTTP_OK:
try:
result = await response.json()
logger.debug(f"Misskey API 请求成功: {endpoint}")
return result
except json.JSONDecodeError as e:
logger.error(f"响应不是有效的 JSON 格式: {e}")
raise APIConnectionError()
# 获取错误响应的详细内容
try:
error_text = await response.text()
logger.error(
f"API 请求失败: {endpoint} - 状态码: {response.status}, 响应: {error_text}"
)
except Exception:
logger.error(
f"API 请求失败: {endpoint} - 状态码: {response.status}, 无法读取错误响应"
)
self._handle_response_status(response, endpoint)
raise APIConnectionError()
</code_context>
<issue_to_address>
**suggestion (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
```suggestion
raise APIConnectionError() from e
```
</issue_to_address>
### Comment 11
<location> `astrbot/core/platform/sources/misskey/misskey_api.py:186` </location>
<code_context>
@retry_async(
max_retries=API_MAX_RETRIES,
retryable_exceptions=(APIConnectionError, APIRateLimitError),
)
async def _make_request(
self, endpoint: str, data: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""发送 API 请求"""
url = f"{self.instance_url}/api/{endpoint}"
payload = {"i": self.access_token}
if data:
payload.update(data)
try:
async with self.session.post(url, json=payload) as response:
return await self._process_response(response, endpoint)
except (aiohttp.ClientError, json.JSONDecodeError) as e:
logger.error(f"HTTP 请求错误: {e}")
raise APIConnectionError() from e
</code_context>
<issue_to_address>
**suggestion (code-quality):** Merge dictionary updates via the union operator ([`dict-assign-update-to-union`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/dict-assign-update-to-union/))
```suggestion
payload |= data
```
</issue_to_address>
### Comment 12
<location> `astrbot/core/platform/sources/misskey/misskey_event.py:20` </location>
<code_context>
async def send(self, message: MessageChain):
content = ""
for item in message.chain:
if isinstance(item, Plain):
content += item.text
elif isinstance(item, Node) and item.content:
for sub_item in item.content:
content += getattr(sub_item, "text", "")
else:
content += getattr(item, "text", "")
if not content:
logger.debug("[MisskeyEvent] 内容为空，跳过发送")
return
try:
original_message_id = getattr(self.message_obj, "message_id", None)
raw_message = getattr(self.message_obj, "raw_message", {})
if raw_message:
user_data = raw_message.get("user", {})
username = user_data.get("username", "")
if username and not content.startswith(f"@{username}"):
content = f"@{username} {content}"
if original_message_id and hasattr(self.client, "create_note"):
visibility = "public"
visible_user_ids = None
if raw_message:
original_visibility = raw_message.get("visibility", "public")
if original_visibility == "specified":
visibility = "specified"
original_visible_users = raw_message.get("visibleUserIds", [])
sender_id = raw_message.get("userId", "")
users_to_include = [sender_id] if sender_id else []
visible_user_ids = list(
set(original_visible_users + users_to_include)
)
visible_user_ids = [uid for uid in visible_user_ids if uid]
else:
visibility = original_visibility
await self.client.create_note(
content,
reply_id=original_message_id,
visibility=visibility,
visible_user_ids=visible_user_ids,
)
elif hasattr(self.client, "send_message") and self.session_id:
sid = str(self.session_id)
if 5 <= len(sid) <= 64 and " " not in sid:
logger.debug(f"[MisskeyEvent] 发送私信: {sid}")
await self.client.send_message(sid, content)
return
elif hasattr(self.client, "create_note"):
logger.debug("[MisskeyEvent] 创建新帖子")
await self.client.create_note(content)
await super().send(message)
except Exception as e:
logger.error(f"[MisskeyEvent] 发送失败: {e}")
</code_context>
<issue_to_address>
**issue (code-quality):** Low code quality found in MisskeyPlatformEvent.send - 23% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Contributor
Author
|
已完成个人需求的功能接入，已在个人 misskey bot 账号完成运行测试，测试无运行问题， |
Contributor
Author
|
|
Contributor
Author
|
目前来看暂时没有需要改进的点了 |
PaloMiku
force-pushed
the
feature/misskey-adapter-add
branch
from
401c81a to
be32999
Compare
Member
|
测试了一下，看起来不错！ 图片目前好像没办法发送，会返回： 
这样的内容 |
Soulter
approved these changes
Sep 18, 2025
Contributor
Author
机器人发送图片吗？ 接收图片是没问题的，但发送图片不大敢做 |
Contributor
Author
主要是 misskey 云盘策略问题，这个完全各实例自己定，而且存储满了也不好释放，要是 misskey 可以插入外部图片也不至于这样 |
Member
了解了！LGTM 此外想请教一下，我发现 Misskey 有聊天功能，如果一起加进去的话难度大吗？ |
Contributor
Author
现在研究下看看🧐 |
Contributor
Author
大概是行了，现在换成 Websockets 连接 Misskey 了，毕竟聊天信息用轮询肯定不行 |
Member
|
看起来可以正常进行私聊对话了！ 但是在群聊中似乎识别有一些问题：
|
Contributor
Author
群聊确实不是我一开始考虑的方向（ |
Contributor
Author
今晚8点半以后看看，6点半到8点半作为大学牲上课（ |
Contributor
Author
|
实际上 Misskey 的聊天和聊天群聊支持就是个大残废，只能 misskey 单个程序和实例自己用不说，差不多是直接照搬好几年以前的 Misskey Fork（Firefish）加入的聊天功能，加入后也没有新的针对性更新，也缺失作为群聊该有的管理性，而且 Misskey 的聊天在服务器数据库也没有加密，私聊不是特别安全 |
Member
|
看起来是有的 curl https://example.tld/api/chat/messages/create-to-room \
--request POST \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer YOUR_SECRET_TOKEN' \
--data '{
"text": null,
"fileId": "",
"toRoomId": ""
}'日志：
|
Contributor
Author
我平常就没仔细研究聊天的 api 接口（， 得亏我为了日常能看明白返回数据特意把代码里返回 json 格式化了，不然你看到的 json 多半只有一行😅 |
Contributor
Author
写好了，等会检查完代码 Push |
Member
太酷啦！ |
Contributor
Author
|
完逝 |
Member
|
我修改了一些地方：
|
: 作为分隔符可能会导致 umo 组装出现问题
Soulter
approved these changes
Sep 18, 2025
Member
|
LGTM |
Larch-C
approved these changes
Sep 18, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
fixes #2295
Motivation / 动机
为 Astrbot 增加了 Misskey 平台适配器，可接入 Misskey Fediverse Bot 生态。
Modifications / 改动点
增加了 Misskey 平台适配器和后台仪表板为 Misskey 平台适配器增加 Misskey Logo 显示。
Verification Steps / 验证步骤
Clone 并启动代码（需本地 node 启动独立控制台验证Logo修正），在后台仪表板平台适配器中添加 Misskey 平台适配器配置。
Screenshots or Test Results / 运行截图或测试结果
Compatibility & Breaking Changes / 兼容性与破坏性变更
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Sourcery 总结
通过引入新的适配器、API 客户端和事件集成，并更新配置和 UI 以包含 Misskey，从而添加对 Misskey 平台的支持。
新功能：
Original summary in English
Summary by Sourcery
Add support for the Misskey platform by introducing a new adapter, API client, and event integration, and updating configuration and UI to include Misskey.
New Features: