85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""
|
|
点赞路由。
|
|
|
|
接口（统一响应格式 {success, data, message}）：
|
|
- POST /api/like/add 点赞（好友或博主，同一用户不能重复点赞）
|
|
- GET /api/like/list 点赞列表（可见性与文章一致）
|
|
"""
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..auth import get_current_user
|
|
from ..database import get_db
|
|
from ..models import (
|
|
ROLE_BLOGGER,
|
|
ROLE_FRIEND,
|
|
VISIBILITY_FRIEND,
|
|
Article,
|
|
Like,
|
|
User,
|
|
)
|
|
from ..schemas import LikeCreate, UnifiedResponse
|
|
from ..security import action_limiter, get_client_ip
|
|
from .deps import can_read_friend_article, get_optional_user
|
|
|
|
router = APIRouter(prefix="/api/like", tags=["like"])
|
|
|
|
|
|
@router.post("/add", response_model=UnifiedResponse)
|
|
def add_like(
|
|
payload: LikeCreate,
|
|
request: Request,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> UnifiedResponse:
|
|
"""点赞：仅好友或博主可点赞，同一用户不能重复点赞。"""
|
|
# 写操作限流（IP 维度）：防止好友账号刷点赞
|
|
if action_limiter.is_blocked(get_client_ip(request)):
|
|
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="操作过于频繁，请稍后再试")
|
|
if current_user.role not in (ROLE_FRIEND, ROLE_BLOGGER):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有好友或博主可以点赞")
|
|
article = db.query(Article).filter(Article.id == payload.article_id).first()
|
|
if article is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文章不存在")
|
|
exists = (
|
|
db.query(Like)
|
|
.filter(Like.article_id == payload.article_id, Like.user_id == current_user.id)
|
|
.first()
|
|
)
|
|
if exists is not None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="不能重复点赞")
|
|
|
|
like = Like(article_id=payload.article_id, user_id=current_user.id)
|
|
db.add(like)
|
|
db.commit()
|
|
db.refresh(like)
|
|
action_limiter.hit(get_client_ip(request))
|
|
|
|
data = {"id": like.id, "article_id": like.article_id, "user_id": like.user_id}
|
|
return UnifiedResponse(success=True, data=data, message="点赞成功")
|
|
|
|
|
|
@router.get("/list", response_model=UnifiedResponse)
|
|
def list_likes(
|
|
article_id: int,
|
|
current_user: Optional[User] = Depends(get_optional_user),
|
|
db: Session = Depends(get_db),
|
|
) -> UnifiedResponse:
|
|
"""点赞列表：公开文章所有人可见，好友文章仅好友/博主可见。"""
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if article is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文章不存在")
|
|
if article.visibility == VISIBILITY_FRIEND and not can_read_friend_article(current_user):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权查看该文章点赞")
|
|
|
|
likes = db.query(Like).filter(Like.article_id == article_id).all()
|
|
# 不返回点赞用户名单（避免泄露用户名与账号关联，防枚举），
|
|
# 仅返回点赞数量与当前登录用户是否已点赞
|
|
data = {
|
|
"count": len(likes),
|
|
"includes_me": current_user is not None and any(l.user_id == current_user.id for l in likes),
|
|
}
|
|
return UnifiedResponse(success=True, data=data, message="获取成功") |