Dev stat

src/backend/app/api/v1/endpoints/statistics.py

@@ -0,0 +1,92 @@
+from typing import List, Optional
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.engine.base import Connection
+
+from backend.app.dependencies.auth_deps import get_current_active_user, get_current_admin
+from backend.app.services.statistics_service import StatisticsService
+from backend.app.dependencies.service_deps import get_statistics_service
+from backend.app.dependencies.db_deps import get_db_connection
+from backend.app.schemas.statistics_schema import (
+    SystemStatistics, 
+    AdminDashboardStatistics,
+    StoreStatistics
+)
+
+router = APIRouter()
+
+
+@router.get("/system", response_model=SystemStatistics)
+async def get_system_statistics(
+    statistics_service: StatisticsService = Depends(get_statistics_service),
+    conn: Connection = Depends(get_db_connection),
+    _=Depends(get_current_admin)
+):
+    """
+    Get overall system statistics (admin only).
+    """
+    return await statistics_service.get_system_statistics(conn=conn)
+
+
+@router.get("/admin-dashboard", response_model=AdminDashboardStatistics)
+async def get_admin_dashboard_statistics(
+    statistics_service: StatisticsService = Depends(get_statistics_service),
+    conn: Connection = Depends(get_db_connection),
+    _=Depends(get_current_admin)
+):
+    """
+    Get admin dashboard statistics (admin only).
+    """
+    return await statistics_service.get_admin_dashboard_statistics(conn=conn)
+
+
+@router.get("/store", response_model=List[StoreStatistics])
+async def get_all_store_statistics(
+    statistics_service: StatisticsService = Depends(get_statistics_service),
+    conn: Connection = Depends(get_db_connection),
+    _=Depends(get_current_admin)
+):
+    """
+    Get statistics for all stores (admin only).
+    """
+    return await statistics_service.get_store_statistics(conn=conn)
+
+
+@router.get("/store/{store_id}", response_model=StoreStatistics)
+async def get_specific_store_statistics(
+    store_id: int,
+    statistics_service: StatisticsService = Depends(get_statistics_service),
+    conn: Connection = Depends(get_db_connection),
+    current_user = Depends(get_current_active_user)
+):
+    """
+    Get statistics for a specific store.
+    Admin users can access any store.
+    Merchant users can only access their own store.
+    """
+    # Check if user is merchant and has access to this store
+    if current_user.UserRole.lower() == "merchant":
+        # Check if user has a store_id attribute
+        if not hasattr(current_user, 'StoreID') or current_user.StoreID is None:
+            raise HTTPException(
+                status_code=403,
+                detail="Merchant without a store cannot access statistics"
+            )
+
+        # Direct comparison of merchant's StoreID with the requested store_id
+        if current_user.StoreID != store_id:
+            raise HTTPException(
+                status_code=403, 
+                detail="You don't have permission to access this store's statistics"
+            )
+    # Non-admin, non-merchant users cannot access store statistics
+    elif current_user.UserRole.lower() != "admin":
+        raise HTTPException(
+            status_code=403,
+            detail="Only admin and merchant users can access store statistics"
+        )
+
+    store_stats = await statistics_service.get_store_statistics(conn=conn, store_id=store_id)
+    if not store_stats:
+        raise HTTPException(status_code=404, detail="Store not found")
+
+    return store_stats[0]

src/backend/app/api/v1/router.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 from .endpoints import user, product, category, auth, cart, address, order, payment, store,\
-    store_change_request, product_change_request, product_change_request_v2, store_change_request_v2
+    store_change_request, product_change_request, product_change_request_v2, store_change_request_v2, statistics
 
 api_router_v1 = APIRouter()
 
@@ -20,3 +20,4 @@
 
 
 api_router_v1.include_router(product_change_request_v2.router, prefix="/product-change-new", tags=["Product Change Requests V2"])
+api_router_v1.include_router(statistics.router, prefix="/statistics", tags=["Statistics"])

src/backend/app/dependencies/db_deps.py

@@ -1,5 +1,5 @@
 from typing import Generator, Callable, Optional
-from sqlalchemy import Connection
+from sqlalchemy.engine.base import Connection
 from fastapi import Depends
 
 from backend.app.core.database import get_engine

src/backend/app/dependencies/service_deps.py

@@ -24,7 +24,7 @@
     ProductChangeRequestService,
     StoreService,
     ProductChangeRequestService2,
-    StoreChangeRequestService2,
+    StoreChangeRequestService2, StatisticsService
 )
 from backend.app.dependencies.crud_deps import *
 from backend.app.crud.store_change_request_crud import get_store_change_request_crud_instance
@@ -232,3 +232,11 @@ def get_store_change_request_service_v2(
         store_crud=store_crud,
         user_crud=user_crud,
     )
+
+# dependency injection for StatisticsService
+def get_statistics_service() -> StatisticsService:
+    """
+    Dependency to get the StatisticsService instance.
+    :return: The StatisticsService instance.
+    """
+    return StatisticsService()

src/backend/app/schemas/statistics_schema.py

@@ -0,0 +1,27 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class SystemStatistics(BaseModel):
+    total_users: int
+    total_products: int
+    total_stores: int
+    total_categories: int
+    pending_orders: int
+    completed_orders: int
+    total_orders: int
+    total_sales: Optional[float] = None
+
+
+class AdminDashboardStatistics(SystemStatistics):
+    """Admin dashboard statistics extends system statistics."""
+    pass
+
+
+class StoreStatistics(BaseModel):
+    store_id: int
+    store_name: str
+    product_count: int
+    order_count: int
+    items_sold: Optional[int] = None
+    total_revenue: Optional[float] = None

src/backend/app/services/__init__.py

@@ -10,3 +10,4 @@
 
 from .product_change_request_service_v2 import ProductChangeRequestService2
 from .store_change_request_service_v2 import StoreChangeRequestService2
+from .statistics_service import StatisticsService

src/backend/app/services/statistics_service.py

@@ -0,0 +1,83 @@
+from typing import List, Optional
+from sqlalchemy import text
+from sqlalchemy.engine.base import Connection
+import logging
+
+from backend.app.schemas.statistics_schema import SystemStatistics, AdminDashboardStatistics, StoreStatistics
+
+logger = logging.getLogger(__name__)
+
+
+class StatisticsService:
+    """
+    统计服务类，负责处理所有与系统统计相关的业务逻辑。
+    """
+
+    def __init__(self):
+        pass
+
+    async def get_system_statistics(self, conn: Connection) -> SystemStatistics:
+        """
+        Get overall system statistics from the system_statistics view.
+        """
+        query = text("SELECT * FROM system_statistics")
+        result = conn.execute(query)
+        row = result.fetchone()
+        if not row:
+            logger.warning("No system statistics found")
+            return SystemStatistics(
+                total_users=0,
+                total_products=0,
+                total_stores=0,
+                total_categories=0,
+                pending_orders=0,
+                completed_orders=0,
+                total_orders=0,
+                total_sales=0
+            )
+        # Use row._mapping instead of dict(row) for newer SQLAlchemy versions
+        stats_dict = row._mapping
+        return SystemStatistics(**stats_dict)
+
+    async def get_admin_dashboard_statistics(self, conn: Connection) -> AdminDashboardStatistics:
+        """
+        Get admin dashboard statistics from the admin_dashboard_statistics view.
+        """
+        query = text("SELECT * FROM admin_dashboard_statistics")
+        result = conn.execute(query)
+        row = result.fetchone()
+        if not row:
+            logger.warning("No admin dashboard statistics found")
+            return AdminDashboardStatistics(
+                total_users=0,
+                total_products=0,
+                total_stores=0,
+                total_categories=0,
+                pending_orders=0,
+                completed_orders=0,
+                total_orders=0,
+                total_sales=0
+            )
+        # Use row._mapping instead of dict(row) for newer SQLAlchemy versions
+        stats_dict = row._mapping
+        return AdminDashboardStatistics(**stats_dict)
+
+    async def get_store_statistics(self, conn: Connection, store_id: Optional[int] = None) -> List[StoreStatistics]:
+        """
+        Get store statistics from the store_statistics view.
+        If store_id is provided, returns statistics for that store only.
+        """
+        if store_id:
+            query = text("SELECT * FROM store_statistics WHERE store_id = :store_id")
+            result = conn.execute(query, {"store_id": store_id})
+        else:
+            query = text("SELECT * FROM store_statistics")
+            result = conn.execute(query)
+
+        rows = result.fetchall()
+        if not rows:
+            logger.warning(f"No store statistics found{' for store_id: ' + str(store_id) if store_id else ''}")
+            return []
+
+        # Use row._mapping instead of dict(row) for newer SQLAlchemy versions
+        return [StoreStatistics(**row._mapping) for row in rows]

src/backend/sql/ddl/014_create_system_statistics_view.sql

@@ -0,0 +1,34 @@
+-- Active: 1748168581480@@127.0.0.1@3306@projtest
+-- 创建系统统计视图
+CREATE OR REPLACE VIEW system_statistics AS
+SELECT
+    (SELECT COUNT(*) FROM User) AS total_users,
+    (SELECT COUNT(*) FROM Product) AS total_products,
+    (SELECT COUNT(*) FROM Store) AS total_stores,
+    (SELECT COUNT(*) FROM ProductCategory) AS total_categories,
+    (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'PROCESSING_BY_MERCHANT') AS pending_orders,
+    (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS completed_orders,
+    (SELECT COUNT(*) FROM `Order`) AS total_orders,
+    (SELECT SUM(FinalAmountForThisOrder) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS total_sales
+;
+
+-- 为不同角色创建更细化的统计视图
+CREATE OR REPLACE VIEW admin_dashboard_statistics AS
+SELECT * FROM system_statistics;
+
+-- 商店统计视图（按商店分组）
+CREATE OR REPLACE VIEW store_statistics AS
+SELECT 
+    s.StoreID AS store_id,
+    s.StoreName AS store_name,
+    COUNT(p.ProductID) AS product_count,
+    COUNT(DISTINCT o.OrderID) AS order_count,
+    SUM(oi.Quantity) AS items_sold,
+    SUM(oi.PriceAtPurchase * oi.Quantity) AS total_revenue
+FROM 
+    Store s
+    LEFT JOIN Product p ON s.StoreID = p.StoreID
+    LEFT JOIN OrderItem oi ON p.ProductID = oi.ProductID
+    LEFT JOIN `Order` o ON oi.OrderID = o.OrderID AND o.OrderStatus = 'COMPLETED'
+GROUP BY 
+    s.StoreID, s.StoreName;

src/backend/test/integration/api_endpoints/mock_statistics_views.py

@@ -0,0 +1,56 @@
+"""
+Create mock views for statistics tests
+
+This module provides functions to create temporary mock views for the statistics tests.
+"""
+from sqlalchemy import text
+from sqlalchemy.engine.base import Connection
+from backend.app.utils import logger
+
+def create_mock_views(conn: Connection):
+    """Create mock views for the statistics tests."""
+    logger.info("Creating mock views for statistics tests")
+
+    try:
+        # Create system statistics view
+        conn.execute(text("""
+        CREATE OR REPLACE VIEW system_statistics AS
+        SELECT
+            (SELECT COUNT(*) FROM User) AS total_users,
+            (SELECT COUNT(*) FROM Product) AS total_products,
+            (SELECT COUNT(*) FROM Store) AS total_stores,
+            (SELECT COUNT(*) FROM ProductCategory) AS total_categories,
+            (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'PROCESSING_BY_MERCHANT') AS pending_orders,
+            (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS completed_orders,
+            (SELECT COUNT(*) FROM `Order`) AS total_orders,
+            (SELECT COALESCE(SUM(FinalAmountForThisOrder), 0) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS total_sales
+        """))
+
+        # Create admin dashboard statistics view
+        conn.execute(text("""
+        CREATE OR REPLACE VIEW admin_dashboard_statistics AS
+        SELECT * FROM system_statistics
+        """))
+
+        # Create store statistics view
+        conn.execute(text("""
+        CREATE OR REPLACE VIEW store_statistics AS
+        SELECT 
+            s.StoreID AS store_id,
+            s.StoreName AS store_name,
+            COUNT(DISTINCT p.ProductID) AS product_count,
+            (SELECT COUNT(*) FROM `Order` WHERE StoreID = s.StoreID) AS order_count,
+            COALESCE(SUM(oi.Quantity), 0) AS items_sold,
+            COALESCE(SUM(oi.Subtotal), 0) AS total_revenue
+        FROM 
+            Store s
+            LEFT JOIN Product p ON s.StoreID = p.StoreID
+            LEFT JOIN OrderItem oi ON oi.StoreID = s.StoreID
+        GROUP BY 
+            s.StoreID, s.StoreName
+        """))
+
+        logger.info("Mock views created successfully")
+    except Exception as e:
+        logger.error(f"Error creating mock views: {e}")
+        raise