128 lines
3.4 KiB
Python
128 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from core.protocol import (
|
|
Attachment,
|
|
IncomingCallback,
|
|
IncomingCommand,
|
|
IncomingEvent,
|
|
IncomingMessage,
|
|
OutgoingEvent,
|
|
OutgoingMessage,
|
|
OutgoingNotification,
|
|
OutgoingTyping,
|
|
OutgoingUI,
|
|
)
|
|
|
|
PLATFORM = "web"
|
|
|
|
|
|
def json_to_incoming(data: dict) -> IncomingEvent | None:
|
|
user_id = data.get("user_id", "")
|
|
chat_id = data.get("chat_id", "C1")
|
|
msg_type = data.get("type", "message")
|
|
|
|
if msg_type == "command":
|
|
return IncomingCommand(
|
|
user_id=user_id,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
command=data.get("command", ""),
|
|
args=data.get("args", []),
|
|
)
|
|
|
|
if msg_type == "callback":
|
|
return IncomingCallback(
|
|
user_id=user_id,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
action=data.get("action", ""),
|
|
payload=data.get("payload", {}),
|
|
)
|
|
|
|
if msg_type == "message":
|
|
attachments = []
|
|
for a in data.get("attachments", []):
|
|
attachments.append(
|
|
Attachment(
|
|
type=a.get("type", "document"),
|
|
url=a.get("url"),
|
|
filename=a.get("filename"),
|
|
mime_type=a.get("mime_type"),
|
|
workspace_path=a.get("workspace_path"),
|
|
)
|
|
)
|
|
return IncomingMessage(
|
|
user_id=user_id,
|
|
platform=PLATFORM,
|
|
chat_id=chat_id,
|
|
text=data.get("text", ""),
|
|
attachments=attachments,
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def outgoing_to_json(event: OutgoingEvent) -> str:
|
|
if isinstance(event, OutgoingMessage):
|
|
payload = {
|
|
"type": "message",
|
|
"chat_id": event.chat_id,
|
|
"text": event.text,
|
|
"parse_mode": event.parse_mode,
|
|
}
|
|
if event.attachments:
|
|
payload["attachments"] = [
|
|
{
|
|
"type": a.type,
|
|
"filename": a.filename or "file",
|
|
"mime_type": a.mime_type,
|
|
"download_url": f"/files/{a.workspace_path}" if a.workspace_path else None,
|
|
}
|
|
for a in event.attachments
|
|
]
|
|
return json.dumps(payload, ensure_ascii=False)
|
|
|
|
if isinstance(event, OutgoingUI):
|
|
buttons = [
|
|
{
|
|
"label": b.label,
|
|
"action": b.action,
|
|
"payload": b.payload,
|
|
"style": b.style,
|
|
}
|
|
for b in event.buttons
|
|
]
|
|
return json.dumps(
|
|
{
|
|
"type": "ui",
|
|
"chat_id": event.chat_id,
|
|
"text": event.text,
|
|
"buttons": buttons,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
if isinstance(event, OutgoingTyping):
|
|
return json.dumps(
|
|
{
|
|
"type": "typing",
|
|
"chat_id": event.chat_id,
|
|
"is_typing": event.is_typing,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
if isinstance(event, OutgoingNotification):
|
|
return json.dumps(
|
|
{
|
|
"type": "notification",
|
|
"chat_id": event.chat_id,
|
|
"text": event.text,
|
|
"level": event.level,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
return json.dumps({"type": "unknown"})
|