from typing import Optional from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import HTMLResponse from sqlmodel import Session, select from app.auth import require_user from app.database import get_session from app.models import ActivityLog, EventAction, MediaFile from app.templates import templates from app.ws import manager router = APIRouter(prefix="/actions") EVENT_TYPES = ["sub", "gift_sub", "bits", "channel_points", "command", "follow", "raid"] ACTION_TYPES = ["sound", "video", "alert"] @router.get("", response_class=HTMLResponse) async def actions_page(request: Request, session: Session = Depends(get_session), user=Depends(require_user)): actions = session.exec(select(EventAction)).all() media_files = session.exec(select(MediaFile)).all() return templates.TemplateResponse("actions.html", { "request": request, "user": user, "actions": actions, "media_files": media_files, "event_types": EVENT_TYPES, "action_types": ACTION_TYPES, }) @router.post("", response_class=HTMLResponse) async def create_action( request: Request, event_type: str = Form(...), event_detail: Optional[str] = Form(default=None), action_type: str = Form(...), media_file_id: Optional[int] = Form(default=None), alert_text: Optional[str] = Form(default=None), cooldown_seconds: int = Form(default=0), session: Session = Depends(get_session), user=Depends(require_user), ): action = EventAction( event_type=event_type, event_detail=event_detail.strip() if event_detail and event_detail.strip() else None, action_type=action_type, media_file_id=media_file_id or None, alert_text=alert_text.strip() if alert_text and alert_text.strip() else None, cooldown_seconds=max(0, cooldown_seconds), ) session.add(action) session.commit() session.refresh(action) media_files = session.exec(select(MediaFile)).all() return templates.TemplateResponse("partials/action_row.html", { "request": request, "action": action, "media_files": media_files, }) @router.post("/{action_id}/test", response_class=HTMLResponse) async def test_action( request: Request, action_id: int, session: Session = Depends(get_session), user=Depends(require_user), ): action = session.get(EventAction, action_id) if not action: raise HTTPException(status_code=404) media_url = None if action.media_file_id: media = session.get(MediaFile, action.media_file_id) if media: media_url = f"/static/media/{media.filename}" await manager.broadcast({ "type": "execute", "action_type": action.action_type, "media_url": media_url, "alert_text": action.alert_text or "Test!", }) session.add(ActivityLog( event_type="test", username=user.twitch_display_name, detail=f"{action.event_type}:{action.event_detail or 'any'}", action_taken=f"{action.action_type}: {media_url or action.alert_text or ''}", )) session.commit() media_files = session.exec(select(MediaFile)).all() return templates.TemplateResponse("partials/action_row.html", { "request": request, "action": action, "media_files": media_files, "just_tested": True, }) @router.delete("/{action_id}", response_class=HTMLResponse) async def delete_action( action_id: int, session: Session = Depends(get_session), user=Depends(require_user), ): action = session.get(EventAction, action_id) if not action: raise HTTPException(status_code=404) session.delete(action) session.commit() return HTMLResponse("") @router.post("/{action_id}/toggle", response_class=HTMLResponse) async def toggle_action( request: Request, action_id: int, session: Session = Depends(get_session), user=Depends(require_user), ): action = session.get(EventAction, action_id) if not action: raise HTTPException(status_code=404) action.enabled = not action.enabled session.add(action) session.commit() session.refresh(action) media_files = session.exec(select(MediaFile)).all() return templates.TemplateResponse("partials/action_row.html", { "request": request, "action": action, "media_files": media_files, })