cătălin
152546982c- Add missing actions and make a clear boundary between actions, services and nfra (i.e: actions shouldn't use stuff from infra/) - Delete stuff not in use: tts, gtts, etc - Add a ton of tests
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
from typing import Literal
|
|
|
|
from litestar import MediaType, Response, get, put
|
|
from litestar.datastructures import UploadFile
|
|
from litestar.response import Template
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
from huesoporro.actions.chatbot.create_or_update_chatbot import (
|
|
CreateOrUpdateChatbotAction,
|
|
)
|
|
from huesoporro.actions.chatbot.get_chatbot_by_user_id import GetChatbotByUserIdAction
|
|
from huesoporro.bot import BotsManager
|
|
from huesoporro.models import Chatbot, User
|
|
|
|
|
|
class ManageBotDTO(BaseModel):
|
|
command: Literal["start", "stop"]
|
|
channel_name: str | None = None
|
|
|
|
|
|
class ImportTextFileDTO(BaseModel):
|
|
file: UploadFile
|
|
channel_name: str
|
|
|
|
model_config = ConfigDict(arbitrary_types_allowed=True)
|
|
|
|
|
|
class UpdateChatbotDTO(BaseModel):
|
|
automatic_generation_timer: int = 300
|
|
automatic_quote_timer: int = 500
|
|
mods: list[str]
|
|
|
|
|
|
@get(
|
|
"/tts",
|
|
media_type=MediaType.HTML,
|
|
)
|
|
async def get_tts_overlay(user: User) -> Template:
|
|
return Template(template_name="tts.html")
|
|
|
|
|
|
@get(
|
|
"/tts/permalink",
|
|
media_type=MediaType.HTML,
|
|
)
|
|
async def get_tts_permalink(access_token: str) -> Template:
|
|
"""Handler for the /tts permalink endpoint to be used by apps that can only give the authentication as a query
|
|
param and not as a cookie, i.e. OBS"""
|
|
|
|
return Template(
|
|
template_name="tts.html",
|
|
)
|
|
|
|
|
|
@get(
|
|
"/",
|
|
media_type=MediaType.HTML,
|
|
)
|
|
async def get_index(
|
|
user: User, get_chatbot_by_user_id_action: GetChatbotByUserIdAction
|
|
) -> Template:
|
|
chatbot_settings = await get_chatbot_by_user_id_action.run(user_id=user.id)
|
|
return Template(
|
|
template_name="index.html",
|
|
context=chatbot_settings.model_dump() if chatbot_settings else {},
|
|
)
|
|
|
|
|
|
@put("/api/v1/bot")
|
|
async def manage_bot(
|
|
user: User,
|
|
data: ManageBotDTO,
|
|
create_or_update_chatbot_action: CreateOrUpdateChatbotAction,
|
|
get_chatbot_by_user_id_action: GetChatbotByUserIdAction,
|
|
bm: BotsManager,
|
|
) -> Response:
|
|
chatbot = await get_chatbot_by_user_id_action.run(
|
|
user_id=user.id
|
|
) or await create_or_update_chatbot_action.run(
|
|
user_id=user.id,
|
|
)
|
|
|
|
if data.command == "start":
|
|
if not data.channel_name:
|
|
return Response({"message": "Channel name is required"}, status_code=400)
|
|
bm.add_bot(user, data.channel_name, chatbot=chatbot) # type: ignore[arg-type]
|
|
if user.username in bm.bots:
|
|
await bm.run_user_bot(user)
|
|
return Response({"message": "Bot started"})
|
|
if data.command == "stop" and user.username in bm.bots:
|
|
await bm.stop_user_bot(user)
|
|
return Response({"message": "Bot stopped"})
|
|
return Response({"message": "Invalid command"}, status_code=400)
|
|
|
|
|
|
@get("/api/v1/bot")
|
|
async def get_bot_status(user: User, bm: BotsManager) -> dict:
|
|
if user.username not in bm.bots:
|
|
return {"status": "ko"}
|
|
return {"status": "ok"}
|
|
|
|
|
|
@get("/api/v1/bot/settings")
|
|
async def get_bot_settings(chatbot: Chatbot) -> Chatbot:
|
|
return chatbot
|
|
|
|
|
|
@put("/api/v1/bot/settings")
|
|
async def save_bot_settings(
|
|
user: User,
|
|
data: UpdateChatbotDTO,
|
|
create_or_update_chatbot_action: CreateOrUpdateChatbotAction,
|
|
) -> dict:
|
|
await create_or_update_chatbot_action.run(
|
|
user_id=user.id,
|
|
automatic_generation_timer=data.automatic_generation_timer,
|
|
automatic_quote_timer=data.automatic_quote_timer,
|
|
mods=data.mods,
|
|
)
|
|
return {"status": "ok"}
|