768 lines
31 KiB
Python
768 lines
31 KiB
Python
"""视频服务：Feed / 详情 / 观看计费 / 完整观看（对应《03-接口文档》第 5 章、《02-技术文档》6.4）"""
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import random
|
||
import uuid
|
||
from datetime import date, datetime, timedelta
|
||
|
||
import redis.asyncio as aioredis
|
||
from fastapi import UploadFile
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.dialects.mysql import insert as mysql_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import biz_config
|
||
from app.core import errors, ratelimit
|
||
from app.core.response import BizException
|
||
from app.deps import Identity
|
||
from app.models.billing import WatchRecord
|
||
from app.models.incentive import VideoEarning
|
||
from app.models.interaction import Favorite, Like
|
||
from app.models.user import User
|
||
from app.models.video import Category, Tag, Video, VideoTag
|
||
from app.schemas.video import PublishVideoReq, UploadTokenReq
|
||
from app.services import membership_service, sensitive_service, storage_service, wallet_service
|
||
|
||
STATUS_PUBLISHED = 2
|
||
|
||
STATUS_TEXT = {0: "转码中", 1: "待审核", 2: "已发布", 3: "拒绝", 4: "下架", 5: "转码失败"}
|
||
|
||
# 观看上报限流：60 次/分钟/计费主体
|
||
WATCH_RATE_LIMIT = 60
|
||
# 免费额度 Set / 已看 Set 过期时间 48h
|
||
SET_TTL = 48 * 3600
|
||
# 推荐候选池上限
|
||
FEED_POOL_SIZE = 500
|
||
# 视频详情缓存 5min
|
||
DETAIL_CACHE_TTL = 300
|
||
|
||
# 免费额度 + 去重 Lua 原子脚本（《02-技术文档》3.4-2）
|
||
FREE_QUOTA_LUA = """
|
||
if redis.call('SISMEMBER', KEYS[1], ARGV[1]) == 1 then return 1 end
|
||
local n = redis.call('SCARD', KEYS[1])
|
||
if n < tonumber(ARGV[2]) then redis.call('SADD', KEYS[1], ARGV[1]) redis.call('EXPIRE', KEYS[1], ARGV[3]) return 2 end
|
||
return 0
|
||
"""
|
||
|
||
|
||
def _fmt(dt: datetime | None) -> str | None:
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None
|
||
|
||
|
||
def _points_cfg() -> dict:
|
||
return biz_config.get("points") or {}
|
||
|
||
|
||
# ---------- Feed 单条序列化 ----------
|
||
|
||
def _author_brief(user: User | None) -> dict:
|
||
return {
|
||
"user_id": user.id if user else 0,
|
||
"nickname": user.nickname if user else "",
|
||
"avatar_url": user.avatar_url if user else "",
|
||
}
|
||
|
||
|
||
async def serialize_videos(
|
||
db: AsyncSession, videos: list[Video], viewer_user_id: int | None = None
|
||
) -> list[dict]:
|
||
"""Feed 单条结构（对齐《03-接口文档》5.1）：批量取作者/标签/登录用户互动状态"""
|
||
if not videos:
|
||
return []
|
||
video_ids = [v.id for v in videos]
|
||
|
||
authors = {
|
||
u.id: u
|
||
for u in (await db.execute(select(User).where(User.id.in_({v.user_id for v in videos})))).scalars()
|
||
}
|
||
|
||
tags_map = await get_tags_map(db, video_ids)
|
||
|
||
liked: set[int] = set()
|
||
favorited: set[int] = set()
|
||
if viewer_user_id:
|
||
liked = set(
|
||
(await db.execute(
|
||
select(Like.video_id).where(Like.user_id == viewer_user_id, Like.video_id.in_(video_ids))
|
||
)).scalars()
|
||
)
|
||
favorited = set(
|
||
(await db.execute(
|
||
select(Favorite.video_id).where(Favorite.user_id == viewer_user_id, Favorite.video_id.in_(video_ids))
|
||
)).scalars()
|
||
)
|
||
|
||
return [
|
||
{
|
||
"video_id": v.id,
|
||
"title": v.title,
|
||
"cover_url": v.cover_url,
|
||
"video_url": v.video_url,
|
||
"duration": v.duration,
|
||
"category_id": v.category_id,
|
||
"author": _author_brief(authors.get(v.user_id)),
|
||
"view_count": v.view_count,
|
||
"like_count": v.like_count,
|
||
"comment_count": v.comment_count,
|
||
"favorite_count": v.favorite_count,
|
||
"tags": tags_map.get(v.id, []),
|
||
"is_liked": v.id in liked,
|
||
"is_favorited": v.id in favorited,
|
||
}
|
||
for v in videos
|
||
]
|
||
|
||
|
||
async def get_tags_map(db: AsyncSession, video_ids: list[int]) -> dict[int, list[str]]:
|
||
if not video_ids:
|
||
return {}
|
||
rows = await db.execute(
|
||
select(VideoTag.video_id, Tag.name)
|
||
.join(Tag, Tag.id == VideoTag.tag_id)
|
||
.where(VideoTag.video_id.in_(video_ids))
|
||
)
|
||
tags_map: dict[int, list[str]] = {}
|
||
for vid, name in rows.all():
|
||
tags_map.setdefault(vid, []).append(name)
|
||
return tags_map
|
||
|
||
|
||
async def get_tags_of_video(db: AsyncSession, video_id: int) -> list[str]:
|
||
return (await get_tags_map(db, [video_id])).get(video_id, [])
|
||
|
||
|
||
# ---------- 游标 ----------
|
||
|
||
def _encode_cursor(payload: dict) -> str:
|
||
return base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode()
|
||
|
||
|
||
def _decode_cursor(cursor: str | None) -> dict:
|
||
if not cursor:
|
||
return {}
|
||
try:
|
||
return json.loads(base64.urlsafe_b64decode(cursor.encode()))
|
||
except Exception:
|
||
raise BizException(errors.PARAM_ERROR, "cursor 无效")
|
||
|
||
|
||
# ---------- Feed ----------
|
||
|
||
async def get_feed(
|
||
db: AsyncSession,
|
||
redis: aioredis.Redis,
|
||
identity: Identity,
|
||
feed_type: str,
|
||
cursor: str | None,
|
||
limit: int,
|
||
category_id: int | None,
|
||
) -> dict:
|
||
if feed_type == "latest":
|
||
return await _feed_latest(db, identity, cursor, limit, category_id)
|
||
if feed_type in ("liked", "viewed"):
|
||
return await _feed_ranked(db, identity, feed_type, cursor, limit, category_id)
|
||
return await _feed_scored(db, redis, identity, feed_type, cursor, limit, category_id)
|
||
|
||
|
||
async def _feed_ranked(
|
||
db: AsyncSession, identity: Identity, feed_type: str, cursor: str | None,
|
||
limit: int, category_id: int | None,
|
||
) -> dict:
|
||
"""liked / viewed：按点赞量 / 播放量严格倒序（值 + id 打破并列游标）"""
|
||
col = Video.like_count if feed_type == "liked" else Video.view_count
|
||
stmt = await _base_pool_stmt(category_id)
|
||
c = _decode_cursor(cursor)
|
||
if c:
|
||
val, vid = int(c["n"]), int(c["i"])
|
||
stmt = stmt.where((col < val) | ((col == val) & (Video.id < vid)))
|
||
rows = list(
|
||
(await db.execute(stmt.order_by(col.desc(), Video.id.desc()).limit(limit + 1))).scalars()
|
||
)
|
||
page, rows = (len(rows) > limit), rows[:limit]
|
||
next_cursor = None
|
||
if page and rows:
|
||
last = rows[-1]
|
||
last_val = last.like_count if feed_type == "liked" else last.view_count
|
||
next_cursor = _encode_cursor({"n": last_val, "i": last.id})
|
||
return {
|
||
"next_cursor": next_cursor,
|
||
"list": await serialize_videos(db, rows, identity.user_id),
|
||
}
|
||
|
||
|
||
async def _base_pool_stmt(category_id: int | None):
|
||
stmt = select(Video).where(Video.status == STATUS_PUBLISHED, Video.published_at.isnot(None))
|
||
if category_id:
|
||
stmt = stmt.where(Video.category_id == category_id)
|
||
return stmt
|
||
|
||
|
||
async def _feed_latest(
|
||
db: AsyncSession, identity: Identity, cursor: str | None, limit: int, category_id: int | None
|
||
) -> dict:
|
||
"""latest：published_at 游标倒序（ts + id 打破并列）"""
|
||
stmt = await _base_pool_stmt(category_id)
|
||
c = _decode_cursor(cursor)
|
||
if c:
|
||
ts, vid = datetime.fromtimestamp(c["p"]), int(c["i"])
|
||
stmt = stmt.where((Video.published_at < ts) | ((Video.published_at == ts) & (Video.id < vid)))
|
||
rows = list(
|
||
(await db.execute(stmt.order_by(Video.published_at.desc(), Video.id.desc()).limit(limit + 1))).scalars()
|
||
)
|
||
page, rows = (len(rows) > limit), rows[:limit]
|
||
next_cursor = None
|
||
if page and rows:
|
||
last = rows[-1]
|
||
next_cursor = _encode_cursor({"p": int(last.published_at.timestamp()), "i": last.id})
|
||
return {
|
||
"next_cursor": next_cursor,
|
||
"list": await serialize_videos(db, rows, identity.user_id),
|
||
}
|
||
|
||
|
||
def _hot_score(v: Video, now: datetime) -> float:
|
||
"""热度分：view×1 + like×5 + comment×10，随发布天数衰减"""
|
||
age_days = max((now - (v.published_at or now)).days, 0)
|
||
base = v.view_count + v.like_count * 5 + v.comment_count * 10
|
||
return base / (1 + age_days)
|
||
|
||
|
||
async def _user_tag_prefs(db: AsyncSession, user_id: int) -> dict[str, float]:
|
||
"""近 7 天观看/点赞/收藏视频的标签偏好（观看+1 点赞+3 收藏+5）；无偏好返回空"""
|
||
since = datetime.now() - timedelta(days=7)
|
||
video_ids: list[tuple[int, float]] = []
|
||
watched = (await db.execute(
|
||
select(WatchRecord.video_id).where(
|
||
WatchRecord.subject_type == 1, WatchRecord.subject_id == str(user_id), WatchRecord.created_at > since
|
||
)
|
||
)).scalars().all()
|
||
video_ids += [(vid, 1.0) for vid in watched]
|
||
liked = (await db.execute(select(Like.video_id).where(Like.user_id == user_id))).scalars().all()
|
||
video_ids += [(vid, 3.0) for vid in liked]
|
||
fav = (await db.execute(select(Favorite.video_id).where(Favorite.user_id == user_id))).scalars().all()
|
||
video_ids += [(vid, 5.0) for vid in fav]
|
||
if not video_ids:
|
||
return {}
|
||
tags_map = await get_tags_map(db, list({vid for vid, _ in video_ids}))
|
||
prefs: dict[str, float] = {}
|
||
for vid, w in video_ids:
|
||
for tag in tags_map.get(vid, []):
|
||
prefs[tag] = prefs.get(tag, 0.0) + w
|
||
return prefs
|
||
|
||
|
||
async def _feed_scored(
|
||
db: AsyncSession,
|
||
redis: aioredis.Redis,
|
||
identity: Identity,
|
||
feed_type: str,
|
||
cursor: str | None,
|
||
limit: int,
|
||
category_id: int | None,
|
||
) -> dict:
|
||
"""recommend / hot：候选池打分 → 当日种子确定性加权抽样 → 当日去重（不足兜底）→ 偏移游标"""
|
||
now = datetime.now()
|
||
today = f"{now:%Y%m%d}"
|
||
subject = identity.subject or "anon"
|
||
|
||
# 1. 候选池：近 30 天已发布，封顶 FEED_POOL_SIZE
|
||
stmt = (await _base_pool_stmt(category_id)).where(
|
||
Video.published_at > now - timedelta(days=30)
|
||
).order_by(Video.published_at.desc()).limit(FEED_POOL_SIZE)
|
||
pool = list((await db.execute(stmt)).scalars())
|
||
if not pool:
|
||
return {"next_cursor": None, "list": []}
|
||
|
||
# 2. 打分
|
||
scores: dict[int, float] = {}
|
||
if feed_type == "recommend" and identity.user_id:
|
||
prefs = await _user_tag_prefs(db, identity.user_id)
|
||
else:
|
||
prefs = {}
|
||
tags_map = await get_tags_map(db, [v.id for v in pool])
|
||
for v in pool:
|
||
hot = _hot_score(v, now)
|
||
if feed_type == "recommend" and prefs:
|
||
# 命中偏好标签加权，未命中取热度分
|
||
pref = sum(prefs.get(t, 0.0) for t in tags_map.get(v.id, []))
|
||
scores[v.id] = pref * 1000 + hot if pref > 0 else hot
|
||
else:
|
||
# hot 策略 / 游客 / 新用户：热度分
|
||
scores[v.id] = hot
|
||
|
||
# 3. 排序：hot = 热度分严格倒序；recommend = 当日种子确定性加权抽样（每日重排，同日稳定）
|
||
if feed_type == "hot":
|
||
ordered = sorted(pool, key=lambda v: (scores[v.id], v.id), reverse=True)
|
||
else:
|
||
seed = int(hashlib.md5(f"{today}{subject}".encode()).hexdigest()[:8], 16)
|
||
rng = random.Random(seed)
|
||
ordered = sorted(pool, key=lambda v: rng.random() ** (1.0 / max(scores[v.id], 1e-6)), reverse=True)
|
||
|
||
# 4. 当日去重（观看上报写入 watched Set）；候选不足时放开过滤兜底
|
||
watched = await redis.smembers(f"watched:{subject}:{today}") # type: ignore[misc]
|
||
filtered = [v for v in ordered if str(v.id) not in watched]
|
||
|
||
offset = int(_decode_cursor(cursor).get("o", 0))
|
||
items = filtered if len(filtered) > offset else ordered
|
||
page_items = items[offset: offset + limit]
|
||
next_cursor = _encode_cursor({"o": offset + limit}) if len(items) > offset + limit else None
|
||
return {
|
||
"next_cursor": next_cursor,
|
||
"list": await serialize_videos(db, page_items, identity.user_id),
|
||
}
|
||
|
||
|
||
# ---------- 视频详情 ----------
|
||
|
||
async def get_detail(db: AsyncSession, redis: aioredis.Redis, identity: Identity, video_id: int) -> dict:
|
||
"""详情：基础数据缓存 5min（中性，不含观看者态）；登录用户叠加 my_position/互动状态"""
|
||
cache_key = f"video:detail:{video_id}"
|
||
cached = await redis.get(cache_key)
|
||
if cached:
|
||
data = json.loads(cached)
|
||
else:
|
||
video = await db.get(Video, video_id)
|
||
if video is None or video.status != STATUS_PUBLISHED:
|
||
raise BizException(errors.VIDEO_NOT_AVAILABLE)
|
||
base = (await serialize_videos(db, [video]))[0]
|
||
category = await db.get(Category, video.category_id) if video.category_id else None
|
||
base.update({
|
||
"description": video.description,
|
||
"category": {"category_id": category.id, "name": category.name} if category else None,
|
||
"published_at": _fmt(video.published_at),
|
||
})
|
||
data = base
|
||
await redis.set(cache_key, json.dumps(data, ensure_ascii=False), ex=DETAIL_CACHE_TTL)
|
||
|
||
data = dict(data)
|
||
data["my_position"] = 0
|
||
if identity.user_id:
|
||
# 登录用户：互动状态 + 续播进度（最近一次观看记录）
|
||
video_id_int = int(data["video_id"])
|
||
data["is_liked"] = bool((await db.execute(
|
||
select(Like.id).where(Like.user_id == identity.user_id, Like.video_id == video_id_int)
|
||
)).first())
|
||
data["is_favorited"] = bool((await db.execute(
|
||
select(Favorite.id).where(Favorite.user_id == identity.user_id, Favorite.video_id == video_id_int)
|
||
)).first())
|
||
last = (await db.execute(
|
||
select(WatchRecord.position).where(
|
||
WatchRecord.subject_type == 1, WatchRecord.subject_id == str(identity.user_id),
|
||
WatchRecord.video_id == video_id_int,
|
||
).order_by(WatchRecord.id.desc()).limit(1)
|
||
)).scalar()
|
||
data["my_position"] = int(last or 0)
|
||
return data
|
||
|
||
|
||
async def invalidate_detail_cache(redis: aioredis.Redis, video_id: int) -> None:
|
||
"""状态变更（编辑/上下架/审核）时失效详情缓存"""
|
||
await redis.delete(f"video:detail:{video_id}")
|
||
|
||
|
||
# ---------- 观看上报（计费） ----------
|
||
|
||
async def _get_published_video(db: AsyncSession, video_id: int) -> Video:
|
||
video = await db.get(Video, video_id)
|
||
if video is None or video.status != STATUS_PUBLISHED:
|
||
raise BizException(errors.VIDEO_NOT_AVAILABLE)
|
||
return video
|
||
|
||
|
||
async def report_watch(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity, video_id: int, position: int
|
||
) -> dict:
|
||
"""计费判定链：同日去重 → 会员 → 本人 → 免费额度 Lua → 游客拦截 → 扣积分"""
|
||
video = await _get_published_video(db, video_id)
|
||
if not identity.subject:
|
||
raise BizException(errors.PARAM_ERROR, "缺少 X-Device-ID")
|
||
today = date.today()
|
||
ymd = f"{today:%Y%m%d}"
|
||
subject = identity.subject
|
||
watched_key = f"watched:{subject}:{ymd}"
|
||
|
||
await ratelimit.hit(redis, f"rl:watch:{subject}:{datetime.now():%Y%m%d%H%M}", WATCH_RATE_LIMIT, 60)
|
||
|
||
# 1. 同日去重：重复观看不计费，仅更新续播进度
|
||
if await redis.sismember(watched_key, str(video_id)): # type: ignore[misc]
|
||
await _upsert_watch_record(db, identity, video_id, today, 3, 0, position)
|
||
await db.commit()
|
||
return await _watch_resp(db, redis, identity, 3, 0, ymd)
|
||
|
||
is_author = bool(identity.user_id and video.user_id == identity.user_id)
|
||
cost_type, cost_points = 3, 0
|
||
|
||
if is_author:
|
||
pass # 本人作品不计费
|
||
elif identity.user_id and await membership_service.is_member(db, identity.user_id):
|
||
cost_type = 2 # 会员免费
|
||
else:
|
||
# 2. 免费额度（Lua 原子：去重 + 计数 + 占位）
|
||
free_limit = int(_points_cfg().get("daily_free_count", 10))
|
||
free_key = f"free:watch:{subject}:{ymd}"
|
||
res = await redis.eval(FREE_QUOTA_LUA, 1, free_key, str(video_id), str(free_limit), str(SET_TTL)) # type: ignore[misc]
|
||
if res == 2:
|
||
cost_type = 0 # 占用免费额度
|
||
elif res == 1:
|
||
cost_type = 3 # 并发下已在免费集合 → 按重复处理
|
||
else:
|
||
# 3. 额度用尽：游客拦截 / 用户扣积分
|
||
if identity.is_guest:
|
||
raise BizException(errors.FREE_LIMIT_EXCEEDED, "今日免费观看次数已用完，登录后继续观看",
|
||
data={"need_login": True})
|
||
cost_type = 1
|
||
cost_points = int(_points_cfg().get("watch_cost", 1))
|
||
try:
|
||
await wallet_service.change_points(
|
||
db, identity.user_id, -cost_points, wallet_service.BIZ_WATCH,
|
||
biz_id=str(video_id), remark="观看视频",
|
||
)
|
||
except BizException as e:
|
||
if e.code == errors.POINTS_NOT_ENOUGH:
|
||
# 文档 5.3：积分不足响应 data 带当前积分
|
||
e.data = {"points": await wallet_service.get_balance(db, identity.user_id)}
|
||
raise
|
||
|
||
first_today = await _upsert_watch_record(db, identity, video_id, today, cost_type, cost_points, position)
|
||
if first_today:
|
||
video.view_count += 1
|
||
await redis.sadd(watched_key, str(video_id)) # type: ignore[misc]
|
||
await redis.expire(watched_key, SET_TTL)
|
||
await db.commit()
|
||
return await _watch_resp(db, redis, identity, cost_type, cost_points, ymd)
|
||
|
||
|
||
async def _upsert_watch_record(
|
||
db: AsyncSession, identity: Identity, video_id: int, watch_date: date,
|
||
cost_type: int, cost_points: int, position: int, watch_duration: int = 0,
|
||
) -> bool:
|
||
"""按 uk_dedup 天然键幂等写入；返回是否当日首次（用于播放量与收益计数）"""
|
||
insert_stmt = mysql_insert(WatchRecord).values(
|
||
subject_type=1 if identity.user_id else 2,
|
||
subject_id=identity.subject,
|
||
video_id=video_id,
|
||
watch_date=watch_date,
|
||
cost_type=cost_type,
|
||
cost_points=cost_points,
|
||
position=position,
|
||
watch_duration=watch_duration,
|
||
)
|
||
stmt = insert_stmt.on_duplicate_key_update(
|
||
position=insert_stmt.inserted.position,
|
||
watch_duration=insert_stmt.inserted.watch_duration,
|
||
)
|
||
result = await db.execute(stmt)
|
||
await db.flush()
|
||
return result.rowcount == 1
|
||
|
||
|
||
async def _watch_resp(db: AsyncSession, redis: aioredis.Redis, identity: Identity,
|
||
cost_type: int, cost_points: int, ymd: str) -> dict:
|
||
free_limit = int(_points_cfg().get("daily_free_count", 10))
|
||
used = await redis.scard(f"free:watch:{identity.subject}:{ymd}") # type: ignore[misc]
|
||
remaining_points = await wallet_service.get_balance(db, identity.user_id) if identity.user_id else 0
|
||
return {
|
||
"cost_type": cost_type,
|
||
"cost_points": cost_points,
|
||
"remaining_points": remaining_points,
|
||
"free_remaining_today": max(free_limit - int(used or 0), 0),
|
||
}
|
||
|
||
|
||
# ---------- 完整观看上报 ----------
|
||
|
||
async def report_complete(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity, video_id: int, watch_duration: int
|
||
) -> dict:
|
||
"""完整观看：本人/当日重复/时长明显不足 → credited=false；有效则累计 video_earnings"""
|
||
video = await _get_published_video(db, video_id)
|
||
if not identity.subject:
|
||
raise BizException(errors.PARAM_ERROR, "缺少 X-Device-ID")
|
||
|
||
credited = False
|
||
is_author = bool(identity.user_id and video.user_id == identity.user_id)
|
||
# 时长合理性：不得明显低于视频时长（低于 50% 视为异常）
|
||
duration_ok = video.duration <= 0 or watch_duration >= video.duration * 0.5
|
||
if not is_author and duration_ok and watch_duration > 0:
|
||
ymd = f"{datetime.now():%Y%m%d}"
|
||
dedup_key = f"cview:{identity.subject}:{video_id}:{ymd}"
|
||
if await redis.set(dedup_key, 1, nx=True, ex=SET_TTL):
|
||
stmt = mysql_insert(VideoEarning).values(
|
||
video_id=video_id, author_id=video.user_id, complete_views=1
|
||
).on_duplicate_key_update(complete_views=VideoEarning.complete_views + 1)
|
||
await db.execute(stmt)
|
||
credited = True
|
||
# 记录实际观看时长（续播记录复用）
|
||
await _upsert_watch_record(db, identity, video_id, date.today(), 3, 0, 0, watch_duration)
|
||
await db.commit()
|
||
return {"credited": credited}
|
||
|
||
|
||
# ---------- 分类 ----------
|
||
|
||
async def list_categories(db: AsyncSession) -> list[dict]:
|
||
rows = (await db.execute(
|
||
select(Category).where(Category.status == 1).order_by(Category.sort, Category.id)
|
||
)).scalars().all()
|
||
return [{"category_id": c.id, "name": c.name} for c in rows]
|
||
|
||
|
||
# ---------- 上传发布（5.5 / 5.6 / 5.7） ----------
|
||
|
||
# 上传凭证（object_key 占位）有效期 24h
|
||
UPLOAD_TOKEN_TTL = 24 * 3600
|
||
|
||
|
||
def _upload_cfg() -> dict:
|
||
return biz_config.get("upload") or {}
|
||
|
||
|
||
async def get_upload_token(redis: aioredis.Redis, identity: Identity, req: UploadTokenReq) -> dict:
|
||
"""5.5 获取上传凭证：开发环境无 OSS，返回 server 中转模式，客户端改走 upload/file"""
|
||
ext = storage_service.video_ext_of(req.file_name)
|
||
if ext not in storage_service.VIDEO_EXTS:
|
||
raise BizException(errors.PARAM_ERROR, "仅支持 mp4/mov/m4v/avi/mkv/webm 格式")
|
||
max_mb = int(_upload_cfg().get("max_size_mb", 500))
|
||
if req.file_size > max_mb * 1024 * 1024:
|
||
raise BizException(errors.PARAM_ERROR, f"文件不能超过 {max_mb}MB")
|
||
object_key = f"upload/tmp/{datetime.now():%Y%m%d}/{uuid.uuid4().hex}.{ext}"
|
||
# 占位登记：upload/file 与发布时校验归属与有效期
|
||
await redis.set(
|
||
f"upload:obj:{object_key}", json.dumps({"uid": identity.user_id, "url": ""}), ex=UPLOAD_TOKEN_TTL
|
||
)
|
||
return {"upload_type": "server", "object_key": object_key, "upload_url": "/videos/upload/file"}
|
||
|
||
|
||
async def _load_pending_upload(redis: aioredis.Redis, object_key: str, user_id: int) -> dict:
|
||
raw = await redis.get(f"upload:obj:{object_key}")
|
||
pending = json.loads(raw) if raw else None
|
||
if not pending or pending.get("uid") != user_id:
|
||
raise BizException(errors.PARAM_ERROR, "object_key 无效或已过期")
|
||
return pending
|
||
|
||
|
||
async def register_upload(
|
||
redis: aioredis.Redis, identity: Identity, object_key: str, file: UploadFile
|
||
) -> dict:
|
||
"""upload/token 配套：multipart 中转上传，保存到本地并登记可访问 url"""
|
||
storage_service.validate_object_key(object_key)
|
||
await _load_pending_upload(redis, object_key, identity.user_id)
|
||
url, size = await storage_service.save_video(file, object_key)
|
||
await redis.set(
|
||
f"upload:obj:{object_key}",
|
||
json.dumps({"uid": identity.user_id, "url": url, "file_size": size}),
|
||
ex=UPLOAD_TOKEN_TTL,
|
||
)
|
||
return {"object_key": object_key, "url": url, "file_size": size}
|
||
|
||
|
||
async def publish_video(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity, req: PublishVideoReq
|
||
) -> dict:
|
||
"""5.6 发布：敏感词过滤 → 分类/上传校验 → 创建 Video（开发环境无转码器，直接待审核）"""
|
||
for text in [req.title, req.description, *req.tags]:
|
||
if text and await sensitive_service.find_sensitive(db, redis, text):
|
||
raise BizException(errors.SENSITIVE_WORD)
|
||
_check_tags(req.tags)
|
||
category = await db.get(Category, req.category_id)
|
||
if category is None or category.status != 1:
|
||
raise BizException(errors.PARAM_ERROR, "分类不存在")
|
||
pending = await _load_pending_upload(redis, req.object_key, identity.user_id)
|
||
if not pending.get("url"):
|
||
raise BizException(errors.PARAM_ERROR, "视频文件尚未上传")
|
||
|
||
is_priority = 1 if await membership_service.is_member(db, identity.user_id) else 0
|
||
video = Video(
|
||
user_id=identity.user_id,
|
||
category_id=req.category_id,
|
||
title=req.title,
|
||
description=req.description,
|
||
video_url=pending["url"],
|
||
file_size=int(pending.get("file_size") or 0),
|
||
status=1, # 待审核：开发环境无转码器，由后台审核通过后直接上架
|
||
is_priority=is_priority,
|
||
)
|
||
db.add(video)
|
||
await db.flush()
|
||
await _upsert_tags(db, video.id, req.tags)
|
||
await redis.delete(f"upload:obj:{req.object_key}")
|
||
await db.commit()
|
||
return {
|
||
"video_id": video.id,
|
||
"status": video.status,
|
||
"status_text": STATUS_TEXT[video.status],
|
||
"is_priority": bool(is_priority),
|
||
}
|
||
|
||
|
||
def _check_tags(names: list[str]) -> None:
|
||
"""标签数量与长度校验（0~5 个，每个 1~20 字）"""
|
||
if len(names) > 5:
|
||
raise BizException(errors.PARAM_ERROR, "标签最多 5 个")
|
||
if any(not (1 <= len(t) <= 20) for t in names):
|
||
raise BizException(errors.PARAM_ERROR, "每个标签 1~20 字")
|
||
|
||
|
||
async def _upsert_tags(db: AsyncSession, video_id: int, names: list[str]) -> None:
|
||
"""标签 upsert + 关联（去重保序，维护冗余 video_count）"""
|
||
for name in dict.fromkeys(names):
|
||
tag = (await db.execute(select(Tag).where(Tag.name == name))).scalar_one_or_none()
|
||
if tag is None:
|
||
tag = Tag(name=name)
|
||
db.add(tag)
|
||
await db.flush()
|
||
db.add(VideoTag(video_id=video_id, tag_id=tag.id))
|
||
tag.video_count += 1
|
||
|
||
|
||
async def _replace_tags(db: AsyncSession, video_id: int, names: list[str]) -> None:
|
||
"""编辑视频时重建标签关联：旧关联删除并回退计数 → 写新关联"""
|
||
old = (await db.execute(
|
||
select(VideoTag, Tag)
|
||
.join(Tag, Tag.id == VideoTag.tag_id)
|
||
.where(VideoTag.video_id == video_id)
|
||
)).all()
|
||
for link, tag in old:
|
||
tag.video_count = max(tag.video_count - 1, 0)
|
||
await db.delete(link)
|
||
await _upsert_tags(db, video_id, names)
|
||
|
||
|
||
# 软删除标记（模型 status：5 转码失败/已删除）
|
||
STATUS_DELETED = 5
|
||
|
||
|
||
async def _get_own_video(db: AsyncSession, identity: Identity, video_id: int) -> Video:
|
||
"""取本人作品：不存在/已删除 → 2003；非作者 → 2107"""
|
||
video = await db.get(Video, video_id)
|
||
if video is None or video.status == STATUS_DELETED:
|
||
raise BizException(errors.VIDEO_NOT_AVAILABLE)
|
||
if video.user_id != identity.user_id:
|
||
raise BizException(errors.NOT_AUTHOR)
|
||
return video
|
||
|
||
|
||
async def update_video(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity, video_id: int, req
|
||
) -> dict:
|
||
"""5.12 编辑视频（仅作者）：敏感词过滤 → 分类/标签校验 → 逐字段更新 → 失效详情缓存"""
|
||
video = await _get_own_video(db, identity, video_id)
|
||
|
||
texts = [t for t in (req.title, req.description) if t] + list(req.tags or [])
|
||
for text in texts:
|
||
if await sensitive_service.find_sensitive(db, redis, text):
|
||
raise BizException(errors.SENSITIVE_WORD)
|
||
|
||
if req.title is not None:
|
||
video.title = req.title
|
||
if req.description is not None:
|
||
video.description = req.description
|
||
if req.category_id is not None:
|
||
category = await db.get(Category, req.category_id)
|
||
if category is None or category.status != 1:
|
||
raise BizException(errors.PARAM_ERROR, "分类不存在")
|
||
video.category_id = req.category_id
|
||
if req.tags is not None:
|
||
_check_tags(req.tags)
|
||
await _replace_tags(db, video.id, req.tags)
|
||
|
||
await db.commit()
|
||
await invalidate_detail_cache(redis, video.id)
|
||
return {"video_id": video.id}
|
||
|
||
|
||
async def delete_video(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity, video_id: int
|
||
) -> None:
|
||
"""5.13 删除视频（仅作者，软删除）：置已删除状态 → 立即下架不可见，互动计数保留"""
|
||
video = await _get_own_video(db, identity, video_id)
|
||
video.status = STATUS_DELETED
|
||
await db.commit()
|
||
await invalidate_detail_cache(redis, video.id)
|
||
|
||
|
||
async def list_my_videos(
|
||
db: AsyncSession, user_id: int, page: int, page_size: int, status: int | None
|
||
) -> dict:
|
||
"""5.7 我的作品：按创建倒序（排除已删除），列表项附加 status/status_text/reject_reason"""
|
||
stmt = select(Video).where(Video.user_id == user_id, Video.status != STATUS_DELETED)
|
||
if status is not None:
|
||
stmt = stmt.where(Video.status == status)
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
|
||
rows = list(
|
||
(await db.execute(stmt.order_by(Video.id.desc()).offset((page - 1) * page_size).limit(page_size))).scalars()
|
||
)
|
||
items = await serialize_videos(db, rows, user_id)
|
||
for item, v in zip(items, rows):
|
||
item.update({
|
||
"status": v.status,
|
||
"status_text": STATUS_TEXT.get(v.status, ""),
|
||
"reject_reason": v.reject_reason,
|
||
"created_at": _fmt(v.created_at),
|
||
})
|
||
return {"total": total, "list": items}
|
||
|
||
|
||
# ---------- 搜索 / 热搜 / 标签（5.9 / 5.10 / 5.11） ----------
|
||
|
||
# 热搜 ZSet（搜索计数，对齐《03-接口文档》5.10/5.11）
|
||
HOT_SEARCH_KEY = "search:hot"
|
||
# 热搜榜返回条数
|
||
HOT_SEARCH_TOP = 10
|
||
|
||
|
||
def _hot_order(stmt):
|
||
"""hot 排序：view×1 + like×5 + comment×10（与 Feed 热度分同权，SQL 内计算）"""
|
||
hot = Video.view_count + Video.like_count * 5 + Video.comment_count * 10
|
||
return stmt.order_by(hot.desc(), Video.id.desc())
|
||
|
||
|
||
def _latest_order(stmt):
|
||
return stmt.order_by(Video.published_at.desc(), Video.id.desc())
|
||
|
||
|
||
async def search_videos(
|
||
db: AsyncSession, redis: aioredis.Redis, identity: Identity,
|
||
keyword: str, sort: str, page: int, page_size: int,
|
||
) -> dict:
|
||
"""5.10 视频搜索：标题/标签 LIKE 匹配；命中敏感词返回空结果；搜索行为计入热搜 ZSet"""
|
||
if await sensitive_service.find_sensitive(db, redis, keyword):
|
||
return {"total": 0, "list": []}
|
||
await redis.zincrby(HOT_SEARCH_KEY, 1, keyword)
|
||
|
||
kw = f"%{keyword}%"
|
||
tag_video_ids = select(VideoTag.video_id).join(Tag, Tag.id == VideoTag.tag_id).where(Tag.name.like(kw))
|
||
stmt = select(Video).where(
|
||
Video.status == STATUS_PUBLISHED,
|
||
Video.title.like(kw) | Video.id.in_(tag_video_ids),
|
||
)
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
|
||
stmt = _latest_order(stmt) if sort == "latest" else _hot_order(stmt)
|
||
rows = list((await db.execute(stmt.offset((page - 1) * page_size).limit(page_size))).scalars())
|
||
return {"total": total, "list": await serialize_videos(db, rows, identity.user_id)}
|
||
|
||
|
||
async def get_hot_search(redis: aioredis.Redis) -> dict:
|
||
"""5.11 热搜榜：search:hot ZSet Top10"""
|
||
words = await redis.zrevrange(HOT_SEARCH_KEY, 0, HOT_SEARCH_TOP - 1)
|
||
return {"words": list(words)}
|
||
|
||
|
||
async def get_tag_videos(
|
||
db: AsyncSession, identity: Identity, tag_name: str, sort: str, page: int, page_size: int
|
||
) -> dict:
|
||
"""5.9 标签视频列表：标签不存在时返回空（tag 字段回显原名）"""
|
||
tag = (await db.execute(select(Tag).where(Tag.name == tag_name))).scalar_one_or_none()
|
||
if tag is None:
|
||
return {"tag": tag_name, "total": 0, "list": []}
|
||
stmt = (
|
||
select(Video)
|
||
.join(VideoTag, VideoTag.video_id == Video.id)
|
||
.where(VideoTag.tag_id == tag.id, Video.status == STATUS_PUBLISHED)
|
||
)
|
||
total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
|
||
stmt = _latest_order(stmt) if sort == "latest" else _hot_order(stmt)
|
||
rows = list((await db.execute(stmt.offset((page - 1) * page_size).limit(page_size))).scalars())
|
||
return {"tag": tag.name, "total": total, "list": await serialize_videos(db, rows, identity.user_id)}
|