DPM/backend/app/admin.py

# -*- coding: utf-8 -*-
"""管理后台页面（FastAPI + Jinja2 服务端渲染）
   GET  /admin          → 登录页 或 管理台（按 Cookie 鉴权）
   POST /admin/login    → 表单登录，成功写入签名 Cookie
   GET  /admin/logout   → 清除 Cookie
   媒体库 / 播放列表 / 设置均由 Jinja 服务端渲染，动态操作用少量 JS 调 REST
"""

import hashlib
import hmac
import logging

from fastapi import Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from pathlib import Path

from .config import settings
from .storage import storage
from .routers import _list_media, _playlist_files

log = logging.getLogger("dpm.admin")

TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "templates"
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))


# ---------------- 签名 Cookie ----------------

def _admin_token():
    return hmac.new(settings.ADMIN_SECRET.encode(), b"dpm-admin", hashlib.sha256).hexdigest()


def _verify_admin_cookie(cookie):
    return bool(cookie) and hmac.compare_digest(cookie, _admin_token())


def _set_admin_cookie(resp):
    resp.set_cookie("dpm_admin", _admin_token(), max_age=7 * 86400, httponly=True, samesite="lax")
    return resp


def _clear_admin_cookie(resp):
    resp.delete_cookie("dpm_admin")
    return resp


# ---------------- 路由 ----------------

async def admin_page(request: Request):
    if not _verify_admin_cookie(request.cookies.get("dpm_admin")):
        return templates.TemplateResponse(
            request=request,
            name="admin.html",
            context={
                "logged_in": False,
                "error": request.query_params.get("error") == "1",
            },
        )

    files = _list_media()["files"]
    playlist = _playlist_files()
    s = storage.get_settings()
    return templates.TemplateResponse(
        request=request,
        name="admin.html",
        context={
            "logged_in": True,
            "error": False,
            "files": files,
            "playlist": playlist["files"],
            "play_mode": s.get("play_mode", "sequential"),
            "image_duration": s.get("image_duration", 5),
            "volume": s.get("volume", 80),
            "sfx_volume": s.get("sfx_volume", 60),
            "fullscreen": s.get("fullscreen", True),
            "autostart": s.get("autostart", False),
            "from_screen": request.query_params.get("from") == "screen",
        },
    )


async def admin_login(request: Request):
    form = await request.form()
    s = storage.get_settings()
    if form.get("username") == s.get("username") and form.get("password") == s.get("password"):
        return _set_admin_cookie(RedirectResponse("/admin", status_code=303))
    return RedirectResponse("/admin?error=1", status_code=303)


async def admin_logout():
    return _clear_admin_cookie(RedirectResponse("/admin", status_code=303))