feat:增加博客统计接口

service/blog_stats_service.py

@@ -0,0 +1,143 @@
+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]