88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
from typing import Callable, List, Optional
|
|
from requests import post as _post
|
|
from sseclient import SSEClient
|
|
from threading import Thread
|
|
from inspect import stack
|
|
from time import sleep
|
|
from json import loads
|
|
from uuid import uuid4
|
|
|
|
import libcommon
|
|
|
|
url = "http://localhost:42069"
|
|
|
|
|
|
def post(method: str, **params):
|
|
return _post(url+"/api/v1/rpc", json={"jsonrpc": "2.0", "method": method, "params": params, "id": str(uuid4())}).json()
|
|
|
|
|
|
def sendMessage(message: str, recipient: str | libcommon.Group):
|
|
if isinstance(recipient, libcommon.Group):
|
|
return post("send", message=message, groupId=recipient.id)
|
|
return post("send", message=message, recipient=recipient)
|
|
|
|
|
|
def listGroups() -> List[libcommon.Group]:
|
|
grps = post("listGroups")["result"]
|
|
groups = []
|
|
for group in grps:
|
|
groups.append(libcommon.Group(group["name"], group["id"]))
|
|
return groups
|
|
|
|
|
|
def onMessage(group: Optional[libcommon.Group] = None) -> Callable:
|
|
def decorator(func: Callable) -> Callable:
|
|
module = stack()[1].filename.split("/")[-1][:-3]
|
|
libcommon.signalMessageBinds[module].append((group, func))
|
|
return func
|
|
return decorator
|
|
|
|
|
|
def main():
|
|
cli = Thread(target=libcommon.signalInit, args=(
|
|
"signal-cli daemon --http localhost:42069",))
|
|
cli.start()
|
|
|
|
sleep(5)
|
|
|
|
for msg in SSEClient(url+"/api/v1/events"):
|
|
try:
|
|
data = loads(msg.data)
|
|
except:
|
|
data = {}
|
|
envelope = data.get("envelope", None)
|
|
if envelope:
|
|
|
|
dataMessage = envelope.get("dataMessage", None)
|
|
if dataMessage:
|
|
groupInfo = dataMessage.get("groupInfo", None)
|
|
if groupInfo:
|
|
name = envelope["sourceName"]
|
|
body = dataMessage["message"]
|
|
groupId = groupInfo["groupId"]
|
|
if body is not None:
|
|
for module in libcommon.loadedMods:
|
|
if module in libcommon.signalMessageBinds:
|
|
for group, func in libcommon.signalMessageBinds[module]:
|
|
if group is None or group.id == groupId:
|
|
func(name, body)
|
|
|
|
syncMessage = envelope.get("syncMessage", None)
|
|
if syncMessage:
|
|
sentMessage = syncMessage.get("sentMessage", None)
|
|
if sentMessage:
|
|
groupInfo = sentMessage.get("groupInfo", None)
|
|
if groupInfo:
|
|
name = envelope["sourceName"]
|
|
body = sentMessage["message"]
|
|
groupId = groupInfo["groupId"]
|
|
if body is not None:
|
|
for module in libcommon.loadedMods:
|
|
if module in libcommon.signalMessageBinds:
|
|
for group, func in libcommon.signalMessageBinds[module]:
|
|
if group is None or group.id == groupId:
|
|
func(name, body)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|