blog-service/service/blog_service.py

from typing import List

from fastapi import Request
from sqlalchemy import select, func, desc, and_
from sqlalchemy.orm import Session

from models.blog import Blog, BlogCategory, BlogVisit, BlogContent, BlogComment
from schemas.blog import BlogResponse, BlogQuery, BlogCategoryResponse, BlogStatsResponse, \
    BlogLatestResponse, BlogCommentCreate, BlogCommentResponse
from schemas.pagination import PageResult
from schemas.paginate_query import paginate_query
from middleware.exceptions import AppException


def query_blog_by_page(db: Session, current_page: int = 1, page_size: int = 10) -> PageResult[BlogResponse]:
    query = select(
        Blog.id,
        Blog.title,
        Blog.top_value.label("topValue"),
        Blog.is_great.label("isGreat"),
        BlogCategory.name.label("category"),
        Blog.summary,
        Blog.word_count.label("wordCount"),
        Blog.read_duration.label("readDuration"),
        func.count(BlogVisit.id).label("visitCount"),
        Blog.create_time.label("createTime"),
        Blog.update_time.label("updateTime")
    ).select_from(Blog).outerjoin(
        BlogCategory, Blog.category_id == BlogCategory.id
    ).outerjoin(
        BlogVisit, Blog.id == BlogVisit.blog_id
    ).group_by(
        Blog.id, BlogCategory.name
    ).order_by(
        desc(Blog.top_value), desc(Blog.update_time)
    )

    return paginate_query(db, query, current_page, page_size)


def query_blog_by_condition(db: Session, blog_query: BlogQuery) -> List[BlogResponse]:
    query = select(
        Blog.id,
        Blog.title,
        Blog.top_value.label("topValue"),
        Blog.is_great.label("isGreat"),
        BlogCategory.name.label("category"),
        Blog.summary,
        BlogContent.content.label("content"),
        Blog.word_count.label("wordCount"),
        Blog.read_duration.label("readDuration"),
        Blog.create_time.label("createTime"),
        Blog.update_time.label("updateTime")
    ).select_from(Blog).outerjoin(
        BlogCategory, Blog.category_id == BlogCategory.id
    ).outerjoin(
        BlogContent, Blog.content_id == BlogContent.id
    )

    conditions = []

    if blog_query.category:
        conditions.append(BlogCategory.name == blog_query.category)

    if blog_query.title:
        conditions.append(Blog.title.like(f"%{blog_query.title}%"))

    if blog_query.year:
        conditions.append(func.extract('year', Blog.create_time) == blog_query.year)

    if conditions:
        query = query.where(and_(*conditions))

    query = query.order_by(desc(Blog.update_time))

    results = db.execute(query).fetchall()

    return [BlogResponse.from_orm(result) for result in results]


def query_blog_by_id(db: Session, blog_id: int) -> BlogResponse:
    query = db.query(
        Blog.id.label("id"),
        Blog.title.label("title"),
        Blog.top_value.label("topValue"),
        Blog.is_great.label("isGreat"),
        BlogCategory.name.label("category"),
        Blog.summary.label("summary"),
        BlogContent.content.label("content"),
        Blog.word_count.label("wordCount"),
        Blog.read_duration.label("readDuration"),
        func.count(BlogVisit.id).label("visitCount"),
        Blog.create_time.label("createTime"),
        Blog.update_time.label("updateTime")
    ).select_from(Blog).outerjoin(
        BlogCategory, Blog.category_id == BlogCategory.id
    ).outerjoin(
        BlogContent, Blog.content_id == BlogContent.id
    ).outerjoin(
        BlogVisit, Blog.id == BlogVisit.blog_id
    ).filter(Blog.id == blog_id)

    blog = query.first()
    if blog[0] is None:
        raise AppException("博客不存在")

    return BlogResponse.from_orm(blog)


def query_blog_category(db: Session) -> List[BlogCategoryResponse]:
    query = select(
        BlogCategory.name,
        func.count(Blog.id).label("count")
    ).select_from(Blog).outerjoin(
        BlogCategory, Blog.category_id == BlogCategory.id
    ).group_by(
        BlogCategory.name
    )

    results = db.execute(query).fetchall()

    return [BlogCategoryResponse.from_orm(result) for result in results]


def query_blog_stats(db: Session) -> BlogStatsResponse:
    return BlogStatsResponse(
        blogCount=db.query(Blog).count(),
        categoryCount=db.query(BlogCategory).count(),
        wordCount=db.query(func.sum(Blog.word_count)).scalar()
    )


def query_blog_latest(db: Session) -> List[BlogLatestResponse]:
    query = select(
        Blog.id,
        Blog.title
    ).select_from(Blog).order_by(desc(Blog.update_time)).limit(5)

    results = db.execute(query).fetchall()

    return [BlogLatestResponse.from_orm(result) for result in results]


def add_blog_visit(db: Session, blog_visit: BlogVisit):
    db.add(blog_visit)
    db.commit()
    db.refresh(blog_visit)


def query_blog_comment(db: Session, blog_id: int) -> List[BlogCommentResponse]:
    query = select(
        BlogComment.id,
        BlogComment.blog_id.label("blogId"),
        BlogComment.parent_id.label("parentId"),
        BlogComment.name,
        BlogComment.website,
        BlogComment.ip_address.label("ipAddress"),
        BlogComment.user_agent.label("userAgent"),
        BlogComment.content,
        BlogComment.is_approved.label("isApproved"),
        BlogComment.create_time.label("createTime")
    ).select_from(BlogComment).where(BlogComment.blog_id == blog_id)
    results = db.execute(query).fetchall()

    return [BlogCommentResponse.from_orm(result) for result in results]


def add_blog_comment(db: Session, request: Request, blog_comment: BlogCommentCreate) -> bool:
    blog = db.query(Blog).filter(Blog.id == blog_comment.blogId).first()
    if not blog:
        raise AppException("博客不存在")

    if blog_comment.parentId != 0:
        parent_comment = db.query(BlogComment).filter(
            BlogComment.id == blog_comment.parentId,
            BlogComment.blog_id == blog_comment.blogId
        ).first()

        if not parent_comment:
            raise AppException("父评论不存在")

    db_comment = BlogComment(
        blog_id=blog_comment.blogId,
        parent_id=blog_comment.parentId,
        name=blog_comment.name,
        website=blog_comment.website,
        content=blog_comment.content,
        ip_address=request.client.host,
        user_agent=request.headers.get("user-agent"),
        is_approved=1
    )

    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)

    return True