Mikhail Putilovskij
3e06a67e24
- Drop reaction-based skill and confirmation helpers from Matrix conversion - Render !settings as a strict read-only dashboard snapshot - Align Matrix adapter regressions with command-only helper text
109 lines
3 KiB
Python
109 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from core.protocol import (
|
|
Attachment,
|
|
IncomingCallback,
|
|
IncomingCommand,
|
|
IncomingEvent,
|
|
IncomingMessage,
|
|
)
|
|
|
|
PLATFORM = "matrix"
|
|
|
|
|
|
def extract_attachments(event: Any) -> list[Attachment]:
|
|
msgtype = getattr(event, "msgtype", None)
|
|
if msgtype is None:
|
|
content = getattr(event, "content", {}) or {}
|
|
msgtype = content.get("msgtype")
|
|
|
|
if msgtype == "m.image":
|
|
return [
|
|
Attachment(
|
|
type="image",
|
|
url=getattr(event, "url", None),
|
|
mime_type=getattr(event, "mimetype", None),
|
|
)
|
|
]
|
|
if msgtype == "m.file":
|
|
return [
|
|
Attachment(
|
|
type="document",
|
|
url=getattr(event, "url", None),
|
|
filename=getattr(event, "body", None),
|
|
mime_type=getattr(event, "mimetype", None),
|
|
)
|
|
]
|
|
if msgtype == "m.audio":
|
|
return [
|
|
Attachment(
|
|
type="audio",
|
|
url=getattr(event, "url", None),
|
|
mime_type=getattr(event, "mimetype", None),
|
|
)
|
|
]
|
|
if msgtype == "m.video":
|
|
return [
|
|
Attachment(
|
|
type="video",
|
|
url=getattr(event, "url", None),
|
|
mime_type=getattr(event, "mimetype", None),
|
|
)
|
|
]
|
|
return []
|
|
|
|
|
|
def from_command(body: str, sender: str, chat_id: str, room_id: str | None = None) -> IncomingEvent:
|
|
raw = body.lstrip("!").strip()
|
|
parts = raw.split()
|
|
command = parts[0].lower() if parts else ""
|
|
args = parts[1:]
|
|
|
|
if command in {"yes", "no"}:
|
|
action = "confirm" if command == "yes" else "cancel"
|
|
return IncomingCallback(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
action=action,
|
|
payload={
|
|
"source": "command",
|
|
"command": command,
|
|
**({"room_id": room_id} if room_id is not None else {}),
|
|
},
|
|
)
|
|
|
|
aliases = {
|
|
"skills": "settings_skills",
|
|
"connectors": "settings_connectors",
|
|
"soul": "settings_soul",
|
|
"safety": "settings_safety",
|
|
"plan": "settings_plan",
|
|
"status": "settings_status",
|
|
"whoami": "settings_whoami",
|
|
}
|
|
command = aliases.get(command, command)
|
|
return IncomingCommand(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
command=command,
|
|
args=args,
|
|
)
|
|
|
|
|
|
def from_room_event(event: Any, room_id: str, chat_id: str) -> IncomingEvent | None:
|
|
body = (getattr(event, "body", None) or "").strip()
|
|
sender = getattr(event, "sender", "")
|
|
if body.startswith("!"):
|
|
return from_command(body, sender=sender, chat_id=chat_id, room_id=room_id)
|
|
return IncomingMessage(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
text=body,
|
|
attachments=extract_attachments(event),
|
|
reply_to=getattr(event, "replyto_event_id", None),
|
|
)
|