196 lines
8 KiB
Python
196 lines
8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from app.api.deps import get_current_username, get_current_username_optional
|
|
from app.models import ApplicationStatus
|
|
from app.security import verify_password
|
|
|
|
|
|
def build_router(templates: Jinja2Templates) -> APIRouter:
|
|
router = APIRouter()
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
async def root(request: Request, username: str | None = Depends(get_current_username_optional)):
|
|
if username:
|
|
return RedirectResponse("/applications", status_code=303)
|
|
return RedirectResponse("/login", status_code=303)
|
|
|
|
@router.get("/health")
|
|
async def health(request: Request):
|
|
return {
|
|
"ok": True,
|
|
"app": request.app.state.settings.app_name,
|
|
"env": request.app.state.settings.app_env,
|
|
"bot_enabled": bool(getattr(request.app.state.bot_runtime, "enabled", False)),
|
|
}
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
async def login_get(request: Request):
|
|
if request.session.get("username"):
|
|
return RedirectResponse("/applications", status_code=303)
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="login.html",
|
|
context={"error": None},
|
|
)
|
|
|
|
@router.post("/login", response_class=HTMLResponse)
|
|
async def login_post(request: Request, username: str = Form(...), password: str = Form(...)):
|
|
user = request.app.state.container.users.get_by_username(username.strip())
|
|
if not user or not verify_password(password, user["password_hash"]):
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="login.html",
|
|
context={"error": "Неверный логин или пароль"},
|
|
status_code=401,
|
|
)
|
|
request.session["username"] = user["username"]
|
|
request.session["role"] = user["role"]
|
|
return RedirectResponse("/applications", status_code=303)
|
|
|
|
@router.post("/logout")
|
|
async def logout(request: Request):
|
|
request.session.clear()
|
|
return RedirectResponse("/login", status_code=303)
|
|
|
|
@router.get("/applications", response_class=HTMLResponse)
|
|
async def applications_page(
|
|
request: Request,
|
|
status_filter: str | None = None,
|
|
search: str | None = None,
|
|
username: str = Depends(get_current_username),
|
|
):
|
|
apps = request.app.state.container.app_service.list_applications(status=status_filter, search=search)
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="applications.html",
|
|
context={
|
|
"username": username,
|
|
"applications": apps,
|
|
"status_filter": status_filter or "",
|
|
"search": search or "",
|
|
"statuses": [s.value for s in ApplicationStatus],
|
|
},
|
|
)
|
|
|
|
@router.get("/applications/{application_id}", response_class=HTMLResponse)
|
|
async def application_page(
|
|
request: Request,
|
|
application_id: int,
|
|
username: str = Depends(get_current_username),
|
|
):
|
|
app = request.app.state.container.app_service.get_application(application_id)
|
|
if not app:
|
|
raise HTTPException(status_code=404, detail="Application not found")
|
|
events = request.app.state.container.app_service.get_events(application_id)
|
|
attachments = json.loads(app.attachments_json) if app.attachments_json else []
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="application_detail.html",
|
|
context={
|
|
"username": username,
|
|
"application": app,
|
|
"events": events,
|
|
"attachments": attachments,
|
|
"statuses": [s.value for s in ApplicationStatus],
|
|
},
|
|
)
|
|
|
|
@router.post("/applications/{application_id}/status")
|
|
async def application_status_change(
|
|
request: Request,
|
|
application_id: int,
|
|
status_value: str = Form(...),
|
|
reason: str = Form(""),
|
|
username: str = Depends(get_current_username),
|
|
):
|
|
if status_value not in {s.value for s in ApplicationStatus}:
|
|
raise HTTPException(status_code=400, detail="Invalid status")
|
|
updated = request.app.state.container.app_service.update_status(
|
|
application_id,
|
|
status=status_value,
|
|
actor=f"web:{username}",
|
|
reason=reason or None,
|
|
)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Application not found")
|
|
return RedirectResponse(f"/applications/{application_id}", status_code=303)
|
|
|
|
@router.get("/chats", response_class=HTMLResponse)
|
|
async def chats_page(request: Request, username: str = Depends(get_current_username)):
|
|
chats = request.app.state.container.chats.list(limit=300)
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="chats.html",
|
|
context={"username": username, "chats": chats},
|
|
)
|
|
|
|
@router.post("/chats/sync")
|
|
async def chats_sync(request: Request, _: str = Depends(get_current_username)):
|
|
runtime = request.app.state.bot_runtime
|
|
counters = await runtime.sync_chats_from_api()
|
|
return JSONResponse(counters)
|
|
|
|
@router.post("/max/webhook")
|
|
async def max_webhook(request: Request):
|
|
payload = await request.json()
|
|
secret_expected = request.app.state.settings.max_webhook_secret
|
|
if secret_expected:
|
|
header_secret = request.headers.get("x-max-webhook-secret", "")
|
|
if header_secret != secret_expected:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid webhook secret")
|
|
await request.app.state.bot_runtime.handle_webhook_payload(payload)
|
|
return {"ok": True}
|
|
|
|
@router.get("/api/applications")
|
|
async def api_applications(
|
|
request: Request,
|
|
status_filter: str | None = None,
|
|
search: str | None = None,
|
|
_: str = Depends(get_current_username),
|
|
):
|
|
apps = request.app.state.container.app_service.list_applications(status=status_filter, search=search)
|
|
return [a.__dict__ for a in apps]
|
|
|
|
@router.get("/api/applications/{application_id}")
|
|
async def api_application(request: Request, application_id: int, _: str = Depends(get_current_username)):
|
|
app = request.app.state.container.app_service.get_application(application_id)
|
|
if not app:
|
|
raise HTTPException(status_code=404, detail="Application not found")
|
|
events = request.app.state.container.app_service.get_events(application_id)
|
|
return {"application": app.__dict__, "events": [e.__dict__ for e in events]}
|
|
|
|
@router.post("/api/applications/{application_id}/status")
|
|
async def api_application_status(
|
|
request: Request,
|
|
application_id: int,
|
|
payload: dict[str, Any],
|
|
username: str = Depends(get_current_username),
|
|
):
|
|
status_value = str(payload.get("status", ""))
|
|
reason = payload.get("reason")
|
|
if status_value not in {s.value for s in ApplicationStatus}:
|
|
raise HTTPException(status_code=400, detail="Invalid status")
|
|
updated = request.app.state.container.app_service.update_status(
|
|
application_id,
|
|
status=status_value,
|
|
actor=f"web:{username}",
|
|
reason=reason,
|
|
)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Application not found")
|
|
return {"ok": True, "application": updated.__dict__}
|
|
|
|
@router.get("/api/chats")
|
|
async def api_chats(request: Request, _: str = Depends(get_current_username)):
|
|
chats = request.app.state.container.chats.list(limit=300)
|
|
return [c.__dict__ for c in chats]
|
|
|
|
return router
|
|
|