""" @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 [], )