sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stats management revamping

Browse Source
This commit is contained in:
Samir Boulahtit
2025-10-25 07:25:36 +02:00
parent 5be47b91a2
commit 3631766d28
5 changed files with 189 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/api/v1/admin/marketplace.py
View File

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_service import admin_service
from app.services.stats_service import stats_service
from models.schema.marketplace_import_job import MarketplaceImportJobResponse
from models.database.user import User
@@ -46,4 +47,4 @@ def get_import_statistics(
current_admin: User = Depends(get_current_admin_user),
):
"""Get marketplace import statistics (Admin only)."""
return admin_service.get_import_statistics(db)
return stats_service.get_import_statistics(db)

3
app/api/v1/admin/users.py
View File

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_service import admin_service
from app.services.stats_service import stats_service
from models.schema.auth import UserResponse
from models.database.user import User
@@ -48,4 +49,4 @@ def get_user_statistics(
current_admin: User = Depends(get_current_admin_user),
):
"""Get user statistics for admin dashboard (Admin only)."""
return admin_service.get_user_statistics(db)
return stats_service.get_user_statistics(db)

24
app/api/v1/admin/vendors.py
View File

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_service import admin_service
from models.schema.stats import VendorStatsResponse
from models.schema.vendor import (
VendorListResponse,
VendorResponse,
@@ -106,6 +107,21 @@ def get_all_vendors_admin(
return VendorListResponse(vendors=vendors, total=total, skip=skip, limit=limit)
@router.get("/stats", response_model=VendorStatsResponse)
def get_vendor_statistics(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Get vendor statistics for admin dashboard (Admin only)."""
stats = get_vendor_statistics(db)
return VendorStatsResponse(
total=stats["total_vendors"],
verified=stats["verified_vendors"],
pending=stats["total_vendors"] - stats["verified_vendors"],
inactive=stats["inactive_vendors"],
)
@router.get("/{vendor_id}", response_model=VendorDetailResponse)
def get_vendor_details(
vendor_id: int,
@@ -296,11 +312,3 @@ def delete_vendor(
message = admin_service.delete_vendor(db, vendor_id)
return {"message": message}
@router.get("/stats/vendors")
def get_vendor_statistics(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Get vendor statistics for admin dashboard (Admin only)."""
return admin_service.get_vendor_statistics(db)

83
app/services/admin_service.py
View File

@@ -621,50 +621,6 @@ class AdminService:
# STATISTICS
# ============================================================================
def get_user_statistics(self, db: Session) -> dict:
"""Get user statistics for admin dashboard."""
try:
total_users = db.query(User).count()
active_users = db.query(User).filter(User.is_active == True).count()
inactive_users = total_users - active_users
admin_users = db.query(User).filter(User.role == "admin").count()
return {
"total_users": total_users,
"active_users": active_users,
"inactive_users": inactive_users,
"admin_users": admin_users,
"activation_rate": (active_users / total_users * 100) if total_users > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get user statistics: {str(e)}")
raise AdminOperationException(
operation="get_user_statistics",
reason="Database query failed"
)
def get_vendor_statistics(self, db: Session) -> dict:
"""Get vendor statistics for admin dashboard."""
try:
total_vendors = db.query(Vendor).count()
active_vendors = db.query(Vendor).filter(Vendor.is_active == True).count()
verified_vendors = db.query(Vendor).filter(Vendor.is_verified == True).count()
inactive_vendors = total_vendors - active_vendors
return {
"total_vendors": total_vendors,
"active_vendors": active_vendors,
"inactive_vendors": inactive_vendors,
"verified_vendors": verified_vendors,
"verification_rate": (verified_vendors / total_vendors * 100) if total_vendors > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get vendor statistics: {str(e)}")
raise AdminOperationException(
operation="get_vendor_statistics",
reason="Database query failed"
)
def get_recent_vendors(self, db: Session, limit: int = 5) -> List[dict]:
"""Get recently created vendors."""
try:
@@ -716,45 +672,6 @@ class AdminService:
logger.error(f"Failed to get recent import jobs: {str(e)}")
return []
def get_product_statistics(self, db: Session) -> dict:
"""Get product statistics."""
# TODO: Implement when Product model is available
return {
"total_products": 0,
"active_products": 0,
"out_of_stock": 0
}
def get_order_statistics(self, db: Session) -> dict:
"""Get order statistics."""
# TODO: Implement when Order model is available
return {
"total_orders": 0,
"pending_orders": 0,
"completed_orders": 0
}
def get_import_statistics(self, db: Session) -> dict:
"""Get import job statistics."""
try:
total = db.query(MarketplaceImportJob).count()
completed = db.query(MarketplaceImportJob).filter(
MarketplaceImportJob.status == "completed"
).count()
failed = db.query(MarketplaceImportJob).filter(
MarketplaceImportJob.status == "failed"
).count()
return {
"total_imports": total,
"completed_imports": completed,
"failed_imports": failed,
"success_rate": (completed / total * 100) if total > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get import statistics: {str(e)}")
return {"total_imports": 0, "completed_imports": 0, "failed_imports": 0, "success_rate": 0}
# ============================================================================
# PRIVATE HELPER METHODS
# ============================================================================

174
app/services/stats_service.py
View File

@@ -20,11 +20,13 @@ from app.exceptions import (
VendorNotFoundException,
AdminOperationException,
)
from models.database.marketplace_product import MarketplaceProduct
from models.database.product import Product
from models.database.inventory import Inventory
from models.database.vendor import Vendor
from models.database.order import Order
from models.database.user import User
from models.database.customer import Customer
from models.database.marketplace_import_job import MarketplaceImportJob
@@ -158,7 +160,7 @@ class StatsService:
self, db: Session, vendor_id: int, period: str = "30d"
) -> Dict[str, Any]:
"""
Get vendor analytics for a time period.
Get a specific vendor analytics for a time period.
Args:
db: Database session
@@ -224,6 +226,28 @@ class StatsService:
target_id=str(vendor_id)
)
def get_vendor_statistics(self, db: Session) -> dict:
"""Get vendor statistics for admin dashboard."""
try:
total_vendors = db.query(Vendor).count()
active_vendors = db.query(Vendor).filter(Vendor.is_active == True).count()
verified_vendors = db.query(Vendor).filter(Vendor.is_verified == True).count()
inactive_vendors = total_vendors - active_vendors
return {
"total_vendors": total_vendors,
"active_vendors": active_vendors,
"inactive_vendors": inactive_vendors,
"verified_vendors": verified_vendors,
"verification_rate": (verified_vendors / total_vendors * 100) if total_vendors > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get vendor statistics: {str(e)}")
raise AdminOperationException(
operation="get_vendor_statistics",
reason="Database query failed"
)
# ========================================================================
# SYSTEM-WIDE STATISTICS (ADMIN)
# ========================================================================
@@ -321,12 +345,128 @@ class StatsService:
reason=f"Database query failed: {str(e)}"
)
def get_user_statistics(self, db: Session) -> Dict[str, Any]:
"""
Get user statistics for admin dashboard.
Args:
db: Database session
Returns:
Dictionary with user statistics
Raises:
AdminOperationException: If database query fails
"""
try:
total_users = db.query(User).count()
active_users = db.query(User).filter(User.is_active == True).count()
inactive_users = total_users - active_users
admin_users = db.query(User).filter(User.role == "admin").count()
return {
"total_users": total_users,
"active_users": active_users,
"inactive_users": inactive_users,
"admin_users": admin_users,
"activation_rate": (active_users / total_users * 100) if total_users > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get user statistics: {str(e)}")
raise AdminOperationException(
operation="get_user_statistics",
reason="Database query failed"
)
def get_import_statistics(self, db: Session) -> Dict[str, Any]:
"""
Get import job statistics.
Args:
db: Database session
Returns:
Dictionary with import statistics
Raises:
AdminOperationException: If database query fails
"""
try:
total = db.query(MarketplaceImportJob).count()
completed = db.query(MarketplaceImportJob).filter(
MarketplaceImportJob.status == "completed"
).count()
failed = db.query(MarketplaceImportJob).filter(
MarketplaceImportJob.status == "failed"
).count()
return {
"total_imports": total,
"completed_imports": completed,
"failed_imports": failed,
"success_rate": (completed / total * 100) if total > 0 else 0
}
except Exception as e:
logger.error(f"Failed to get import statistics: {str(e)}")
return {
"total_imports": 0,
"completed_imports": 0,
"failed_imports": 0,
"success_rate": 0
}
def get_order_statistics(self, db: Session) -> Dict[str, Any]:
"""
Get order statistics.
Args:
db: Database session
Returns:
Dictionary with order statistics
Note:
TODO: Implement when Order model is fully available
"""
return {
"total_orders": 0,
"pending_orders": 0,
"completed_orders": 0
}
def get_product_statistics(self, db: Session) -> Dict[str, Any]:
"""
Get product statistics.
Args:
db: Database session
Returns:
Dictionary with product statistics
Note:
TODO: Implement when Product model is fully available
"""
return {
"total_products": 0,
"active_products": 0,
"out_of_stock": 0
}
# ========================================================================
# PRIVATE HELPER METHODS
# ========================================================================
def _parse_period(self, period: str) -> int:
"""Parse period string to days."""
"""
Parse period string to days.
Args:
period: Period string (7d, 30d, 90d, 1y)
Returns:
Number of days
"""
period_map = {
"7d": 7,
"30d": 30,
@@ -336,7 +476,15 @@ class StatsService:
return period_map.get(period, 30)
def _get_unique_brands_count(self, db: Session) -> int:
"""Get count of unique brands."""
"""
Get count of unique brands.
Args:
db: Database session
Returns:
Count of unique brands
"""
return (
db.query(MarketplaceProduct.brand)
.filter(
@@ -348,7 +496,15 @@ class StatsService:
)
def _get_unique_categories_count(self, db: Session) -> int:
"""Get count of unique categories."""
"""
Get count of unique categories.
Args:
db: Database session
Returns:
Count of unique categories
"""
return (
db.query(MarketplaceProduct.google_product_category)
.filter(
@@ -360,7 +516,15 @@ class StatsService:
)
def _get_inventory_statistics(self, db: Session) -> Dict[str, int]:
"""Get inventory-related statistics."""
"""
Get inventory-related statistics.
Args:
db: Database session
Returns:
Dictionary with inventory statistics
"""
total_entries = db.query(Inventory).count()
total_quantity = db.query(func.sum(Inventory.quantity)).scalar() or 0
total_reserved = db.query(func.sum(Inventory.reserved_quantity)).scalar() or 0