feat:清除gitignore文件
This commit is contained in:
Binary file not shown.
Binary file not shown.
@@ -1,326 +0,0 @@
|
||||
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, BlogVisitResponse, BlogCreate, BlogUpdate
|
||||
from schemas.pagination import PageResult
|
||||
from schemas.paginate_query import paginate_query
|
||||
from middleware.exceptions import AppException
|
||||
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:
|
||||
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 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)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def add_blog_category(db: Session, category: str) -> int:
|
||||
db_blog_category = db.query(BlogCategory).filter(BlogCategory.name == category).first()
|
||||
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 = db.query(Blog).filter(Blog.id == blog_id).first()
|
||||
if not db_blog:
|
||||
raise AppException("博客不存在")
|
||||
|
||||
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.query(BlogCategory).filter(BlogCategory.name == category).first()
|
||||
|
||||
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 = db.query(BlogContent).filter(BlogContent.id == blog_content_id).first()
|
||||
if not db_blog_content:
|
||||
raise AppException("博客内容不存在")
|
||||
|
||||
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 = db.query(Blog).filter(Blog.id == blog_id).first()
|
||||
if not db_blog:
|
||||
raise AppException("博客不存在")
|
||||
|
||||
delete_blog_content(db, db_blog.content_id)
|
||||
|
||||
db.delete(db_blog)
|
||||
db.commit()
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def delete_blog_content(db: Session, blog_content_id: int) -> bool:
|
||||
db_blog_content = db.query(BlogContent).filter(BlogContent.id == blog_content_id).first()
|
||||
if not db_blog_content:
|
||||
raise AppException("博客内容不存在")
|
||||
|
||||
db.delete(db_blog_content)
|
||||
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 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_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
|
||||
@@ -1,23 +0,0 @@
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from schemas.session import SessionResponse
|
||||
from config.auth import create_token, create_payload
|
||||
from middleware.exceptions import AppException
|
||||
|
||||
|
||||
def create(db: Session, username: str, password: str) -> SessionResponse:
|
||||
# 1. 校验账户是否存在
|
||||
if username != "admin":
|
||||
raise AppException(f"该账户{username}不存在")
|
||||
|
||||
# 2. 校验密码是否正确
|
||||
if password != "19940822Cxx":
|
||||
raise AppException("密码错误")
|
||||
|
||||
return SessionResponse(
|
||||
access_token=create_token(create_payload(username)),
|
||||
token_type="Bearer"
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user