Jared Vititoe
50de731a82
Modular bot using matrix-nio[e2e] with E2EE support, deployed as systemd service on Synapse LXC. Includes 10 commands: help, ping, 8ball, fortune, flip, roll, random, rps, poll, champion. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
82 lines
2.2 KiB
Python
82 lines
2.2 KiB
Python
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from nio import AsyncClient, RoomSendResponse
|
|
|
|
from config import MAX_INPUT_LENGTH
|
|
|
|
|
|
def setup_logging(level="INFO"):
|
|
Path("logs").mkdir(exist_ok=True)
|
|
|
|
logger = logging.getLogger("matrixbot")
|
|
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
|
|
|
|
file_handler = RotatingFileHandler(
|
|
"logs/matrixbot.log",
|
|
maxBytes=10 * 1024 * 1024,
|
|
backupCount=5,
|
|
)
|
|
file_handler.setFormatter(
|
|
logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
|
)
|
|
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setFormatter(
|
|
logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
|
)
|
|
|
|
logger.addHandler(file_handler)
|
|
logger.addHandler(stream_handler)
|
|
return logger
|
|
|
|
|
|
async def send_text(client: AsyncClient, room_id: str, text: str):
|
|
logger = logging.getLogger("matrixbot")
|
|
resp = await client.room_send(
|
|
room_id,
|
|
message_type="m.room.message",
|
|
content={"msgtype": "m.text", "body": text},
|
|
)
|
|
if not isinstance(resp, RoomSendResponse):
|
|
logger.error("send_text failed: %s", resp)
|
|
return resp
|
|
|
|
|
|
async def send_html(client: AsyncClient, room_id: str, plain: str, html: str):
|
|
logger = logging.getLogger("matrixbot")
|
|
resp = await client.room_send(
|
|
room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": plain,
|
|
"format": "org.matrix.custom.html",
|
|
"formatted_body": html,
|
|
},
|
|
)
|
|
if not isinstance(resp, RoomSendResponse):
|
|
logger.error("send_html failed: %s", resp)
|
|
return resp
|
|
|
|
|
|
async def send_reaction(client: AsyncClient, room_id: str, event_id: str, emoji: str):
|
|
return await client.room_send(
|
|
room_id,
|
|
message_type="m.reaction",
|
|
content={
|
|
"m.relates_to": {
|
|
"rel_type": "m.annotation",
|
|
"event_id": event_id,
|
|
"key": emoji,
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def sanitize_input(text: str, max_length: int = MAX_INPUT_LENGTH) -> str:
|
|
text = text.strip()[:max_length]
|
|
text = "".join(char for char in text if char.isprintable())
|
|
return text
|