Update.

Browse Source
This commit is contained in:
retoor 2025-06-14 15:42:29 +02:00
parent c4739ff7d3
commit 0341e2f2d8
2 changed files with 102 additions and 61 deletions
src/snek
service
chat.py
view
rpc.py

7
src/snek/service/chat.py
View File

@ -42,7 +42,12 @@ class ChatService(BaseService):
if channel_message:
channel_message["message"] = message
channel_message["is_final"] = is_final
await self.services.channel_message.save(channel_message)
if not channel_message["is_final"]:
async with self.app.no_save():
channel_message["updated_at"] = now()
await self.services.channel_message.save(channel_message)
else:
await self.services.channel_message.save(channel_message)
else:
channel_message = await self.services.channel_message.create(
channel_uid, user_uid, message, is_final

156
src/snek/view/rpc.py
View File

@ -6,19 +6,18 @@
# MIT License: Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions.
import asyncio
import json
import logging
import traceback
from datetime import datetime
from aiohttp import web
from snek.system.model import now
from snek.system.profiler import Profiler
from snek.system.view import BaseView
import time
import time
logger = logging.getLogger(__name__)
@ -30,19 +29,19 @@ class RPCView(BaseView):
self.services = self.app.services
self.ws = ws
self.user_session = {}
self._finalize_task = None
self._scheduled = []
self._finalize_task = None
async def _session_ensure(self):
uid = self.view.session.get("uid")
uid = await self.view.session_get("uid")
if uid not in self.user_session:
self.user_session[uid] = {
"said_hello": False,
}
async def session_get(self, key, default=None):
async def session_get(self, key, default):
await self._session_ensure()
return self.user_session.get(self.user_uid, {}).get(key, default)
return self.user_session[self.user_uid].get(key, default)
async def session_set(self, key, value):
await self._session_ensure()
@ -51,6 +50,7 @@ class RPCView(BaseView):
async def db_insert(self, table_name, record):
self._require_login()
return await self.services.db.insert(self.user_uid, table_name, record)
async def db_update(self, table_name, record):
@ -65,7 +65,7 @@ class RPCView(BaseView):
return await self.services.socket.broadcast(
channel_uid,
{
"channel_uid": channel_uid,
"channel_uid": "293ecf12-08c9-494b-b423-48ba1a2d12c2",
"event": "set_typing",
"data": {
"event": "set_typing",
@ -96,7 +96,7 @@ class RPCView(BaseView):
self.user_uid, table_name, record, keys
)
async def db_query(self, table_name, sql, args):
async def db_query(self, table_name, args):
self._require_login()
return await self.services.db.query(self.user_uid, table_name, sql, args)
@ -130,23 +130,21 @@ class RPCView(BaseView):
self.view.session["logged_in"] = True
self.view.session["username"] = user["username"]
self.view.session["user_nick"] = user["nick"]
record = dict(user.record)
if "password" in record:
del record["password"]
if "deleted_at" in record:
del record["deleted_at"]
record = user.record
del record["password"]
del record["deleted_at"]
await self.services.socket.add(
self.ws, self.view.session.get("uid")
self.ws, self.view.request.session.get("uid")
)
async for subscription in self.services.channel_member.find(
user_uid=self.view.session.get("uid"),
user_uid=self.view.request.session.get("uid"),
deleted_at=None,
is_banned=False,
):
await self.services.socket.subscribe(
self.ws,
subscription["channel_uid"],
self.view.session.get("uid"),
self.view.request.session.get("uid"),
)
return record
@ -154,25 +152,22 @@ class RPCView(BaseView):
self._require_login()
return [user["username"] for user in await self.services.user.search(query)]
async def get_user(self, user_uid=None):
async def get_user(self, user_uid):
self._require_login()
if not user_uid:
user_uid = self.user_uid
user = await self.services.user.get(uid=user_uid)
record = dict(user.record)
if "password" in record:
del record["password"]
if "deleted_at" in record:
del record["deleted_at"]
record = user.record
del record["password"]
del record["deleted_at"]
if user_uid != user["uid"]:
if "email" in record:
del record["email"]
del record["email"]
return record
async def get_messages(self, channel_uid, offset=0, timestamp=None):
self._require_login()
messages = []
async for message in self.services.channel_message.offset(
for message in await self.services.channel_message.offset(
channel_uid, offset or 0, timestamp or None
):
extended_dict = await self.services.channel_message.to_extended_dict(
@ -203,56 +198,54 @@ class RPCView(BaseView):
"new_count": subscription["new_count"],
"is_moderator": subscription["is_moderator"],
"is_read_only": subscription["is_read_only"],
"new_count": subscription["new_count"],
"color": color,
}
)
return channels
async def clear_channel(self, channel_uid):
self._require_login()
user = await self.services.user.get(uid=self.user_uid)
if not user["is_admin"]:
raise Exception("Not allowed")
channel = await self.services.channel.get(uid=channel_uid)
if not channel:
raise Exception("Channel not found")
channel['history_start'] = datetime.now()
await self.services.channel.save(channel)
return await self.services.channel_message.clear(channel_uid)
async def write_container(self, channel_uid, content, timeout=3):
async def write_container(self, channel_uid, content,timeout=3):
self._require_login()
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise Exception("Not allowed")
container_name = await self.services.container.get_container_name(channel_uid)
class SessionCall:
def __init__(self, app, channel_uid, container_name):
self.app = app
self.channel_uid = channel_uid
self.container_name = container_name
def __init__(self, app,channel_uid_uid, container_name):
self.app = app
self.channel_uid = channel_uid
self.container_name = container_name
self.time_last_output = time.time()
self.output = b''
async def stdout_event_handler(self, data):
self.time_last_output = time.time()
self.output += data
return True
return True
async def communicate(self, content, timeout=3):
async def communicate(self,content, timeout=3):
await self.app.services.container.add_event_listener(self.container_name, "stdout", self.stdout_event_handler)
await self.app.services.container.write_stdin(self.channel_uid, content)
while time.time() - self.time_last_output < timeout:
await asyncio.sleep(0.1)
await self.app.services.container.remove_event_listener(self.container_name, "stdout", self.stdout_event_handler)
return self.output
sc = SessionCall(self, channel_uid, container_name)
return (await sc.communicate(content)).decode("utf-8", "ignore")
sc = SessionCall(self, channel_uid,container_name)
return (await sc.communicate(content)).decode("utf-8","ignore")
async def get_container(self, channel_uid):
self._require_login()
@ -300,7 +293,7 @@ class RPCView(BaseView):
if not channel_member:
raise Exception("Not allowed")
return await self.services.container.get_status(channel_uid)
async def finalize_message(self, message_uid):
self._require_login()
message = await self.services.channel_message.get(message_uid)
@ -369,10 +362,26 @@ class RPCView(BaseView):
self._finalize_task = asyncio.create_task(self._finalize_message_task(message_uid, timeout=5))
return {"success": True}
async def clear_channel(self, channel_uid):
self._require_login()
user = await self.services.user.get(uid=self.user_uid)
if not user["is_admin"]:
raise Exception("Not allowed")
channel = await self.services.channel.get(uid=channel_uid)
if not channel:
raise Exception("Channel not found")
channel['history_start'] = datetime.now()
await self.services.channel.save(channel)
return await self.services.channel_message.clear(channel_uid)
async def send_message(self, channel_uid, message, is_final=True):
self._require_login()
await self.services.chat.send(self.user_uid, channel_uid, message, is_final)
message = await self.services.chat.send(
self.user_uid, channel_uid, message, is_final
)
return message["uid"]
async def echo(self, *args):
self._require_login()
@ -386,7 +395,13 @@ class RPCView(BaseView):
any(
keyword in lowercase
for keyword in [
"drop", "alter", "update", "delete", "replace", "insert", "truncate"
"drop",
"alter",
"update",
"delete",
"replace",
"insert",
"truncate",
]
)
and "select" not in lowercase
@ -396,13 +411,25 @@ class RPCView(BaseView):
dict(record) async for record in self.services.channel.query(args[0])
]
for record in records:
for field in ("email", "password", "message", "html"):
if field in record:
try:
del record[field]
except Exception:
pass
return records
try:
del record["email"]
except KeyError:
pass
try:
del record["password"]
except KeyError:
pass
try:
del record["message"]
except:
pass
try:
del record["html"]
except:
pass
return [
dict(record) async for record in self.services.channel.query(args[0])
]
async def __call__(self, data):
try:
@ -430,9 +457,10 @@ class RPCView(BaseView):
{"callId": call_id, "success": success, "data": result}
)
except Exception as ex:
print(str(ex), flush=True)
logger.exception(ex)
await self._send_json(
{"callId": data.get("callId"), "success": False, "data": str(ex)}
{"callId": call_id, "success": False, "data": str(ex)}
)
async def _send_json(self, obj):
@ -441,9 +469,10 @@ class RPCView(BaseView):
except Exception as ex:
await self.services.socket.delete(self.ws)
await self.ws.close()
async def get_online_users(self, channel_uid):
self._require_login()
results = [
record
async for record in self.services.channel.get_recent_users(channel_uid)
@ -451,8 +480,13 @@ class RPCView(BaseView):
results = sorted(results, key=lambda x: x["nick"])
return results
async def echo(self, obj):
await self.ws.send_json(obj)
return "noresponse"
async def get_recent_users(self, channel_uid):
self._require_login()
return [
{
"uid": record["uid"],
@ -465,6 +499,7 @@ class RPCView(BaseView):
async def get_users(self, channel_uid):
self._require_login()
return [
{
"uid": record["uid"],
@ -489,6 +524,7 @@ class RPCView(BaseView):
return {"pong": args}
async def stars_render(self, channel_uid, message):
for user in await self.get_online_users(channel_uid):
try:
await self.services.socket.send_to_user(
@ -499,7 +535,7 @@ class RPCView(BaseView):
},
)
except Exception as ex:
logger.exception(ex)
print(ex)
async def get(self):
scheduled = []
@ -522,7 +558,7 @@ class RPCView(BaseView):
await self.services.socket.subscribe(
ws, subscription["channel_uid"], self.request.session.get("uid")
)
if not scheduled and getattr(self.request.app, "uptime_seconds", 0) < 5:
if not scheduled and self.request.app.uptime_seconds < 5:
await schedule(
self.request.session.get("uid"),
0,
@ -531,7 +567,7 @@ class RPCView(BaseView):
await schedule(
self.request.session.get("uid"),
15,
{"event": "deployed", "data": {"uptime": getattr(self.request.app, "uptime", 0)}},
{"event": "deployed", "data": {"uptime": self.request.app.uptime}},
)
rpc = RPCView.RPCApi(self, ws)
@ -540,6 +576,7 @@ class RPCView(BaseView):
try:
await rpc(msg.json())
except Exception as ex:
print("Deleting socket", ex, flush=True)
logger.exception(ex)
await self.services.socket.delete(ws)
break
@ -548,4 +585,3 @@ class RPCView(BaseView):
elif msg.type == web.WSMsgType.CLOSE:
pass
return ws