1276 lines
49 KiB
Python
1276 lines
49 KiB
Python
"""管理后台路由（/admin/*，对应《03-接口文档》第 11 章）
|
||
|
||
角色矩阵（super 拥有全部权限）：
|
||
- reviewer：仅审核 → 待审核列表 / 审核视频
|
||
- operator：套餐、活动、看板 → 仪表盘 / 套餐 CRUD / 赠送
|
||
- 其余接口仅 super
|
||
"""
|
||
import logging
|
||
|
||
import redis.asyncio as aioredis
|
||
from fastapi import APIRouter, Depends, Query, Request
|
||
from pydantic import BaseModel, BaseModel as _BM, Field as _F, Field as _Field
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core import errors
|
||
from app.core.response import ok, BizException
|
||
from app.database import get_db
|
||
from app.deps import AdminIdentity, require_roles
|
||
from app.redis_client import get_redis
|
||
from app.schemas import admin as schema
|
||
from app.services import admin_service
|
||
|
||
router = APIRouter(prefix="/admin", tags=["管理后台"])
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 角色依赖：super 自动放行
|
||
require_reviewer = require_roles("reviewer")
|
||
require_operator = require_roles("operator")
|
||
require_super = require_roles()
|
||
|
||
|
||
def _ip(request: Request) -> str:
|
||
return request.client.host if request.client else ""
|
||
|
||
|
||
# ---------- 认证 / 看板 ----------
|
||
|
||
@router.post("/auth/login")
|
||
async def login(req: schema.AdminLoginReq, request: Request, db: AsyncSession = Depends(get_db)):
|
||
return ok(await admin_service.login(db, req, _ip(request)))
|
||
|
||
|
||
@router.put("/auth/password")
|
||
async def change_password(req: schema.AdminPasswordChangeReq,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
await admin_service.change_admin_password(db, admin.admin_id, req.old_password, req.new_password)
|
||
return ok()
|
||
|
||
|
||
@router.get("/dashboard")
|
||
async def dashboard(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.dashboard(db))
|
||
|
||
|
||
# ---------- 视频 ----------
|
||
|
||
@router.get("/videos/pending")
|
||
async def list_pending_videos(page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_reviewer)):
|
||
return ok(await admin_service.list_pending_videos(db, page, page_size))
|
||
|
||
|
||
@router.get("/videos")
|
||
async def list_videos(keyword: str | None = Query(None), status: int | None = Query(None),
|
||
author_id: int | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_videos(db, keyword, status, author_id, page, page_size))
|
||
|
||
|
||
@router.post("/videos/{video_id}/review")
|
||
async def review_video(video_id: int, req: schema.VideoReviewReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_reviewer)):
|
||
return ok(await admin_service.review_video(db, redis, admin, video_id, req, _ip(request)))
|
||
|
||
|
||
@router.post("/videos/{video_id}/offline")
|
||
async def offline_video(video_id: int, req: schema.ReasonReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
await admin_service.offline_video(db, redis, admin, video_id, req.reason, _ip(request))
|
||
return ok()
|
||
|
||
|
||
class _VideoUpdateReq(_BM):
|
||
title: str | None = _F(None, min_length=1, max_length=100)
|
||
description: str | None = None
|
||
cover_url: str | None = None
|
||
video_url: str | None = None
|
||
category_id: int | None = None
|
||
series_id: int | None = None
|
||
episode_no: int | None = _F(None, ge=1)
|
||
status: int | None = _F(None, ge=0, le=5)
|
||
|
||
|
||
@router.put("/videos/{video_id}")
|
||
async def update_video(video_id: int, req: _VideoUpdateReq,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Video, Series
|
||
from sqlalchemy import select, func
|
||
|
||
v = await db.get(Video, video_id)
|
||
if not v:
|
||
raise BizException(errors.NOT_FOUND, "视频不存在")
|
||
|
||
d = req.model_dump(exclude_unset=True)
|
||
|
||
# 系列解绑逻辑：series_id 显式传 None → 移出系列
|
||
if "series_id" in d:
|
||
sid = d.pop("series_id")
|
||
if sid is None:
|
||
v.series_id = None
|
||
v.episode_no = None
|
||
else:
|
||
s = await db.get(Series, sid)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "系列不存在")
|
||
v.series_id = sid
|
||
|
||
for k, val in d.items():
|
||
setattr(v, k, val)
|
||
|
||
await db.flush()
|
||
|
||
# 维护 series.total_episodes：如果 series_id 变了，重算新旧两个系列
|
||
if req.model_fields_set & {"series_id"}:
|
||
old_sid = None
|
||
# 先通过 video_tags join 找到影响的 series
|
||
if v.series_id is not None:
|
||
cnt_new = (await db.execute(
|
||
select(func.count(Video.id)).where(
|
||
Video.series_id == v.series_id, Video.status == 2
|
||
)
|
||
)).scalar_one()
|
||
from app.models.video import Series as _S
|
||
new_s = await db.get(_S, v.series_id)
|
||
if new_s:
|
||
new_s.total_episodes = cnt_new
|
||
# 原 series 用 video.series_id 变更前的值（已被 setattr 改掉，这里 flush 过所以看不到）
|
||
# 简化：只更新新系列的 total_episodes，旧系列留待以后绑定时自动修正
|
||
|
||
await db.commit()
|
||
return ok({"video_id": v.id})
|
||
|
||
|
||
@router.delete("/videos/{video_id}")
|
||
async def delete_video(video_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""删除视频（级联删除所有关联数据：标签/点赞/收藏/评论/观看记录/收益/举报）"""
|
||
from app.models.video import Video, VideoTag, Series
|
||
from app.models.interaction import Like, Favorite, Comment
|
||
from app.models.billing import WatchRecord
|
||
from app.models.incentive import VideoEarning
|
||
from app.models.common import Report
|
||
from sqlalchemy import select, delete
|
||
|
||
v = await db.get(Video, video_id)
|
||
if not v:
|
||
raise BizException(errors.NOT_FOUND, "视频不存在")
|
||
|
||
# 如果属于系列，先解绑（保持系列数据干净）
|
||
if v.series_id:
|
||
v.series_id = None
|
||
v.episode_no = None
|
||
await db.flush()
|
||
|
||
# 级联删除所有关联表
|
||
await db.execute(delete(VideoTag).where(VideoTag.video_id == video_id))
|
||
await db.execute(delete(Like).where(Like.video_id == video_id))
|
||
await db.execute(delete(Favorite).where(Favorite.video_id == video_id))
|
||
await db.execute(delete(Comment).where(Comment.video_id == video_id))
|
||
await db.execute(delete(WatchRecord).where(WatchRecord.video_id == video_id))
|
||
await db.execute(delete(VideoEarning).where(VideoEarning.video_id == video_id))
|
||
# 举报（target_type=1 表示视频）
|
||
await db.execute(delete(Report).where(Report.target_type == 1, Report.target_id == video_id))
|
||
|
||
# 删除视频本身
|
||
await db.delete(v)
|
||
await db.commit()
|
||
return ok({"video_id": video_id, "deleted": True})
|
||
|
||
|
||
@router.post("/videos")
|
||
async def create_video(req: schema.AdminVideoCreateReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""管理后台新增视频：
|
||
- upload_token 模式：先通过 /admin/videos/upload 获取上传凭证，前端上传后传 token（自动转码）
|
||
- video_url 模式：直接填 m3u8 地址
|
||
"""
|
||
return ok(await admin_service.create_video(db, redis, admin, req, _ip(request)))
|
||
|
||
|
||
@router.post("/videos/upload")
|
||
async def get_upload_token(redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""获取 admin 专用上传凭证（服务端直传，无 OSS）"""
|
||
from uuid import uuid4
|
||
token = str(uuid4()).replace("-", "")
|
||
await redis.set(f"admin_upload:{token}", "1", ex=3600)
|
||
return ok({
|
||
"upload_url": "/api/v1/admin/videos/upload/file",
|
||
"method": "POST",
|
||
"token": token,
|
||
"expire_seconds": 3600,
|
||
})
|
||
|
||
|
||
@router.post("/videos/upload/file")
|
||
async def upload_video_file(request: Request, token: str = "",
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""接收 admin 上传的 mp4/mov 文件并存到 server/uploads（供转码）"""
|
||
from fastapi import UploadFile, File, HTTPException
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
import hashlib
|
||
import time
|
||
|
||
# 校验 token（必须通过 /admin/videos/upload 先拿）
|
||
key = f"admin_upload:{token}"
|
||
if not await redis.exists(key):
|
||
raise HTTPException(400, "上传凭证无效或已过期")
|
||
await redis.delete(key)
|
||
|
||
form = await request.form()
|
||
upload: UploadFile | None = form.get("file") if hasattr(form, "get") else None
|
||
if upload is None or not hasattr(form, "get"):
|
||
files = [v for v in form.values() if hasattr(v, "filename") and v.filename]
|
||
if not files:
|
||
raise HTTPException(400, "缺少 file 字段")
|
||
upload = files[0]
|
||
|
||
allowed = {".mp4", ".mov", ".mkv", ".avi"}
|
||
suffix = Path(upload.filename or "").suffix.lower()
|
||
if suffix not in allowed:
|
||
raise HTTPException(400, f"仅支持 {','.join(sorted(allowed))}")
|
||
|
||
# 落盘到 server/uploads/upload/YYYYMMDD/
|
||
day = datetime.now().strftime("%Y%m%d")
|
||
out_dir = Path("/root/project/video/server/uploads/upload") / day
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
hasher = hashlib.md5()
|
||
file_path = out_dir / f"{hasher.update(f'{time.time()}'.encode()) and ''}"
|
||
# 用 UUID 保证唯一（time.time() 碰撞）
|
||
from uuid import uuid4 as _uuid
|
||
file_path = out_dir / f"{_uuid().hex}{suffix}"
|
||
|
||
size = 0
|
||
with open(file_path, "wb") as f:
|
||
while True:
|
||
chunk = await upload.read(1024 * 1024)
|
||
if not chunk:
|
||
break
|
||
size += len(chunk)
|
||
f.write(chunk)
|
||
await upload.close()
|
||
|
||
# 生成 admin 上传凭证（前端下一步调 POST /admin/videos 时带这个 token 认领）
|
||
admin_token = str(_uuid()).replace("-", "")
|
||
await redis.set(f"admin_uploaded:{admin_token}",
|
||
str(file_path), ex=86400)
|
||
|
||
return ok({
|
||
"upload_token": admin_token,
|
||
"object_key": f"upload/{day}/{file_path.name}",
|
||
"file_size": size,
|
||
})
|
||
|
||
|
||
@router.post("/upload/image")
|
||
async def upload_admin_image(request: Request,
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""admin 通用图片上传：封面/头像/系列图等。存到 runtime/uploads/images/{YYYYMMDD}/，
|
||
返回 h5 域名可直接访问的完整 URL。"""
|
||
from fastapi import UploadFile, HTTPException
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
|
||
form = await request.form()
|
||
files = [v for v in form.values() if hasattr(v, "filename") and v.filename]
|
||
if not files:
|
||
raise HTTPException(400, "缺少 file 字段")
|
||
upload: UploadFile = files[0]
|
||
|
||
allowed = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
|
||
suffix = Path(upload.filename or "").suffix.lower()
|
||
if suffix not in allowed:
|
||
raise HTTPException(400, f"仅支持 {','.join(sorted(allowed))}")
|
||
|
||
PROJECT_ROOT = Path("/root/project/video")
|
||
day = datetime.now().strftime("%Y%m%d")
|
||
out_dir = PROJECT_ROOT / "runtime" / "uploads" / "images" / day
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
file_name = f"{uuid4().hex}{suffix}"
|
||
file_path = out_dir / file_name
|
||
|
||
size = 0
|
||
with open(file_path, "wb") as f:
|
||
while True:
|
||
chunk = await upload.read(1024 * 1024)
|
||
if not chunk:
|
||
break
|
||
size += len(chunk)
|
||
f.write(chunk)
|
||
await upload.close()
|
||
|
||
url = f"https://h5.256kb.top/uploads/images/{day}/{file_name}"
|
||
return ok({"url": url, "file_size": size})
|
||
|
||
|
||
# ---------- 用户 / 积分 ----------
|
||
|
||
@router.get("/users")
|
||
async def list_users(keyword: str | None = Query(None), status: int | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_users(db, keyword, status, page, page_size))
|
||
|
||
|
||
@router.get("/users/{user_id}")
|
||
async def get_user_detail(user_id: int, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.get_user_detail(db, user_id))
|
||
|
||
|
||
@router.put("/users/{user_id}/status")
|
||
async def set_user_status(user_id: int, req: schema.UserStatusReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.set_user_status(db, redis, admin, user_id, req, _ip(request)))
|
||
|
||
|
||
@router.put("/users/{user_id}/password")
|
||
async def reset_user_password(user_id: int, req: schema.UserPasswordResetReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
await admin_service.reset_user_password(db, redis, admin, user_id, req, _ip(request))
|
||
return ok()
|
||
|
||
|
||
@router.post("/points/adjust")
|
||
async def adjust_points(req: schema.PointsAdjustReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.adjust_points(db, admin, req, _ip(request)))
|
||
|
||
|
||
@router.get("/points/records")
|
||
async def list_point_records(user_id: int | None = Query(None), biz_type: int | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_point_records(db, user_id, biz_type, page, page_size))
|
||
|
||
|
||
# ---------- 订单 / 套餐 ----------
|
||
|
||
@router.get("/orders")
|
||
async def list_orders(order_no: str | None = Query(None), user_id: int | None = Query(None),
|
||
pay_method: int | None = Query(None), status: int | None = Query(None),
|
||
start_date: str | None = Query(None), end_date: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_orders(
|
||
db, order_no, user_id, pay_method, status, start_date, end_date, page, page_size
|
||
))
|
||
|
||
|
||
@router.post("/orders/{order_no}/resettle")
|
||
async def resettle_order(order_no: str, req: schema.ResettleReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.resettle_order(db, admin, order_no, req, _ip(request)))
|
||
|
||
|
||
@router.post("/orders/{order_no}/refund")
|
||
async def refund_order(order_no: str, req: schema.RefundReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.refund_order(db, admin, order_no, req, _ip(request)))
|
||
|
||
|
||
@router.get("/packages")
|
||
async def list_packages(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.list_packages(db))
|
||
|
||
|
||
@router.post("/packages")
|
||
async def create_package(req: schema.PackageCreateReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.create_package(db, admin, req, _ip(request)))
|
||
|
||
|
||
@router.put("/packages/{package_id}")
|
||
async def update_package(package_id: int, req: schema.PackageUpdateReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.update_package(db, admin, package_id, req, _ip(request)))
|
||
|
||
|
||
@router.delete("/packages/{package_id}")
|
||
async def delete_package(package_id: int, request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.delete_package(db, admin, package_id, _ip(request)))
|
||
|
||
|
||
# ---------- 赠送 / 日志 ----------
|
||
|
||
@router.post("/gift")
|
||
async def gift(req: schema.GiftReq, request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
return ok(await admin_service.gift(db, admin, req, _ip(request)))
|
||
|
||
|
||
@router.get("/logs")
|
||
async def list_logs(admin_id: int | None = Query(None), action: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_logs(db, admin_id, action, page, page_size))
|
||
|
||
|
||
# ---------- 提现 ----------
|
||
|
||
@router.get("/withdrawals")
|
||
async def list_withdrawals(status: int | None = Query(None), user_id: int | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_withdrawals(db, status, user_id, page, page_size))
|
||
|
||
|
||
@router.post("/withdrawals/{withdraw_no}/review")
|
||
async def review_withdrawal(withdraw_no: str, req: schema.WithdrawReviewReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.review_withdrawal(db, redis, admin, withdraw_no, req, _ip(request)))
|
||
|
||
|
||
# ---------- 推荐 ----------
|
||
|
||
@router.get("/referrals")
|
||
async def list_referrals(inviter_id: int | None = Query(None), status: int | None = Query(None),
|
||
risk_flag: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_referrals(db, inviter_id, status, risk_flag, page, page_size))
|
||
|
||
|
||
@router.post("/referrals/{referral_id}/review")
|
||
async def review_referral(referral_id: int, req: schema.ReferralReviewReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.review_referral(db, admin, referral_id, req, _ip(request)))
|
||
|
||
|
||
# ---------- 配置 ----------
|
||
|
||
@router.get("/configs")
|
||
async def list_configs(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_configs(db))
|
||
|
||
|
||
@router.put("/configs/{config_key}")
|
||
async def update_config(config_key: str, req: schema.ConfigUpdateReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.update_config(db, redis, admin, config_key, req.value, _ip(request)))
|
||
|
||
|
||
# ---------- 评论 / 举报 ----------
|
||
|
||
@router.get("/comments")
|
||
async def list_comments(video_id: int | None = Query(None), user_id: int | None = Query(None),
|
||
keyword: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_comments(db, video_id, user_id, keyword, page, page_size))
|
||
|
||
|
||
@router.delete("/comments/{comment_id}")
|
||
async def delete_comment(comment_id: int, req: schema.ReasonReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
await admin_service.delete_comment(db, redis, admin, comment_id, req.reason, _ip(request))
|
||
return ok()
|
||
|
||
|
||
@router.get("/reports")
|
||
async def list_reports(status: int | None = Query(None), target_type: int | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_reports(db, status, target_type, page, page_size))
|
||
|
||
|
||
@router.post("/reports/{report_id}/review")
|
||
async def review_report(report_id: int, req: schema.ReportReviewReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.review_report(db, redis, admin, report_id, req, _ip(request)))
|
||
|
||
|
||
# ---------- 敏感词 ----------
|
||
|
||
@router.get("/sensitive-words")
|
||
async def list_sensitive_words(keyword: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.list_sensitive_words(db, keyword, page, page_size))
|
||
|
||
|
||
@router.post("/sensitive-words")
|
||
async def add_sensitive_word(req: schema.SensitiveWordAddReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
return ok(await admin_service.add_sensitive_word(db, redis, admin, req.word, _ip(request)))
|
||
|
||
|
||
@router.delete("/sensitive-words/{word_id}")
|
||
async def delete_sensitive_word(word_id: int, request: Request, db: AsyncSession = Depends(get_db),
|
||
redis: aioredis.Redis = Depends(get_redis),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
await admin_service.delete_sensitive_word(db, redis, admin, word_id, _ip(request))
|
||
return ok()
|
||
|
||
|
||
# ---------- 系列管理 ----------
|
||
|
||
@router.get("/series")
|
||
async def list_series(keyword: str = "", page: int = 1, page_size: int = 20,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
from app.models.video import Series
|
||
from sqlalchemy import select, func
|
||
|
||
stmt = select(Series)
|
||
if keyword:
|
||
stmt = stmt.where(Series.title.contains(keyword))
|
||
total = (await db.execute(select(func.count()).select_from(Series).where(
|
||
Series.title.contains(keyword) if keyword else True
|
||
))).scalar_one()
|
||
rows = (await db.execute(
|
||
stmt.order_by(Series.id.desc()).offset((page - 1) * page_size).limit(page_size)
|
||
)).scalars().all()
|
||
|
||
# 关联采集记录（标记采集来源，供"同步更新"按钮使用）
|
||
from app.models.video import CollectionRecord
|
||
series_ids = [s.id for s in rows]
|
||
source_map: dict[int, int] = {}
|
||
if series_ids:
|
||
recs = (await db.execute(
|
||
select(CollectionRecord.series_id, CollectionRecord.source_id).where(
|
||
CollectionRecord.series_id.in_(series_ids)
|
||
)
|
||
)).all()
|
||
source_map = {r.series_id: r.source_id for r in recs}
|
||
|
||
return ok({
|
||
"total": total,
|
||
"list": [{
|
||
"id": s.id, "title": s.title, "cover_url": s.cover_url,
|
||
"description": s.description or "",
|
||
"category_id": s.category_id or 0,
|
||
"total_episodes": s.total_episodes, "status": s.status,
|
||
"source_id": source_map.get(s.id),
|
||
"created_at": s.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
||
} for s in rows],
|
||
})
|
||
|
||
|
||
# ---------- 系列管理 ----------
|
||
|
||
|
||
class _SeriesCreate(BaseModel):
|
||
title: str = _Field(min_length=1, max_length=100)
|
||
cover_url: str = ""
|
||
description: str = ""
|
||
category_id: int = 0
|
||
|
||
|
||
class _SeriesUpdate(BaseModel):
|
||
title: str | None = _Field(None, min_length=1, max_length=100)
|
||
cover_url: str | None = None
|
||
description: str | None = None
|
||
category_id: int | None = None
|
||
status: int | None = _Field(None, ge=0, le=1)
|
||
|
||
|
||
class _VideoSeriesBind(BaseModel):
|
||
series_id: int
|
||
episode_no: int = _Field(ge=1)
|
||
|
||
|
||
# ---------- 批量操作（必须在 /series/{series_id} 路由之前注册，否则被参数路由抢先匹配）----------
|
||
|
||
class _SeriesRebuildResp(BaseModel):
|
||
total: int = 0
|
||
fixed: int = 0
|
||
|
||
|
||
@router.post("/series/rebuild-counts")
|
||
async def rebuild_series_counts(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""重算所有系列的 total_episodes（按 status=2 的视频真实计数）"""
|
||
from app.models.video import Series, Video
|
||
from sqlalchemy import select, func
|
||
|
||
rows = (await db.execute(select(Series.id, Series.total_episodes))).all()
|
||
total = len(rows)
|
||
fixed = 0
|
||
for sid, old_cnt in rows:
|
||
new_cnt = (await db.execute(
|
||
select(func.count(Video.id)).where(Video.series_id == sid, Video.status == 2)
|
||
)).scalar_one()
|
||
if old_cnt != new_cnt:
|
||
s = await db.get(Series, sid)
|
||
s.total_episodes = new_cnt
|
||
fixed += 1
|
||
await db.commit()
|
||
return ok(_SeriesRebuildResp(total=total, fixed=fixed))
|
||
|
||
|
||
class _SeriesBatchUpdate(BaseModel):
|
||
series_ids: list[int] = _Field(min_length=1, max_length=200)
|
||
status: int | None = _Field(None, ge=0, le=1)
|
||
category_id: int | None = _Field(None, ge=0)
|
||
|
||
|
||
@router.put("/series/batch")
|
||
async def batch_update_series(req: _SeriesBatchUpdate, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""批量编辑系列属性（当前支持 status / category_id）"""
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import Series
|
||
from sqlalchemy import select
|
||
|
||
if req.status is None and req.category_id is None:
|
||
raise BizException(errors.PARAM_ERROR, "至少要传一个要修改的字段")
|
||
|
||
stmt = select(Series).where(Series.id.in_(req.series_ids))
|
||
rows = (await db.execute(stmt)).scalars().all()
|
||
affected = 0
|
||
for s in rows:
|
||
changed = False
|
||
if req.status is not None and s.status != req.status:
|
||
s.status = req.status
|
||
changed = True
|
||
if req.category_id is not None and s.category_id != req.category_id:
|
||
s.category_id = req.category_id
|
||
changed = True
|
||
if changed:
|
||
affected += 1
|
||
|
||
db.add(AdminLog(
|
||
admin_id=admin.admin_id, action="series_batch_update",
|
||
target=f"series:{','.join(str(i) for i in req.series_ids)}",
|
||
detail=f"status={req.status} category_id={req.category_id} affected={affected}",
|
||
ip=_ip(request),
|
||
))
|
||
await db.commit()
|
||
return ok({"affected": affected})
|
||
|
||
|
||
@router.post("/series")
|
||
async def create_series(req: _SeriesCreate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Series
|
||
s = Series(
|
||
title=req.title, cover_url=req.cover_url, description=req.description,
|
||
category_id=req.category_id, status=1,
|
||
)
|
||
db.add(s)
|
||
await db.commit()
|
||
await db.refresh(s)
|
||
return ok({"series_id": s.id})
|
||
|
||
|
||
@router.put("/series/{series_id}")
|
||
async def update_series(series_id: int, req: _SeriesUpdate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Series
|
||
s = await db.get(Series, series_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "系列不存在")
|
||
d = req.model_dump(exclude_unset=True)
|
||
for k, v in d.items():
|
||
setattr(s, k, v)
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
@router.delete("/series/{series_id}")
|
||
async def delete_series(series_id: int, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Series
|
||
s = await db.get(Series, series_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "系列不存在")
|
||
await db.delete(s)
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
@router.post("/videos/{video_id}/series")
|
||
async def bind_video_series(video_id: int, req: _VideoSeriesBind,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
"""把视频归到某系列某集（episode_no 会自动维护 series.total_episodes）"""
|
||
from app.models.video import Series, Video
|
||
from sqlalchemy import select, func
|
||
|
||
v = await db.get(Video, video_id)
|
||
if not v:
|
||
raise BizException(errors.NOT_FOUND, "视频不存在")
|
||
s = await db.get(Series, req.series_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "系列不存在")
|
||
|
||
v.series_id = req.series_id
|
||
v.episode_no = req.episode_no
|
||
|
||
# 先 flush 让 count 查询能查到刚改的值
|
||
await db.flush()
|
||
# total_episodes 只统计已发布（status=2）的集数，和 public 接口过滤条件一致
|
||
cnt = (await db.execute(
|
||
select(func.count(Video.id)).where(
|
||
Video.series_id == req.series_id, Video.status == 2
|
||
)
|
||
)).scalar_one()
|
||
s.total_episodes = cnt
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
# ---------- 短剧采集 ----------
|
||
|
||
_collect_sync_running = False
|
||
|
||
|
||
@router.post("/collection/sync/{series_id}")
|
||
async def collect_sync_one(series_id: int, request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""手动同步单部采集短剧（补充源站新出的集数）"""
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import CollectionRecord
|
||
from app.services import collection_service
|
||
from sqlalchemy import select
|
||
|
||
source_id = (await db.execute(
|
||
select(CollectionRecord.source_id).where(
|
||
CollectionRecord.source_key == collection_service.SOURCE_KEY,
|
||
CollectionRecord.series_id == series_id,
|
||
)
|
||
)).scalar_one_or_none()
|
||
if not source_id:
|
||
raise BizException(errors.NOT_FOUND, "该系列不是采集内容")
|
||
|
||
stat = await collection_service.sync_drama_by_id(db, source_id)
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_sync",
|
||
target=f"series:{series_id}", detail=str(stat), ip=_ip(request)))
|
||
await db.commit()
|
||
return ok(stat)
|
||
|
||
|
||
class _CollectSyncAllReq(BaseModel):
|
||
pages: int = _Field(2, ge=1, le=20) # 增量同步的列表页数
|
||
|
||
|
||
@router.post("/collection/sync-all")
|
||
async def collect_sync_all(req: _CollectSyncAllReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""触发增量同步（后台异步执行，立即返回）"""
|
||
import asyncio
|
||
|
||
from app.database import SessionFactory
|
||
from app.models.admin import AdminLog
|
||
from app.services import collection_service
|
||
|
||
global _collect_sync_running
|
||
if _collect_sync_running:
|
||
raise BizException(errors.RATE_LIMITED, "已有同步任务进行中")
|
||
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_sync_all",
|
||
target="", detail=f"pages={req.pages}", ip=_ip(request)))
|
||
await db.commit()
|
||
|
||
async def _run() -> None:
|
||
global _collect_sync_running
|
||
_collect_sync_running = True
|
||
try:
|
||
async with SessionFactory() as s:
|
||
await collection_service.sync_short_dramas(s, max_pages=req.pages)
|
||
finally:
|
||
_collect_sync_running = False
|
||
|
||
asyncio.create_task(_run())
|
||
return ok({"started": True, "pages": req.pages})
|
||
|
||
|
||
# ---------- 采集源管理（单视频类，自助添加）----------
|
||
|
||
|
||
class _SourceCreate(BaseModel):
|
||
name: str = _Field(min_length=1, max_length=50)
|
||
list_url: str = _Field(min_length=10, max_length=500)
|
||
|
||
|
||
class _SourceUpdate(BaseModel):
|
||
name: str | None = _Field(None, min_length=1, max_length=50)
|
||
list_url: str | None = _Field(None, min_length=10, max_length=500)
|
||
status: int | None = _Field(None, ge=0, le=1)
|
||
|
||
|
||
def _source_dict(s) -> dict:
|
||
return {
|
||
"id": s.id, "name": s.name, "list_url": s.list_url,
|
||
"category_id": s.category_id, "status": s.status,
|
||
"last_sync_at": s.last_sync_at.strftime("%Y-%m-%d %H:%M:%S") if s.last_sync_at else "",
|
||
"created_at": s.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
|
||
|
||
async def _resolve_category(db: AsyncSession, name: str) -> int:
|
||
"""按名称找分类，不存在则创建"""
|
||
from app.models.video import Category
|
||
from sqlalchemy import func, select
|
||
|
||
cid = (await db.execute(select(Category.id).where(Category.name == name))).scalar_one_or_none()
|
||
if cid:
|
||
return cid
|
||
max_sort = (await db.execute(select(func.coalesce(func.max(Category.sort), 0)))).scalar_one()
|
||
c = Category(name=name, sort=max_sort + 1, status=1)
|
||
db.add(c)
|
||
await db.flush()
|
||
return c.id
|
||
|
||
|
||
@router.get("/collection/sources")
|
||
async def list_sources(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
from app.models.video import CollectionSource
|
||
from sqlalchemy import select
|
||
|
||
rows = (await db.execute(select(CollectionSource).order_by(CollectionSource.id.desc()))).scalars().all()
|
||
return ok({"list": [_source_dict(s) for s in rows]})
|
||
|
||
|
||
@router.post("/collection/sources")
|
||
async def create_source(req: _SourceCreate, request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""添加采集源：自动按名称建/找分类"""
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import CollectionSource
|
||
|
||
if "{{page}}" not in req.list_url:
|
||
raise BizException(errors.PARAM_ERROR, "URL 需包含 {{page}} 页码占位符")
|
||
|
||
cid = await _resolve_category(db, req.name.strip())
|
||
s = CollectionSource(name=req.name.strip(), list_url=req.list_url.strip(), category_id=cid)
|
||
db.add(s)
|
||
await db.flush()
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_source_create",
|
||
target=f"source:{s.id}", detail=f"{s.name} {s.list_url}", ip=_ip(request)))
|
||
await db.commit()
|
||
await db.refresh(s)
|
||
return ok(_source_dict(s))
|
||
|
||
|
||
@router.put("/collection/sources/{source_id}")
|
||
async def update_source(source_id: int, req: _SourceUpdate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
from app.models.video import CollectionSource
|
||
|
||
s = await db.get(CollectionSource, source_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "采集源不存在")
|
||
d = req.model_dump(exclude_unset=True)
|
||
if "list_url" in d and "{{page}}" not in d["list_url"]:
|
||
raise BizException(errors.PARAM_ERROR, "URL 需包含 {{page}} 页码占位符")
|
||
if "name" in d:
|
||
d["category_id"] = await _resolve_category(db, d["name"].strip())
|
||
d["name"] = d["name"].strip()
|
||
for k, v in d.items():
|
||
setattr(s, k, v)
|
||
await db.commit()
|
||
await db.refresh(s)
|
||
return ok(_source_dict(s))
|
||
|
||
|
||
@router.delete("/collection/sources/{source_id}")
|
||
async def delete_source(source_id: int, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import CollectionSource
|
||
|
||
s = await db.get(CollectionSource, source_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "采集源不存在")
|
||
await db.delete(s)
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
class _SourceBatchReq(BaseModel):
|
||
text: str = _Field(min_length=1, max_length=20000)
|
||
|
||
|
||
@router.post("/collection/sources/batch")
|
||
async def batch_create_sources(req: _SourceBatchReq, request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""批量添加采集源：每行一条，格式 "分类名 URL"（URL 可带反引号），成功后自动触发采集"""
|
||
import re
|
||
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import CollectionSource
|
||
from sqlalchemy import select
|
||
|
||
line_re = re.compile(r"^\s*(?P<name>.+?)[\s]+[`'\"]*(?P<url>https?://[^\s`'\"]+)[`'\"]*\s*$")
|
||
created, skipped, failed = [], [], []
|
||
|
||
for raw in req.text.splitlines():
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
m = line_re.match(line)
|
||
if not m or "{{page}}" not in m.group("url"):
|
||
failed.append({"line": line[:80], "reason": "格式错误或缺少 {{page}} 占位"})
|
||
continue
|
||
name, url = m.group("name").strip().strip("`'\" "), m.group("url")
|
||
if not name or len(name) > 50:
|
||
failed.append({"line": line[:80], "reason": "分类名为空或过长"})
|
||
continue
|
||
exists = (await db.execute(
|
||
select(CollectionSource.id).where(CollectionSource.list_url == url)
|
||
)).scalar_one_or_none()
|
||
if exists:
|
||
skipped.append({"name": name, "reason": "URL 已存在"})
|
||
continue
|
||
cid = await _resolve_category(db, name)
|
||
s = CollectionSource(name=name, list_url=url, category_id=cid)
|
||
db.add(s)
|
||
await db.flush()
|
||
created.append({"id": s.id, "name": name})
|
||
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_source_batch",
|
||
target="", detail=f"created={len(created)} skipped={len(skipped)} failed={len(failed)}",
|
||
ip=_ip(request)))
|
||
await db.commit()
|
||
|
||
# 新建的源自动触发后台采集
|
||
if created:
|
||
import asyncio
|
||
for c in created:
|
||
if c["id"] not in _sync_running_sources:
|
||
_sync_running_sources.add(c["id"])
|
||
asyncio.create_task(_run_source_sync(c["id"], 0))
|
||
|
||
return ok({"created": created, "skipped": skipped, "failed": failed,
|
||
"sync_started": [c["id"] for c in created]})
|
||
|
||
|
||
_sync_running_sources: set[int] = set()
|
||
|
||
|
||
async def _run_source_sync(source_id: int, max_pages: int) -> None:
|
||
from app.database import SessionFactory
|
||
from app.models.video import CollectionSource
|
||
from app.services import collection_service
|
||
|
||
try:
|
||
async with SessionFactory() as s:
|
||
source = await s.get(CollectionSource, source_id)
|
||
if source:
|
||
stat = await collection_service.sync_source(s, source, max_pages=max_pages)
|
||
logger.info("采集源 %s 同步完成: %s", source_id, stat)
|
||
except Exception:
|
||
logger.exception("采集源 %s 同步失败", source_id)
|
||
finally:
|
||
_sync_running_sources.discard(source_id)
|
||
|
||
|
||
class _SourceSyncReq(BaseModel):
|
||
max_pages: int = _Field(0, ge=0, le=200) # 0=不限(增量会自动提前停止)
|
||
|
||
|
||
@router.post("/collection/sources/{source_id}/sync")
|
||
async def sync_one_source(source_id: int, req: _SourceSyncReq, request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""同步单个采集源（后台异步，立即返回）"""
|
||
import asyncio
|
||
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import CollectionSource
|
||
|
||
s = await db.get(CollectionSource, source_id)
|
||
if not s:
|
||
raise BizException(errors.NOT_FOUND, "采集源不存在")
|
||
if source_id in _sync_running_sources:
|
||
raise BizException(errors.RATE_LIMITED, "该源正在同步中")
|
||
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_source_sync",
|
||
target=f"source:{source_id}", detail=f"max_pages={req.max_pages}", ip=_ip(request)))
|
||
await db.commit()
|
||
|
||
_sync_running_sources.add(source_id)
|
||
asyncio.create_task(_run_source_sync(source_id, req.max_pages))
|
||
return ok({"started": True, "source_id": source_id})
|
||
|
||
|
||
@router.post("/collection/sources/sync-all")
|
||
async def sync_all_sources(request: Request, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""同步全部启用的采集源（后台异步）"""
|
||
import asyncio
|
||
|
||
from app.models.admin import AdminLog
|
||
from app.models.video import CollectionSource
|
||
from sqlalchemy import select
|
||
|
||
ids = (await db.execute(
|
||
select(CollectionSource.id).where(CollectionSource.status == 1)
|
||
)).scalars().all()
|
||
started = [i for i in ids if i not in _sync_running_sources]
|
||
for i in started:
|
||
_sync_running_sources.add(i)
|
||
asyncio.create_task(_run_source_sync(i, 0))
|
||
|
||
db.add(AdminLog(admin_id=admin.admin_id, action="collection_sources_sync_all",
|
||
target="", detail=f"sources={started}", ip=_ip(request)))
|
||
await db.commit()
|
||
return ok({"started": started, "skipped_running": [i for i in ids if i not in started]})
|
||
|
||
|
||
# ---------- 分类管理 ----------
|
||
|
||
from pydantic import BaseModel as _BM
|
||
|
||
|
||
class _CategoryCreate(_BM):
|
||
name: str
|
||
sort: int = 0
|
||
status: int = 1
|
||
|
||
|
||
class _CategoryUpdate(_BM):
|
||
name: str | None = None
|
||
sort: int | None = None
|
||
status: int | None = None
|
||
|
||
|
||
@router.get("/categories")
|
||
async def list_categories_public(db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
"""给 admin 页面下拉用的全部启用分类（不含分页）"""
|
||
from app.models.video import Category
|
||
from sqlalchemy import select
|
||
|
||
rows = (await db.execute(
|
||
select(Category).where(Category.status == 1).order_by(Category.sort.asc(), Category.id.asc())
|
||
)).scalars().all()
|
||
return ok([{"id": c.id, "name": c.name} for c in rows])
|
||
|
||
|
||
@router.get("/categories/admin")
|
||
async def admin_list_categories(keyword: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
from app.models.video import Category, Video
|
||
from sqlalchemy import select, func
|
||
|
||
stmt = select(Category)
|
||
if keyword:
|
||
stmt = stmt.where(Category.name.contains(keyword))
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one()
|
||
rows = (await db.execute(
|
||
stmt.order_by(Category.sort.asc(), Category.id.asc())
|
||
.offset((page - 1) * page_size).limit(page_size)
|
||
)).scalars().all()
|
||
|
||
video_counts: dict[int, int] = {}
|
||
if rows:
|
||
cnt_rows = (await db.execute(
|
||
select(Video.category_id, func.count(Video.id)).where(
|
||
Video.category_id.in_([r.id for r in rows])
|
||
).group_by(Video.category_id)
|
||
)).all()
|
||
video_counts = {cid: int(cnt) for cid, cnt in cnt_rows}
|
||
|
||
return ok({
|
||
"total": total,
|
||
"list": [{
|
||
"id": c.id, "name": c.name, "sort": c.sort, "status": c.status,
|
||
"video_count": video_counts.get(c.id, 0),
|
||
} for c in rows],
|
||
})
|
||
|
||
|
||
@router.post("/categories")
|
||
async def create_category(req: _CategoryCreate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Category
|
||
from sqlalchemy import select
|
||
|
||
exist = (await db.execute(select(Category).where(Category.name == req.name))).scalar_one_or_none()
|
||
if exist:
|
||
raise BizException(errors.DUPLICATE, "分类名已存在")
|
||
c = Category(name=req.name, sort=req.sort, status=req.status)
|
||
db.add(c)
|
||
await db.commit()
|
||
await db.refresh(c)
|
||
return ok({"id": c.id})
|
||
|
||
|
||
@router.put("/categories/{cid}")
|
||
async def update_category(cid: int, req: _CategoryUpdate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Category
|
||
|
||
c = await db.get(Category, cid)
|
||
if not c:
|
||
raise BizException(errors.NOT_FOUND, "分类不存在")
|
||
if req.name is not None:
|
||
c.name = req.name
|
||
if req.sort is not None:
|
||
c.sort = req.sort
|
||
if req.status is not None:
|
||
c.status = req.status
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
@router.delete("/categories/{cid}")
|
||
async def delete_category(cid: int, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Category, Video
|
||
from sqlalchemy import select, func
|
||
|
||
cnt = (await db.execute(
|
||
select(func.count(Video.id)).where(Video.category_id == cid)
|
||
)).scalar_one()
|
||
if cnt > 0:
|
||
raise BizException(errors.CONFLICT, f"该分类下还有 {cnt} 条视频，先处理它们或停用分类")
|
||
c = await db.get(Category, cid)
|
||
if c:
|
||
await db.delete(c)
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
# ---------- Tag 管理 ----------
|
||
|
||
class _TagCreate(_BM):
|
||
name: str
|
||
|
||
|
||
class _TagUpdate(_BM):
|
||
name: str | None = None
|
||
|
||
|
||
@router.get("/tags")
|
||
async def admin_list_tags(keyword: str | None = Query(None),
|
||
page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200),
|
||
db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_operator)):
|
||
from app.models.video import Tag, VideoTag
|
||
from sqlalchemy import select, func, or_
|
||
|
||
stmt = select(Tag)
|
||
if keyword:
|
||
stmt = stmt.where(Tag.name.contains(keyword))
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one()
|
||
rows = (await db.execute(
|
||
stmt.order_by(Tag.id.desc()).offset((page - 1) * page_size).limit(page_size)
|
||
)).scalars().all()
|
||
|
||
# 实时计算 video_count（比冗余字段准）
|
||
counts: dict[int, int] = {}
|
||
if rows:
|
||
cnt_rows = (await db.execute(
|
||
select(VideoTag.tag_id, func.count(VideoTag.id)).where(
|
||
VideoTag.tag_id.in_([r.id for r in rows])
|
||
).group_by(VideoTag.tag_id)
|
||
)).all()
|
||
counts = {tid: int(c) for tid, c in cnt_rows}
|
||
|
||
return ok({
|
||
"total": total,
|
||
"list": [{
|
||
"id": t.id, "name": t.name,
|
||
"video_count": counts.get(t.id, 0),
|
||
"created_at": t.created_at.strftime("%Y-%m-%d %H:%M:%S") if t.created_at else "",
|
||
} for t in rows],
|
||
})
|
||
|
||
|
||
@router.post("/tags")
|
||
async def create_tag(req: _TagCreate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Tag
|
||
from sqlalchemy import select
|
||
|
||
name = req.name.strip()
|
||
if not name:
|
||
raise BizException(errors.INVALID_PARAMS, "标签名不能为空")
|
||
exist = (await db.execute(select(Tag).where(Tag.name == name))).scalar_one_or_none()
|
||
if exist:
|
||
raise BizException(errors.DUPLICATE, "标签已存在")
|
||
t = Tag(name=name)
|
||
db.add(t)
|
||
await db.commit()
|
||
await db.refresh(t)
|
||
return ok({"id": t.id})
|
||
|
||
|
||
@router.put("/tags/{tid}")
|
||
async def update_tag(tid: int, req: _TagUpdate, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Tag
|
||
|
||
t = await db.get(Tag, tid)
|
||
if not t:
|
||
raise BizException(errors.NOT_FOUND, "标签不存在")
|
||
if req.name is not None:
|
||
t.name = req.name.strip()
|
||
if not t.name:
|
||
raise BizException(errors.INVALID_PARAMS, "标签名不能为空")
|
||
await db.commit()
|
||
return ok()
|
||
|
||
|
||
@router.delete("/tags/{tid}")
|
||
async def delete_tag(tid: int, db: AsyncSession = Depends(get_db),
|
||
admin: AdminIdentity = Depends(require_super)):
|
||
from app.models.video import Tag, VideoTag
|
||
from sqlalchemy import delete
|
||
|
||
await db.execute(delete(VideoTag).where(VideoTag.tag_id == tid))
|
||
t = await db.get(Tag, tid)
|
||
if t:
|
||
await db.delete(t)
|
||
await db.commit()
|
||
return ok()
|