bwstudioandClaude Opus 4.7
c3a8e3e311
- jokes.py: Public joke listing, detail, and random endpoints - categories.py: Type and crowd listing endpoints - auth.py: Login endpoint with JWT token generation - admin.py: Full CRUD for jokes, types, crowds with JWT protection - Fix request body schemas to use Create DTOs instead of Response models Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import random
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models.joke import Joke
|
|
from app.schemas.joke import JokeResponse, PaginatedJokeResponse
|
|
|
|
router = APIRouter(prefix="/jokes", tags=["笑话"])
|
|
|
|
|
|
def joke_to_response(joke: Joke) -> JokeResponse:
|
|
"""Convert Joke model to JokeResponse schema."""
|
|
return JokeResponse(
|
|
id=joke.id,
|
|
title=joke.title,
|
|
content=joke.content,
|
|
type_id=joke.type_id,
|
|
crowd_id=joke.crowd_id,
|
|
status=joke.status,
|
|
view_count=joke.view_count,
|
|
like_count=joke.like_count,
|
|
created_at=joke.created_at,
|
|
updated_at=joke.updated_at,
|
|
type_name=joke.type.name if joke.type else None,
|
|
crowd_name=joke.crowd.name if joke.crowd else None,
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=PaginatedJokeResponse)
|
|
def list_jokes(
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
type_id: int | None = None,
|
|
crowd_id: int | None = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""获取笑话列表（仅返回已审核通过的笑话）"""
|
|
query = db.query(Joke).filter(Joke.status == "approved")
|
|
|
|
if type_id is not None:
|
|
query = query.filter(Joke.type_id == type_id)
|
|
if crowd_id is not None:
|
|
query = query.filter(Joke.crowd_id == crowd_id)
|
|
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
jokes = query.order_by(Joke.created_at.desc()).offset(offset).limit(page_size).all()
|
|
|
|
return PaginatedJokeResponse(
|
|
items=[joke_to_response(j) for j in jokes],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.get("/{joke_id}", response_model=JokeResponse)
|
|
def get_joke(joke_id: int, db: Session = Depends(get_db)):
|
|
"""获取单条笑话详情"""
|
|
joke = db.query(Joke).filter(Joke.id == joke_id).first()
|
|
if not joke:
|
|
raise HTTPException(status_code=404, detail="笑话不存在")
|
|
# 增加浏览次数
|
|
joke.view_count += 1
|
|
db.commit()
|
|
return joke_to_response(joke)
|
|
|
|
|
|
@router.get("/random", response_model=JokeResponse)
|
|
def get_random_joke(db: Session = Depends(get_db)):
|
|
"""随机获取一条已审核通过的笑话"""
|
|
joke = db.query(Joke).filter(Joke.status == "approved").order_by(func.random()).first()
|
|
if not joke:
|
|
raise HTTPException(status_code=404, detail="暂无笑话")
|
|
return joke_to_response(joke) |