simonkoson
125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
"""
|
||
仪表盘 API — 题图上传 / 查询
|
||
|
||
权限：
|
||
- POST /cover — 仅 zhipianren / zebian（biandao 返回 403）
|
||
- GET /cover — 三角色均可读
|
||
"""
|
||
|
||
import os
|
||
import uuid
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, status
|
||
from sqlmodel import Session
|
||
|
||
from app.core.deps import get_current_user, require_role
|
||
from app.db.session import get_session
|
||
from app.models.user import User, UserRole
|
||
|
||
router = APIRouter(prefix="/api/dashboard", tags=["仪表盘"])
|
||
|
||
# 题图文件存放目录（相对于 backend/）
|
||
STATIC_COVERS_DIR = Path(__file__).parent.parent.parent / "static" / "covers"
|
||
STATIC_COVERS_URL = "/static/covers"
|
||
|
||
# 确保目录存在
|
||
STATIC_COVERS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
ALLOWED_TYPES = {"image/png", "image/jpeg", "image/webp"}
|
||
MAX_SIZE = 5 * 1024 * 1024 # 5MB
|
||
|
||
|
||
def require_upload_role():
|
||
return require_role(UserRole.zhipianren, UserRole.zebian)
|
||
|
||
|
||
@router.get("/cover")
|
||
def get_current_cover(
|
||
session: Session = Depends(get_session),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""查询当前题图设置（三角色均可读）。"""
|
||
from app.models.cover_settings import CoverSettings
|
||
|
||
row = session.exec(
|
||
select(CoverSettings).where(CoverSettings.key == "dashboard_cover")
|
||
).first()
|
||
|
||
if not row:
|
||
return {"cover_path": None, "episode_number": None, "episode_title": None}
|
||
|
||
return {
|
||
"cover_path": row.cover_path,
|
||
"episode_number": row.episode_number,
|
||
"episode_title": row.episode_title,
|
||
}
|
||
|
||
|
||
@router.post("/cover")
|
||
def upload_cover(
|
||
file: UploadFile = File(...),
|
||
episode_number: int = Form(...),
|
||
episode_title: str = Form(...),
|
||
session: Session = Depends(get_session),
|
||
current_user: User = Depends(require_upload_role()),
|
||
):
|
||
"""上传题图（仅制片人/责编）。
|
||
|
||
文件存 static/covers/{uuid}.{ext}，更新 cover_settings 表。
|
||
"""
|
||
# 校验文件类型
|
||
if file.content_type not in ALLOWED_TYPES:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=f"仅支持 PNG/JPG/WEBP，当前：{file.content_type}",
|
||
)
|
||
|
||
# 校验文件大小
|
||
contents = file.file.read()
|
||
if len(contents) > MAX_SIZE:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="文件大小不能超过 5MB",
|
||
)
|
||
file.file.seek(0)
|
||
|
||
# 生成文件名
|
||
ext = file.filename.split(".")[-1] if "." in file.filename else "jpg"
|
||
safe_ext = ext.lower() if ext.lower() in {"png", "jpg", "jpeg", "webp"} else "jpg"
|
||
filename = f"{uuid.uuid4().hex}.{safe_ext}"
|
||
filepath = STATIC_COVERS_DIR / filename
|
||
|
||
# 写文件
|
||
with open(filepath, "wb") as f:
|
||
f.write(contents)
|
||
|
||
# 更新数据库
|
||
from app.models.cover_settings import CoverSettings
|
||
|
||
cover_path = f"{STATIC_COVERS_URL}/{filename}"
|
||
now = datetime.now()
|
||
|
||
row = session.exec(
|
||
select(CoverSettings).where(CoverSettings.key == "dashboard_cover")
|
||
).first()
|
||
|
||
if row:
|
||
row.cover_path = cover_path
|
||
row.episode_number = episode_number
|
||
row.episode_title = episode_title
|
||
row.updated_at = now
|
||
else:
|
||
row = CoverSettings(
|
||
key="dashboard_cover",
|
||
cover_path=cover_path,
|
||
episode_number=episode_number,
|
||
episode_title=episode_title,
|
||
updated_at=now,
|
||
)
|
||
session.add(row)
|
||
|
||
session.commit()
|
||
|
||
return {"success": True, "cover_path": cover_path} |