374 lines
11 KiB
Python
374 lines
11 KiB
Python
from typing import List
|
|
|
|
from fastapi import Request
|
|
from sqlalchemy import select, func, desc, and_, asc, delete
|
|
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, BlogVisitResponse, BlogCreate, BlogUpdate, \
|
|
BlogAdjacentResponse
|
|
from schemas.blog_elastic import BlogElastic
|
|
from schemas.pagination import PageResult
|
|
from schemas.paginate_query import paginate_query
|
|
from middleware.exceptions import AppException
|
|
from service.blog_elastic_service import add_blog_elastic
|
|
from utils.blog_utils import get_blog_summary, get_word_count, get_read_duration
|
|
|
|
|
|
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:
|
|
check_blog_exist(db, blog_id)
|
|
|
|
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()
|
|
|
|
return BlogResponse.from_orm(blog)
|
|
|
|
|
|
def add_blog(db: Session, blog: BlogCreate) -> bool:
|
|
word_count = get_word_count(blog.content)
|
|
|
|
db_blog = Blog(
|
|
title=blog.title,
|
|
top_value=blog.topValue,
|
|
is_great=blog.isGreat,
|
|
category_id=add_blog_category(db, blog.category),
|
|
content_id=add_blog_content(db, blog.content),
|
|
summary=get_blog_summary(blog.content),
|
|
word_count=word_count,
|
|
read_duration=get_read_duration(word_count)
|
|
)
|
|
|
|
db.add(db_blog)
|
|
db.commit()
|
|
db.refresh(db_blog)
|
|
|
|
blog_elastic = BlogElastic(
|
|
id=db_blog.id,
|
|
title=db_blog.title,
|
|
content=blog.content,
|
|
category=blog.category
|
|
)
|
|
add_blog_elastic(blog_elastic)
|
|
|
|
return True
|
|
|
|
|
|
def add_blog_category(db: Session, category: str) -> int:
|
|
db_blog_category = db.execute(select(BlogCategory).where(BlogCategory.name == category)).scalar_one_or_none()
|
|
|
|
if db_blog_category is None:
|
|
db_new_blog_category = BlogCategory(name=category)
|
|
db.add(db_new_blog_category)
|
|
db.commit()
|
|
category_id = db_new_blog_category.id
|
|
else:
|
|
category_id = db_blog_category.id
|
|
|
|
return category_id
|
|
|
|
|
|
def add_blog_content(db: Session, content: str) -> int:
|
|
db_blog_content = BlogContent(content=content.encode('utf-8'))
|
|
db.add(db_blog_content)
|
|
db.commit()
|
|
|
|
return db_blog_content.id
|
|
|
|
|
|
def update_blog(db: Session, blog_id: int, blog: BlogUpdate) -> bool:
|
|
db_blog = check_blog_exist(db, blog_id)
|
|
|
|
update_data = blog.model_dump(exclude_unset=True)
|
|
|
|
update_blog_content(db, db_blog.content_id, blog.content)
|
|
|
|
db_blog.title = update_data["title"]
|
|
db_blog.top_value = update_data["topValue"]
|
|
db_blog.is_great = update_data["isGreat"]
|
|
db_blog.category_id = update_blog_category(db, blog.category)
|
|
|
|
word_count = get_word_count(blog.content)
|
|
db_blog.summary = get_blog_summary(blog.content),
|
|
db_blog.word_count = word_count
|
|
db_blog.read_duration = get_read_duration(word_count)
|
|
|
|
db.commit()
|
|
db.refresh(db_blog)
|
|
|
|
return True
|
|
|
|
|
|
def update_blog_category(db: Session, category: str) -> int:
|
|
db_blog_category = db.execute(select(BlogCategory).where(BlogCategory.name == category)).scalar_one_or_none()
|
|
|
|
if db_blog_category is None:
|
|
return add_blog_category(db, category)
|
|
else:
|
|
return db_blog_category.id
|
|
|
|
|
|
def update_blog_content(db: Session, blog_content_id: int, blog_content: str) -> bool:
|
|
db_blog_content = check_blog_content_exist(db, blog_content_id)
|
|
|
|
db_blog_content.content = blog_content.encode('utf-8')
|
|
db.commit()
|
|
db.refresh(db_blog_content)
|
|
|
|
return True
|
|
|
|
|
|
def delete_blog(db: Session, blog_id: int) -> bool:
|
|
db_blog = check_blog_exist(db, blog_id)
|
|
|
|
delete_blog_content(db, db_blog.content_id)
|
|
db.execute(delete(Blog).where(Blog.id == blog_id))
|
|
db.commit()
|
|
|
|
return True
|
|
|
|
|
|
def delete_blog_content(db: Session, blog_content_id: int) -> bool:
|
|
check_blog_content_exist(db, blog_content_id)
|
|
|
|
db.execute(delete(BlogContent).where(BlogContent.id == blog_content_id))
|
|
db.commit()
|
|
|
|
return True
|
|
|
|
|
|
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 query_blog_adjacent(db: Session, blog_id: int) -> List[BlogAdjacentResponse]:
|
|
check_blog_exist(db, blog_id)
|
|
|
|
prev_result = db.execute(
|
|
select(Blog.id, Blog.title)
|
|
.where(Blog.id < blog_id)
|
|
.order_by(desc(Blog.id))
|
|
.limit(1)
|
|
).first()
|
|
|
|
next_result = db.execute(
|
|
select(Blog.id, Blog.title)
|
|
.where(Blog.id > blog_id)
|
|
.order_by(asc(Blog.id))
|
|
.limit(1)
|
|
).first()
|
|
|
|
return [
|
|
BlogAdjacentResponse(
|
|
id=prev_result[0] if prev_result else 0,
|
|
title=prev_result[1] if prev_result else ""
|
|
),
|
|
BlogAdjacentResponse(
|
|
id=next_result[0] if next_result else 0,
|
|
title=next_result[1] if next_result else ""
|
|
)
|
|
]
|
|
|
|
|
|
def add_blog_visit(db: Session, blog_visit: BlogVisit):
|
|
db.add(blog_visit)
|
|
db.commit()
|
|
db.refresh(blog_visit)
|
|
|
|
|
|
def query_blog_visit(db: Session, current_page: int = 1, page_size: int = 10) -> PageResult[BlogVisitResponse]:
|
|
query = select(
|
|
BlogVisit.ip,
|
|
BlogVisit.os,
|
|
BlogVisit.browser,
|
|
BlogVisit.uri,
|
|
Blog.title,
|
|
BlogVisit.create_time.label("visitTime")
|
|
).select_from(BlogVisit).outerjoin(
|
|
Blog, Blog.id == BlogVisit.blog_id
|
|
).order_by(desc(BlogVisit.create_time))
|
|
|
|
return paginate_query(db, query, current_page, page_size)
|
|
|
|
|
|
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_id: int, blog_comment: BlogCommentCreate) -> bool:
|
|
check_blog_exist(db, blog_id)
|
|
|
|
if blog_comment.parentId != 0:
|
|
parent_comment = db.execute(
|
|
select(BlogComment)
|
|
.where(BlogComment.id == blog_comment.parentId)
|
|
.where(BlogComment.blog_id == blog_id)
|
|
).scalar_one_or_none()
|
|
|
|
if not parent_comment:
|
|
raise AppException("父评论不存在")
|
|
|
|
db_comment = BlogComment(
|
|
blog_id=blog_id,
|
|
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
|
|
|
|
|
|
def check_blog_exist(db: Session, blog_id: int) -> Blog:
|
|
result = db.execute(select(Blog).where(Blog.id == blog_id)).scalar_one_or_none()
|
|
|
|
if not result:
|
|
raise AppException("博客不存在")
|
|
|
|
return result
|
|
|
|
|
|
def check_blog_content_exist(db: Session, blog_content_id: int) -> BlogContent:
|
|
result = db.execute(select(BlogContent).where(BlogContent.id == blog_content_id)).scalar_one_or_none()
|
|
if not result:
|
|
raise AppException("博客内容不存在")
|
|
|
|
return result
|