项目1 - 个人博客系统¶

# backend/app/models.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Table
from sqlalchemy.orm import relationship
from datetime import datetime, timezone
from database import Base

# 文章-标签关联表
post_tag = Table(
    'post_tag',
    Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)

# 用户模型：存储博客系统的用户信息
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)  # 用户名，唯一
    email = Column(String(100), unique=True, nullable=False)    # 邮箱，唯一
    hashed_password = Column(String(255), nullable=False)       # 密码哈希值（不存明文）
    is_admin = Column(Integer, default=0)                       # 是否管理员：0=普通用户，1=管理员
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))  # 注册时间

    posts = relationship("Post", back_populates="author")  # 一对多：用户拥有多篇文章

# 文章模型：博客核心内容
class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)         # 文章标题
    slug = Column(String(200), unique=True, nullable=False)  # URL友好的短标识
    content = Column(Text, nullable=False)               # 文章正文（Markdown格式）
    summary = Column(String(500))                        # 文章摘要
    cover_image = Column(String(500))                    # 封面图片URL
    is_published = Column(Integer, default=0)            # 发布状态：0=草稿，1=已发布
    view_count = Column(Integer, default=0)              # 浏览次数
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))  # lambda匿名函数：简洁的单行函数
    updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))

    # 外键关联
    author_id = Column(Integer, ForeignKey('users.id'))       # 作者ID
    category_id = Column(Integer, ForeignKey('categories.id'))  # 分类ID

    # 关系映射
    author = relationship("User", back_populates="posts")
    category = relationship("Category", back_populates="posts")
    tags = relationship("Tag", secondary=post_tag, back_populates="posts")  # 多对多：文章与标签
    comments = relationship("Comment", back_populates="post")  # 一对多：文章的评论

class Category(Base):
    __tablename__ = 'categories'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    slug = Column(String(50), unique=True, nullable=False)

    posts = relationship("Post", back_populates="category")

class Tag(Base):
    __tablename__ = 'tags'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    slug = Column(String(50), unique=True, nullable=False)

    posts = relationship("Post", secondary=post_tag, back_populates="tags")

# 评论模型：访客对文章的评论
class Comment(Base):
    __tablename__ = 'comments'

    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)                  # 评论内容
    author_name = Column(String(50), nullable=False)        # 评论者昵称
    author_email = Column(String(100), nullable=False)      # 评论者邮箱
    is_approved = Column(Integer, default=0)                # 审核状态：0=待审核，1=已通过
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    post_id = Column(Integer, ForeignKey('posts.id'))       # 所属文章ID
    post = relationship("Post", back_populates="comments")

# backend/app/routers/posts.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
import schemas, crud
from database import get_db

router = APIRouter(prefix="/posts", tags=["posts"])

@router.get("/", response_model=list[schemas.PostList])
def list_posts(
    skip: int = 0,
    limit: int = 10,
    category: str | None = None,
    tag: str | None = None,
    search: str | None = None,
    db: Session = Depends(get_db)
):
    """获取文章列表"""
    posts = crud.get_posts(
        db, skip=skip, limit=limit,
        category=category, tag=tag, search=search
    )
    return posts

@router.get("/{slug}", response_model=schemas.PostDetail)
def get_post(slug: str, db: Session = Depends(get_db)):
    """获取文章详情"""
    post = crud.get_post_by_slug(db, slug=slug)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    # 增加浏览量
    crud.increment_view_count(db, post.id)
    return post

@router.post("/", response_model=schemas.PostDetail)
def create_post(
    post: schemas.PostCreate,
    db: Session = Depends(get_db),
    current_user: schemas.User = Depends(get_current_user)
):
    """创建文章（需要登录）"""
    return crud.create_post(db, post, author_id=current_user.id)

@router.put("/{post_id}", response_model=schemas.PostDetail)
def update_post(
    post_id: int,
    post: schemas.PostUpdate,
    db: Session = Depends(get_db),
    current_user: schemas.User = Depends(get_current_user)
):
    """更新文章"""
    db_post = crud.get_post(db, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="Post not found")
    if db_post.author_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Not authorized")
    return crud.update_post(db, post_id, post)

@router.delete("/{post_id}")
def delete_post(
    post_id: int,
    db: Session = Depends(get_db),
    current_user: schemas.User = Depends(get_current_user)
):
    """删除文章"""
    db_post = crud.get_post(db, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="Post not found")
    if db_post.author_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Not authorized")
    crud.delete_post(db, post_id)
    return {"message": "Post deleted"}

# backend/app/auth.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

# JWT配置常量
SECRET_KEY = "your-secret-key"          # 密钥（生产环境应使用环境变量）
ALGORITHM = "HS256"                      # JWT签名算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30         # Token有效期（分钟）

# 密码加密上下文，使用bcrypt算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security = HTTPBearer()

def verify_password(plain_password, hashed_password):
    """验证明文密码与哈希值是否匹配"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    """将明文密码转换为bcrypt哈希值"""
    return pwd_context.hash(password)

def create_access_token(data: dict):
    """生成JWT访问令牌，包含用户信息和过期时间"""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})  # 设置过期时间
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):  # async def定义异步函数；用await调用
    """从请求头中提取并验证JWT，返回当前登录用户"""
    token = credentials.credentials
    try:  # try/except捕获异常
        # 解码JWT并提取用户ID
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        # 从数据库获取用户信息
        return get_user_by_id(int(user_id))
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")