143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from adapter.matrix.reactions import CANCEL_REACTION, CONFIRM_REACTION, reaction_to_skill_index
|
|
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) -> 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},
|
|
)
|
|
|
|
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_reaction(event: Any, sender: str, chat_id: str) -> IncomingCallback | None:
|
|
content = getattr(event, "content", {}) or {}
|
|
relates_to = content.get("m.relates_to", {})
|
|
key = getattr(event, "key", None) or relates_to.get("key")
|
|
event_id = getattr(event, "event_id", None) or relates_to.get("event_id")
|
|
if not key:
|
|
return None
|
|
|
|
if key == CONFIRM_REACTION:
|
|
return IncomingCallback(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
action="confirm",
|
|
payload={"event_id": event_id, "reaction": key},
|
|
)
|
|
if key == CANCEL_REACTION:
|
|
return IncomingCallback(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
action="cancel",
|
|
payload={"event_id": event_id, "reaction": key},
|
|
)
|
|
|
|
skill_index = reaction_to_skill_index(key)
|
|
if skill_index is not None:
|
|
return IncomingCallback(
|
|
user_id=sender,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
action="toggle_skill",
|
|
payload={"event_id": event_id, "reaction": key, "skill_index": skill_index},
|
|
)
|
|
return None
|
|
|
|
|
|
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)
|
|
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),
|
|
)
|