blog_api/routers/blogmanage.py

from fastapi import Depends, APIRouter, Query, Path
from internal.models import *
from internal.database import (
    fetch_one,
    fetch_all,
    execute_query,
    response_success,
    raise_if_exists,
    raise_if_not_found,
)
from dependencies import get_current_active_user
import json
router = APIRouter(prefix="/blogs", tags=["博客管理"])

# 获取列表


@router.get("/list")
async def blog_list(page: int = Query(None), page_size: int = Query(None)):
    limit_clause = ""
    if page is not None and page_size is not None:
        offset = (page - 1) * page_size
        limit_clause = f"LIMIT {page_size} OFFSET {offset}"
    # 列表参数：博客名称、博客内容、创建时间、博客图片、博客查看时间、博客阅读次数、博客字数、类型名称、标签名列表
    select_query = f"""
        SELECT blogs.id,blogs.blogtitle, blogs.blogcontent, blogs.create_at, blogs.imglink,
        blogs.wordcount, blogtypes.typename,JSON_ARRAYAGG(labels.labelname) AS labelnames
        FROM blogs
        LEFT JOIN `blogtypes` ON blogs.typeid = blogtypes.id
        LEFT JOIN blog_label ON blog_label.blogid = blogs.id
        LEFT JOIN labels ON blog_label.labelid = labels.id
        GROUP BY blogs.id, blogs.blogtitle, blogs.blogcontent, blogs.create_at, blogs.imglink,
        blogs.wordcount, blogtypes.typename ORDER BY create_at DESC
        {limit_clause};
    """
    blog_list = fetch_all(select_query)
    count_query = "SELECT COUNT(*) AS total FROM blogs;"
    total_records = fetch_one(count_query)["total"]
    return response_success({
        "blogs": blog_list,
        "total": total_records,
    }, "blog get list success")


@router.get("/list/{id}")
async def blog_one(id: int):

    # 列表参数：博客名称、博客内容、创建时间、博客图片、博客查看时间、博客阅读次数、博客字数、类型名称、标签名列表
    select_query = """
        SELECT id, blogtitle, blogcontent FROM blogs
        WHERE id = %s
        ORDER BY create_at DESC;
    """
    blog_one = fetch_one(select_query, (id,))
    return response_success(blog_one, "blog get blog_one success")


# 博客新增
@router.post("/add")
async def blog_add(blog: Blog, labels: list[Label], _: User = Depends(get_current_active_user)):
    select_query = "SELECT * FROM blogs WHERE blogtitle = %s"
    existing_blog = fetch_one(select_query, (blog.blogtitle,))
    raise_if_exists(existing_blog, "Blog already exists")
    insert_query = (
        "INSERT INTO blogs (blogtitle, blogcontent,imglink, typeid, descr) VALUES (%s, %s, %s, %s,%s)"
    )
    insert_data = (blog.blogtitle, blog.blogcontent,
                   blog.imglink, blog.typeid, blog.descr)
    blog_id = execute_query(insert_query, insert_data, lastrowid=True)
    for label in labels:
        insert_label_query = "INSERT INTO blog_label (blogid, labelid) VALUES (%s, %s)"
        execute_query(insert_label_query, (blog_id, label.id))

    return {"message": "Blog created successfully", "blog_id": blog_id}


# 博客删除
@router.delete("/delete/{id}")
async def blog_delete(id: str = Path(description="博客id")):
    select_query = "SELECT * FROM blogs WHERE id = %s"
    existing_blog = fetch_one(select_query, (id,))
    raise_if_not_found(existing_blog, "blog not found")
    insert_query = "DELETE FROM blogs WHERE id = %s"
    execute_query(insert_query, (id,))
    return response_success(message="blog delete success")


@router.put("/update/{id}")
async def blog_update(id: int, blog: Blog, labels: list[Label], _: User = Depends(get_current_active_user)):
    # 检查要编辑的博客是否存在
    select_query = "SELECT * FROM blogs WHERE id = %s"
    existing_blog = fetch_one(select_query, (id,))
    raise_if_not_found(existing_blog, "blog not found")

    # 更新博客信息
    update_query = (
        "UPDATE blogs SET blogtitle = %s, blogcontent = %s, imglink = %s, typeid = %s, descr = %s WHERE id = %s"
    )
    update_data = (blog.blogtitle, blog.blogcontent,
                   blog.imglink, blog.typeid, blog.descr, id)
    execute_query(update_query, update_data)

    # 首先删除原有的关联标签
    delete_query = "DELETE FROM blog_label WHERE blogid = %s"
    execute_query(delete_query, (id,))

    # 然后插入新的关联标签
    for label in labels:
        insert_label_query = "INSERT INTO blog_label (blogid, labelid) VALUES (%s, %s)"
        execute_query(insert_label_query, (id, label.id))
    return response_success("blog update sucess")


# 博客模糊查询
@router.get("/list/search")
async def blog_list_search(
    blogtitle: str = Query(None, description="博客标题"),
    typename: str = Query(None, description="博客类型"),
    start_date: str = Query(None, description="开始时间"),
    end_date: str = Query(None, description="结束时间"),
):
    select_query = """
        SELECT blogs.id, blogtitle, blogcontent,wordcount, typename, create_at, update_at, blogs.descr,JSON_ARRAYAGG(labels.labelname) AS labelnames  
        FROM blogs 
        LEFT JOIN `blogtypes` ON blogs.typeid = blogtypes.id 
        LEFT JOIN blog_label ON blogs.id = blog_label.blogid
        LEFT JOIN labels ON blog_label.labelid = labels.id
        WHERE 1=1
        """
    params = []
    if blogtitle:
        select_query += " AND blogtitle LIKE %s"
        params.append(f"%{blogtitle}%")
    if typename:
        select_query += " AND typename LIKE %s"
        params.append(f"%{typename}%")
    if start_date:
        select_query += " AND create_at >= %s"
        params.append(start_date)
    if end_date:
        select_query += " AND create_at <= %s"
        params.append(end_date)
    select_query += "GROUP BY blogs.id, blogs.blogtitle, blogs.blogcontent,blogs.wordcount, blogtypes.typename, blogs.create_at, blogs.update_at, blogs.descr ORDER BY create_at DESC"
    blog_list = fetch_all(select_query, params=params, fetchall=True)
    return response_success(data=blog_list, message="blog serach succuessfully!")


# 根据id查询博客
@router.get("/list/search/{id}")
async def get_id_blog(id: str = Path(description="博客id")):
    select_query = """SELECT blogs.id, blogtitle, blogcontent,wordcount, blogs.typeid, blogs.descr,JSON_ARRAYAGG(labels.id) AS labelnames,imglink  FROM blogs 
                LEFT JOIN `blogtypes` ON blogs.typeid = blogtypes.id 
                LEFT JOIN blog_label ON blogs.id = blog_label.blogid
                LEFT JOIN labels ON blog_label.labelid = labels.id
                WHERE blogs.id = %s
                GROUP BY blogs.id;

"""
    blog_list = execute_query(select_query, (id,))

    if blog_list and isinstance(blog_list, dict):
        if 'labelnames' in blog_list and isinstance(blog_list['labelnames'], str):
            try:
                blog_list['labelnames'] = json.loads(blog_list['labelnames'])
            except json.JSONDecodeError:
                blog_list['labelnames'] = []
    return response_success(data=blog_list, message="blog search success")


# 我就测试一下