295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""
|
||
@service StorefrontCmsService (Backend/app/services/StorefrontCmsService.py)
|
||
@purpose Business logic for reading and updating Storefront CMS config —
|
||
Footer Info, Store Settings, Mega Menu, Catalog Filters, and Review moderation.
|
||
"""
|
||
from __future__ import annotations
|
||
from typing import Any, Dict, List, Optional
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import select
|
||
|
||
from app.models.StorefrontCmsModel import (
|
||
StorefrontSettings,
|
||
StorefrontMegaMenu,
|
||
StorefrontCatalogFilters,
|
||
StorefrontFooterInfo,
|
||
)
|
||
from app.models.CategoryModel import Category
|
||
from app.models.ProductReviewModel import ProductReview
|
||
from app.schemas.StorefrontCmsSchema import (
|
||
FooterInfoResponse,
|
||
FooterInfoUpdate,
|
||
StorefrontSettingsResponse,
|
||
StorefrontSettingsUpdate,
|
||
MegaMenuResponse,
|
||
MegaMenuUpdate,
|
||
AllMegaMenusResponse,
|
||
CatalogFiltersResponse,
|
||
CatalogFiltersUpdate,
|
||
CategoryCmsUpdate,
|
||
ReviewModerationResponse,
|
||
ReviewApprovalPayload,
|
||
)
|
||
|
||
|
||
class StorefrontCmsService:
|
||
def __init__(self, db: Session):
|
||
self.db = db
|
||
|
||
# ------------------------------------------------------------------
|
||
# Helper: safe JSON list
|
||
# ------------------------------------------------------------------
|
||
def _safe_list(self, v: Any) -> List:
|
||
return v if isinstance(v, list) else []
|
||
|
||
# ------------------------------------------------------------------
|
||
# FOOTER INFO
|
||
# ------------------------------------------------------------------
|
||
def get_footer_info(self) -> FooterInfoResponse:
|
||
row = self.db.get(StorefrontFooterInfo, "default")
|
||
if not row:
|
||
return FooterInfoResponse()
|
||
return FooterInfoResponse(
|
||
phone=row.phone,
|
||
email=row.email,
|
||
address=row.address,
|
||
copyright=row.copyright,
|
||
social_links=self._safe_list(row.social_links),
|
||
columns=self._safe_list(row.columns),
|
||
payment_methods=self._safe_list(row.payment_methods),
|
||
)
|
||
|
||
def update_footer_info(self, payload: FooterInfoUpdate, updated_by: str = "admin") -> FooterInfoResponse:
|
||
row = self.db.get(StorefrontFooterInfo, "default")
|
||
if not row:
|
||
row = StorefrontFooterInfo(id="default")
|
||
self.db.add(row)
|
||
|
||
if payload.phone is not None:
|
||
row.phone = payload.phone
|
||
if payload.email is not None:
|
||
row.email = payload.email
|
||
if payload.address is not None:
|
||
row.address = payload.address
|
||
if payload.copyright is not None:
|
||
row.copyright = payload.copyright
|
||
if payload.social_links is not None:
|
||
row.social_links = [s.model_dump() for s in payload.social_links]
|
||
if payload.columns is not None:
|
||
row.columns = [c.model_dump() for c in payload.columns]
|
||
if payload.payment_methods is not None:
|
||
row.payment_methods = [p.model_dump() for p in payload.payment_methods]
|
||
|
||
self.db.commit()
|
||
self.db.refresh(row)
|
||
return self.get_footer_info()
|
||
|
||
# ------------------------------------------------------------------
|
||
# STOREFRONT SETTINGS (flat key→value map)
|
||
# ------------------------------------------------------------------
|
||
_KNOWN_KEYS = [
|
||
"store_name", "logo_url", "primary_wordmark_url", "secondary_wordmark_url",
|
||
"favicon_url", "support_phone", "currency_code", "advance_percent", "theme_color",
|
||
"size_chart",
|
||
]
|
||
|
||
def get_settings(self) -> Dict[str, Any]:
|
||
rows = self.db.execute(select(StorefrontSettings)).scalars().all()
|
||
result: Dict[str, Any] = {
|
||
# Provide defaults for required CMS keys
|
||
"store_name": "iFixKart",
|
||
"logo_url": None,
|
||
"primary_wordmark_url": None,
|
||
"secondary_wordmark_url": None,
|
||
"favicon_url": None,
|
||
"support_phone": None,
|
||
"currency_code": "INR",
|
||
"advance_percent": 20.0,
|
||
"theme_color": None,
|
||
"size_chart": [], # list of {size, width, height, depth} rows
|
||
}
|
||
for r in rows:
|
||
result[r.key] = r.value
|
||
return result
|
||
|
||
def update_settings(self, payload: StorefrontSettingsUpdate, updated_by: str = "admin") -> Dict[str, Any]:
|
||
data = payload.model_dump(exclude_none=True)
|
||
for key, value in data.items():
|
||
row = self.db.get(StorefrontSettings, key)
|
||
if row:
|
||
row.value = value
|
||
row.updated_by = updated_by
|
||
else:
|
||
row = StorefrontSettings(key=key, value=value, updated_by=updated_by)
|
||
self.db.add(row)
|
||
self.db.commit()
|
||
return self.get_settings()
|
||
|
||
# ------------------------------------------------------------------
|
||
# MEGA MENU
|
||
# ------------------------------------------------------------------
|
||
def get_mega_menu(self, nav_key: Optional[str] = None) -> Any:
|
||
if nav_key:
|
||
row = self.db.get(StorefrontMegaMenu, nav_key)
|
||
if not row:
|
||
return MegaMenuResponse(nav_key=nav_key)
|
||
return MegaMenuResponse(
|
||
nav_key=row.nav_key,
|
||
groups=self._safe_list(row.groups),
|
||
promo=row.promo,
|
||
featured_category_ids=self._safe_list(row.featured_category_ids),
|
||
)
|
||
|
||
# Return all three menus in one payload
|
||
keys = ["shop", "deals", "products"]
|
||
rows = {r.nav_key: r for r in self.db.execute(
|
||
select(StorefrontMegaMenu).where(StorefrontMegaMenu.nav_key.in_(keys))
|
||
).scalars().all()}
|
||
|
||
def _build(k: str) -> MegaMenuResponse:
|
||
r = rows.get(k)
|
||
if not r:
|
||
return MegaMenuResponse(nav_key=k)
|
||
return MegaMenuResponse(
|
||
nav_key=r.nav_key,
|
||
groups=self._safe_list(r.groups),
|
||
promo=r.promo,
|
||
featured_category_ids=self._safe_list(r.featured_category_ids),
|
||
)
|
||
|
||
return AllMegaMenusResponse(shop=_build("shop"), deals=_build("deals"), products=_build("products"))
|
||
|
||
def update_mega_menu(self, payload: MegaMenuUpdate, updated_by: str = "admin") -> MegaMenuResponse:
|
||
row = self.db.get(StorefrontMegaMenu, payload.nav_key)
|
||
if not row:
|
||
row = StorefrontMegaMenu(nav_key=payload.nav_key)
|
||
self.db.add(row)
|
||
|
||
if payload.groups is not None:
|
||
row.groups = [g.model_dump() for g in payload.groups]
|
||
if payload.promo is not None:
|
||
row.promo = payload.promo.model_dump()
|
||
if payload.featured_category_ids is not None:
|
||
row.featured_category_ids = payload.featured_category_ids
|
||
|
||
self.db.commit()
|
||
self.db.refresh(row)
|
||
return self.get_mega_menu(payload.nav_key)
|
||
|
||
# ------------------------------------------------------------------
|
||
# CATALOG FILTERS
|
||
# ------------------------------------------------------------------
|
||
_DEFAULT_HIGHLIGHTS = [
|
||
{"label": "Sale", "value": "sale", "icon": "tag"},
|
||
{"label": "Trending", "value": "trending", "icon": "trending-up"},
|
||
{"label": "New Arrivals", "value": "new", "icon": "sparkles"},
|
||
]
|
||
_DEFAULT_PRICE_RANGES = [
|
||
{"label": "Under ₹500", "min": None, "max": 500},
|
||
{"label": "₹500 – ₹1,000", "min": 500, "max": 1000},
|
||
{"label": "₹1,000 – ₹5,000", "min": 1000, "max": 5000},
|
||
{"label": "₹5,000 – ₹15,000", "min": 5000, "max": 15000},
|
||
{"label": "Above ₹15,000", "min": 15000, "max": None},
|
||
]
|
||
|
||
def get_catalog_filters(self) -> CatalogFiltersResponse:
|
||
row = self.db.get(StorefrontCatalogFilters, "default")
|
||
if not row:
|
||
return CatalogFiltersResponse(
|
||
highlights=self._DEFAULT_HIGHLIGHTS,
|
||
price_ranges=self._DEFAULT_PRICE_RANGES,
|
||
)
|
||
return CatalogFiltersResponse(
|
||
highlights=self._safe_list(row.highlights) or self._DEFAULT_HIGHLIGHTS,
|
||
price_ranges=self._safe_list(row.price_ranges) or self._DEFAULT_PRICE_RANGES,
|
||
)
|
||
|
||
def update_catalog_filters(self, payload: CatalogFiltersUpdate) -> CatalogFiltersResponse:
|
||
row = self.db.get(StorefrontCatalogFilters, "default")
|
||
if not row:
|
||
row = StorefrontCatalogFilters(filter_key="default")
|
||
self.db.add(row)
|
||
|
||
if payload.highlights is not None:
|
||
row.highlights = [h.model_dump() for h in payload.highlights]
|
||
if payload.price_ranges is not None:
|
||
row.price_ranges = [p.model_dump() for p in payload.price_ranges]
|
||
|
||
self.db.commit()
|
||
self.db.refresh(row)
|
||
return self.get_catalog_filters()
|
||
|
||
# ------------------------------------------------------------------
|
||
# CATEGORY CMS UPDATE
|
||
# ------------------------------------------------------------------
|
||
def update_category_cms(self, category_id: str, payload: CategoryCmsUpdate) -> Dict[str, Any]:
|
||
cat = self.db.get(Category, category_id)
|
||
if not cat:
|
||
from fastapi import HTTPException
|
||
raise HTTPException(status_code=404, detail="Category not found")
|
||
|
||
for field, value in payload.model_dump(exclude_none=True).items():
|
||
setattr(cat, field, value)
|
||
|
||
self.db.commit()
|
||
self.db.refresh(cat)
|
||
return {
|
||
"category_id": cat.category_id,
|
||
"name": cat.name,
|
||
"slug": cat.slug,
|
||
"show_in_sidebar": cat.show_in_sidebar,
|
||
"mega_group": cat.mega_group,
|
||
"badge": cat.badge,
|
||
"is_active": cat.is_active,
|
||
"description": cat.description,
|
||
"image_url": cat.image_url,
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# REVIEW MODERATION
|
||
# ------------------------------------------------------------------
|
||
def list_reviews_for_moderation(self, approved: Optional[bool] = None) -> List[ReviewModerationResponse]:
|
||
stmt = select(ProductReview).order_by(ProductReview.created_at.desc())
|
||
if approved is not None:
|
||
stmt = stmt.where(ProductReview.is_approved == approved)
|
||
reviews = self.db.execute(stmt).scalars().all()
|
||
return [self._map_review(r) for r in reviews]
|
||
|
||
def approve_review(self, review_id: str, payload: ReviewApprovalPayload) -> ReviewModerationResponse:
|
||
review = self.db.get(ProductReview, review_id)
|
||
if not review:
|
||
from fastapi import HTTPException
|
||
raise HTTPException(status_code=404, detail="Review not found")
|
||
review.is_approved = True
|
||
if payload.admin_reply is not None:
|
||
review.admin_reply = payload.admin_reply
|
||
self.db.commit()
|
||
self.db.refresh(review)
|
||
return self._map_review(review)
|
||
|
||
def reject_review(self, review_id: str) -> ReviewModerationResponse:
|
||
review = self.db.get(ProductReview, review_id)
|
||
if not review:
|
||
from fastapi import HTTPException
|
||
raise HTTPException(status_code=404, detail="Review not found")
|
||
review.is_approved = False
|
||
self.db.commit()
|
||
self.db.refresh(review)
|
||
return self._map_review(review)
|
||
|
||
def _map_review(self, r: ProductReview) -> ReviewModerationResponse:
|
||
return ReviewModerationResponse(
|
||
review_id=r.review_id,
|
||
product_id=r.product_id,
|
||
author_name=r.author_name,
|
||
author_email=getattr(r, "author_email", None),
|
||
rating=r.rating,
|
||
title=r.title,
|
||
comment=r.comment,
|
||
verified_purchase=r.verified_purchase,
|
||
is_approved=r.is_approved,
|
||
helpful_count=r.helpful_count,
|
||
admin_reply=r.admin_reply,
|
||
review_date=r.created_at.strftime("%Y-%m-%d") if r.created_at else None,
|
||
images=[img.image_url for img in r.images if img.is_approved] if r.images else [],
|
||
)
|