You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
162 lines
6.6 KiB
162 lines
6.6 KiB
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():
|
|
|
|
# 列表参数：博客名称、博客内容、创建时间、博客图片、博客查看时间、博客阅读次数、博客字数、类型名称、标签名列表
|
|
select_query = """
|
|
SELECT blogs.id,blogs.blogtitle, blogs.blogcontent, blogs.create_at, blogs.imglink,
|
|
blogs.wordcount, types.typename,JSON_ARRAYAGG(labels.labelname) AS labelnames
|
|
FROM blogs
|
|
LEFT JOIN `types` ON blogs.typeid = types.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, types.typename ORDER BY create_at DESC;
|
|
"""
|
|
blog_list = fetch_all(select_query)
|
|
return response_success(blog_list, "blog get list 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_put(blog: Blog, 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")
|
|
# update_query = (
|
|
# "UPDATE blogs SET blogtitle=%s,blogcontent=%s,typeid=%s,descr=%s WHERE id=%s;"
|
|
# )
|
|
# update_data = (blog.blogtitle, blog.blogcontent,
|
|
# blog.typeid, blog.descr, id)
|
|
# execute_query(update_query, update_data)
|
|
# return response_success("blog update sucess")
|
|
|
|
@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 `types` ON blogs.typeid = types.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, types.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 `types` ON blogs.typeid = types.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")
|
|
|
|
|
|
# 我就测试一下
|