144 lines
4.4 KiB
Python
144 lines
4.4 KiB
Python
from datetime import datetime
|
|
from typing import List
|
|
|
|
from sqlalchemy import select, func, distinct, extract
|
|
from sqlalchemy.orm import Session
|
|
|
|
from models.blog import Blog, BlogVisit, BlogComment, BlogCategory
|
|
from schemas.blog_stats import BlogOverview, BlogChartStats
|
|
|
|
|
|
def query_blog_overview(db: Session) -> BlogOverview:
|
|
blog_count_stmt = select(func.count(Blog.id)).where(Blog.is_approved == 1)
|
|
blog_count = db.execute(blog_count_stmt).scalar() or 0
|
|
|
|
word_count_stmt = select(func.sum(Blog.word_count)).where(Blog.is_approved == 1)
|
|
word_count = db.execute(word_count_stmt).scalar() or 0
|
|
|
|
category_count_stmt = select(func.count(distinct(Blog.category_id))).where(Blog.is_approved == 1)
|
|
category_count = db.execute(category_count_stmt).scalar() or 0
|
|
|
|
great_count_stmt = select(func.count(Blog.id)).where(Blog.is_approved == 1, Blog.is_great == 1)
|
|
great_count = db.execute(great_count_stmt).scalar() or 0
|
|
|
|
visit_count_stmt = select(func.count(BlogVisit.id))
|
|
visit_count = db.execute(visit_count_stmt).scalar() or 0
|
|
|
|
return BlogOverview(
|
|
blogCount=blog_count,
|
|
categoryCount=category_count,
|
|
wordCount=word_count,
|
|
greatCount=great_count,
|
|
visitCount=visit_count
|
|
)
|
|
|
|
|
|
def query_blog_category(db: Session) -> List[BlogChartStats]:
|
|
query = (
|
|
select(
|
|
BlogCategory.name.label("name"),
|
|
func.count(Blog.id).label("value")
|
|
)
|
|
.where(Blog.is_approved == 1)
|
|
.outerjoin(BlogCategory, Blog.category_id == BlogCategory.id)
|
|
.group_by(BlogCategory.name))
|
|
|
|
results = db.execute(query).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|
|
|
|
|
|
def query_blog_approved_monthly(db: Session, year: int = None) -> List[BlogChartStats]:
|
|
current_year = datetime.now().year
|
|
target_year = year if year else current_year
|
|
|
|
# extract()：返回日期/时间的单独部分
|
|
stmt = (
|
|
select(
|
|
func.concat(extract('month', Blog.create_time), '月').label('name'),
|
|
func.count(Blog.id).label('value')
|
|
)
|
|
.where(
|
|
Blog.is_approved == 1,
|
|
extract('year', Blog.create_time) == target_year
|
|
)
|
|
.group_by('name')
|
|
)
|
|
|
|
results = db.execute(stmt).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|
|
|
|
|
|
def query_blog_visit_monthly(db: Session, year: int = None) -> List[BlogChartStats]:
|
|
current_year = datetime.now().year
|
|
target_year = year if year else current_year
|
|
|
|
stmt = (
|
|
select(
|
|
func.concat(extract('month', BlogVisit.create_time), '月').label('name'),
|
|
func.count(BlogVisit.id).label('value')
|
|
)
|
|
.where(
|
|
extract('year', BlogVisit.create_time) == target_year
|
|
)
|
|
.group_by('name')
|
|
)
|
|
|
|
results = db.execute(stmt).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|
|
|
|
|
|
def query_blog_visit_rank(db: Session, limit: int = 5) -> List[BlogChartStats]:
|
|
stmt = (
|
|
select(
|
|
Blog.title.label('name'),
|
|
func.count(BlogVisit.id).label('value')
|
|
)
|
|
.join(BlogVisit, Blog.id == BlogVisit.blog_id)
|
|
.where(Blog.is_approved == 1)
|
|
.group_by(Blog.title)
|
|
.order_by(func.count(BlogVisit.id).desc())
|
|
.limit(limit)
|
|
)
|
|
|
|
results = db.execute(stmt).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|
|
|
|
|
|
def query_blog_read_rank(db: Session, limit: int = 5) -> List[BlogChartStats]:
|
|
stmt = (
|
|
select(
|
|
Blog.title.label('name'),
|
|
func.round(func.sum(Blog.read_duration)).label('value')
|
|
)
|
|
.where(Blog.is_approved == 1)
|
|
.group_by(Blog.title)
|
|
.order_by(func.sum(Blog.read_duration).desc())
|
|
.limit(limit)
|
|
)
|
|
|
|
results = db.execute(stmt).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|
|
|
|
|
|
def query_blog_comment_rank(db: Session, limit: int = 5) -> List[BlogChartStats]:
|
|
stmt = (
|
|
select(
|
|
Blog.title.label('name'),
|
|
func.count(BlogComment.id).label('value')
|
|
)
|
|
.join(BlogComment, Blog.id == BlogComment.blog_id)
|
|
.where(Blog.is_approved == 1)
|
|
.group_by(Blog.title)
|
|
.order_by(func.count(BlogComment.id).desc())
|
|
.limit(limit)
|
|
)
|
|
|
|
results = db.execute(stmt).fetchall()
|
|
|
|
return [BlogChartStats.model_validate(result) for result in results]
|