"""前台公开商品 API — 无需登录""" from datetime import datetime, timezone from decimal import Decimal from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query, Request # Query kept for other params from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, or_ from app.api.deps import get_db, get_tenant_by_appid_or_domain as get_tenant_by_domain, get_optional_customer from app.core.cache import cache_get, cache_set, cache_delete_pattern from app.core.models.customer import Customer from app.core.models.product import Product, ProductImage, ProductPriceRule, ProductVariant, product_categories from app.core.models.stock_status import StockStatus from app.core.models.category import Category from app.core.models.tenant_settings import TenantSettings from app.schemas.common import PageResult from app.schemas.product import ProductListOut, ProductOut, StockStatusBrief from app.core.services.customer_fields import normalize_fields as _normalize_cpf from app.core.services.expiry import effective_expiry_date, expiry_status as calc_expiry_status router = APIRouter(prefix="/store", tags=["前台商城"]) # 商品页服务保障展示项默认值（租户未配置时返回，供各前端复用） DEFAULT_SERVICE_BADGES = [ {"icon": "🚚", "text": "全国包邮", "text_en": "Free Shipping", "url": ""}, {"icon": "↩️", "text": "7天退换", "text_en": "7-Day Returns", "url": ""}, {"icon": "🔒", "text": "正品保障", "text_en": "Authentic", "url": ""}, ] async def _expiry_warning_days(db: AsyncSession, tenant_id: int) -> int: r = await db.execute(select(TenantSettings.extra).where(TenantSettings.tenant_id == tenant_id)) extra = r.scalar_one_or_none() or {} try: return max(0, int(extra.get("expiry_warning_days", 30))) except (TypeError, ValueError): return 30 async def _resolve_store_display( db: AsyncSession, *, tenant_id: int, product_id: int, variant_id: Optional[int], member_level_id: Optional[int], now: Optional[datetime] = None, ) -> tuple[Decimal, Decimal, bool, Optional[dict]]: """按 store 渠道 + 当前客户 + 单件数量解析商品展示价。 返回 (display_price, original_price, is_promotion, rule_summary)： - display_price 命中规则价/未命中回落到 base+modifier - original_price 划线原价（base+modifier） - is_promotion display_price 是否来自 is_promotion=true 的规则 - rule_summary 命中规则的可审计摘要（None = 未命中规则） """ from app.core.services.pricing import _resolve_rule_for_line rule_now = now or datetime.now(timezone.utc).replace(tzinfo=None) rule, rule_unit = await _resolve_rule_for_line( db, tenant_id=tenant_id, product_id=product_id, variant_id=variant_id, channel="store", member_level_id=member_level_id, qty=Decimal("1"), now=rule_now, ) base = (await db.execute( select(Product.base_price).where( Product.id == product_id, Product.tenant_id == tenant_id, ) )).scalar_one_or_none() base_price = Decimal(base) if base is not None else Decimal("0") if variant_id is not None: mod = (await db.execute( select(ProductVariant.price_modifier).where(ProductVariant.id == variant_id) )).scalar_one_or_none() if mod is not None: base_price = base_price + Decimal(mod) original_price = base_price display_price = rule_unit if rule_unit is not None else base_price is_promotion = bool(rule.is_promotion) if rule is not None else False summary = None if rule is not None: summary = { "id": rule.id, "channel_scope": rule.channel_scope, "variant_id": rule.variant_id, "member_level_id": rule.member_level_id, "min_quantity": rule.min_quantity, "price_type": rule.price_type, "price_value": float(rule.price_value), "priority": rule.priority, "is_promotion": bool(rule.is_promotion), "starts_at": rule.starts_at.isoformat() if rule.starts_at else None, "ends_at": rule.ends_at.isoformat() if rule.ends_at else None, } return display_price, original_price, is_promotion, summary async def _resolve_category_ids(db: AsyncSession, tenant_id: int, category: str) -> set[int]: """把分类 slug/name 解析为要筛选的分类 id 集合。 租户开启 category_filter_recursive 时，连带所选分类的全部子孙分类； 否则仅精确匹配所选分类本身（保持原行为）。 """ rows = (await db.execute( select(Category.id, Category.parent_id, Category.slug, Category.name) .where(Category.tenant_id == tenant_id) )).all() matched = {r.id for r in rows if r.slug == category or r.name == category} if not matched: return set() st = (await db.execute( select(TenantSettings.extra).where(TenantSettings.tenant_id == tenant_id) )).scalar_one_or_none() or {} if not bool(st.get("category_filter_recursive", False)): return matched # 内存 BFS 收集全部后代 ponytail: 分类量级小，内存遍历足够；过万再换递归 CTE children: dict[int, list[int]] = {} for r in rows: children.setdefault(r.parent_id, []).append(r.id) ids, stack = set(), list(matched) while stack: c = stack.pop() if c in ids: continue ids.add(c) stack.extend(children.get(c, [])) return ids @router.get("/products", response_model=PageResult[ProductListOut], summary="商品列表（公开）") async def list_products( page: int = Query(1, ge=1), page_size: int = Query(12, ge=1, le=100), keyword: Optional[str] = None, category: Optional[str] = None, brand: Optional[str] = None, min_price: Optional[float] = None, max_price: Optional[float] = None, ids: Optional[str] = None, sort: Optional[str] = Query("popular"), # popular / newest / price_asc / price_desc db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), customer: Customer | None = Depends(get_optional_customer), ): from app.core.models.brand import Brand as BrandModel q = select(Product).where( Product.tenant_id == tid, Product.status == "active", ) if keyword: kw = f"%{keyword.strip()}%" q = q.where(or_(Product.name.ilike(kw), Product.sku.ilike(kw))) if ids: try: selected_ids = [int(value) for value in ids.split(",") if value.strip()] except ValueError: selected_ids = [] q = q.where(Product.id.in_(selected_ids) if selected_ids else Product.id.is_(None)) if category: # 同时支持按 slug（URL友好）或 name 匹配，兼容前端两种传参方式 cat_ids = await _resolve_category_ids(db, tid, category) if cat_ids: cat_sub = select(product_categories.c.product_id).where( product_categories.c.category_id.in_(cat_ids) ) q = q.where(Product.id.in_(cat_sub)) else: q = q.where(Product.id.is_(None)) # 分类不存在 → 空结果，与原精确匹配一致 if brand: # 同时支持按 slug（URL友好）或 name 匹配 brand_sub = ( select(BrandModel.id) .where( or_(BrandModel.slug == brand, BrandModel.name == brand), BrandModel.tenant_id == tid, ) ) q = q.where(Product.brand_id.in_(brand_sub)) if min_price is not None and min_price > 0: q = q.where(Product.base_price >= min_price) if max_price is not None and max_price > 0: q = q.where(Product.base_price <= max_price) # 排序 if sort == "price_asc": q = q.order_by(Product.base_price.asc()) elif sort == "price_desc": q = q.order_by(Product.base_price.desc()) elif sort == "newest": q = q.order_by(Product.created_at.desc(), Product.id.desc()) else: # popular q = q.order_by(Product.sales_count.desc(), Product.created_at.desc()) total_r = await db.execute(select(func.count()).select_from(q.subquery())) total = total_r.scalar() or 0 q = q.offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) products = result.scalars().all() if not products: return PageResult(items=[], total=total, page=page, page_size=page_size) product_ids = [p.id for p in products] cats_r = await db.execute( select(product_categories.c.product_id, Category.name) .join(Category, product_categories.c.category_id == Category.id) .where(product_categories.c.product_id.in_(product_ids)) ) product_cat_map: dict[int, list[str]] = {} for pid, cname in cats_r.all(): product_cat_map.setdefault(pid, []).append(cname) # 批量获取主图，避免 N+1 covers_r = await db.execute( select(ProductImage.product_id, ProductImage.url) .where(ProductImage.product_id.in_(product_ids), ProductImage.is_primary == True) .distinct(ProductImage.product_id) .order_by(ProductImage.product_id, ProductImage.sort_order) ) cover_map: dict[int, str] = {} for pid, url in covers_r.all(): cover_map.setdefault(pid, url) # 批量获取 tier prices（按会员等级的固定价格覆盖），供前端展示会员价 from app.core.models.product import ProductTierPrice tier_r = await db.execute( select(ProductTierPrice).where( ProductTierPrice.product_id.in_(product_ids), ProductTierPrice.variant_id.is_(None), # 商品级定价 ) ) tier_prices_map: dict[int, list[dict]] = {} for tp in tier_r.scalars().all(): tier_prices_map.setdefault(tp.product_id, []).append({ "member_level_id": tp.member_level_id, "price": float(tp.price), }) ss_ids = {p.stock_status_id for p in products if p.stock_status_id} ss_map: dict[int, StockStatusBrief] = {} if ss_ids: ss_r = await db.execute(select(StockStatus).where(StockStatus.id.in_(ss_ids))) for ss in ss_r.scalars().all(): ss_map[ss.id] = StockStatusBrief.model_validate(ss) items = [] for p in products: display_price, original_price, is_promotion, summary = await _resolve_store_display( db, tenant_id=tid, product_id=p.id, variant_id=None, member_level_id=customer.member_level_id if customer else None, ) items.append(ProductListOut( id=p.id, name=p.name, name_en=p.name_en, sku=p.sku, categories=product_cat_map.get(p.id, []), base_price=p.base_price, member_price=p.member_price, stock_qty=p.stock_qty, status=p.status, cover_url=cover_map.get(p.id), slug=p.slug, sales_count=getattr(p, 'sales_count', 0) or 0, created_at=p.created_at, tier_prices=tier_prices_map.get(p.id, []), stock_status=ss_map.get(p.stock_status_id), display_price=display_price, original_price=original_price, is_promotion=is_promotion, active_rule_summary=summary, )) return PageResult(items=items, total=total, page=page, page_size=page_size) @router.get("/products/featured", response_model=list[ProductListOut], summary="推荐商品（公开）") async def featured_products( limit: int = Query(8, ge=1, le=50), db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), customer: Customer | None = Depends(get_optional_customer), ): page = await list_products(page=1, page_size=limit, sort="popular", db=db, tid=tid, customer=customer) return [item.model_dump() for item in page.items] @router.get("/products/{slug}", response_model=ProductOut, summary="商品详情（公开）") async def get_product( slug: str, db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), customer: Customer | None = Depends(get_optional_customer), ): # slug 或 id 均可查询 q = select(Product).where(Product.tenant_id == tid, Product.status == "active") if slug.isdigit(): q = q.where(Product.id == int(slug)) else: q = q.where(Product.slug == slug) r = await db.execute(q) p = r.scalar_one_or_none() if not p: raise HTTPException(404, "商品不存在或已下架") # Get category names for response cats_r = await db.execute( select(Category.name) .join(product_categories, Category.id == product_categories.c.category_id) .where(product_categories.c.product_id == p.id) ) category_names = [row[0] for row in cats_r.all()] # Fetch images separately (relationship uses lazy="noload", selectinload unreliable) img_r = await db.execute( select(ProductImage.url) .where(ProductImage.product_id == p.id) .order_by(ProductImage.is_primary.desc(), ProductImage.sort_order) .limit(20) ) img_urls = [row[0] for row in img_r.all()] cover_url = img_urls[0] if img_urls else None # Fetch active variants var_r = await db.execute( select(ProductVariant) .where(ProductVariant.product_id == p.id, ProductVariant.is_active == True) .order_by(ProductVariant.sort_order) ) variant_objs = var_r.scalars().all() warning_days = await _expiry_warning_days(db, p.tenant_id) product_expiry = effective_expiry_date(p.expiry_date, [v.expiry_date for v in variant_objs]) variants = [] for v in variant_objs: variant_display, variant_original, variant_promotion, variant_summary = await _resolve_store_display( db, tenant_id=tid, product_id=p.id, variant_id=v.id, member_level_id=customer.member_level_id if customer else None, ) variants.append({ "id": v.id, "product_id": v.product_id, "tenant_id": v.tenant_id, "sku": v.sku, "barcode": v.barcode, "price_modifier": str(v.price_modifier or "0"), "stock_qty": v.stock_qty or 0, "reserved_qty": v.reserved_qty or 0, "image_url": v.image_url, "weight_grams": int(float(v.weight) * 1000) if v.weight else None, "sort_order": v.sort_order or 0, "attributes": v.attributes or {}, "is_active": bool(v.is_active), "expiry_date": v.expiry_date or p.expiry_date, "expiry_status": calc_expiry_status(v.expiry_date or p.expiry_date, warning_days=warning_days), "display_price": variant_display, "original_price": variant_original, "is_promotion": variant_promotion, "active_rule_summary": variant_summary, }) ss_brief = None if p.stock_status_id: ss_r = await db.execute(select(StockStatus).where(StockStatus.id == p.stock_status_id)) ss_obj = ss_r.scalar_one_or_none() if ss_obj: ss_brief = StockStatusBrief.model_validate(ss_obj) # 详情页规则展示：按 store 渠道 + 单件数量 + 当前登录会员等级；未登录按 walk-in display_price, original_price, is_promotion, summary = await _resolve_store_display( db, tenant_id=tid, product_id=p.id, variant_id=None, member_level_id=customer.member_level_id if customer else None, ) return ProductOut( id=p.id, name=p.name, sku=p.sku, categories=category_names, description=p.description, base_price=p.base_price, cost_price=p.cost_price, stock_qty=p.stock_qty, low_stock_threshold=5, status=p.status, weight_grams=int(float(p.weight) * 1000) if p.weight else 0, slug=p.slug, meta_title=p.meta_title, meta_description=p.meta_description, seo_keywords=p.seo_keywords, cover_url=cover_url, sales_count=getattr(p, 'sales_count', 0) or 0, created_at=p.created_at, tenant_id=p.tenant_id, images=img_urls, variants=variants, shelf_life=p.shelf_life, shelf_life_en=p.shelf_life_en, expiry_date=product_expiry, expiry_status=calc_expiry_status(product_expiry, warning_days=warning_days), ai_description=p.ai_description, name_en=p.name_en, description_en=p.description_en, ai_description_en=p.ai_description_en, meta_title_en=p.meta_title_en, meta_description_en=p.meta_description_en, seo_keywords_en=p.seo_keywords_en, stock_status=ss_brief, stock_status_id=p.stock_status_id, display_price=display_price, original_price=original_price, is_promotion=is_promotion, active_rule_summary=summary, ) class StoreCategoryNode(BaseModel): id: int name: str slug: str parent_id: Optional[int] image_url: Optional[str] sort_order: int children: list["StoreCategoryNode"] = [] name_en: Optional[str] = None model_config = {"from_attributes": True} StoreCategoryNode.model_rebuild() class StoreCategoryDetail(BaseModel): id: int name: str slug: str parent_id: Optional[int] image_url: Optional[str] description: Optional[str] = None meta_title: Optional[str] = None meta_description: Optional[str] = None seo_keywords: Optional[str] = None sort_order: int name_en: Optional[str] = None description_en: Optional[str] = None meta_title_en: Optional[str] = None meta_description_en: Optional[str] = None seo_keywords_en: Optional[str] = None model_config = {"from_attributes": True} @router.get("/categories", response_model=list[StoreCategoryNode], summary="分类树（公开）") async def list_categories( db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), ): cache_key = f"categories:{tid}" cached = await cache_get(cache_key) if cached is not None: return cached r = await db.execute( select( Category.id, Category.name, Category.slug, Category.parent_id, Category.image_url, Category.sort_order, Category.name_en, ) .where(Category.tenant_id == tid, Category.is_active == 1) .order_by(Category.sort_order, Category.id) ) rows = r.all() if not rows: # 数据库无分类时返回默认兜底 return [ StoreCategoryNode(id=-1, name="服装", slug="clothing", parent_id=None, image_url=None, sort_order=0), StoreCategoryNode(id=-2, name="鞋靴", slug="shoes", parent_id=None, image_url=None, sort_order=1), StoreCategoryNode(id=-3, name="配饰", slug="accessories",parent_id=None,image_url=None, sort_order=2), StoreCategoryNode(id=-4, name="包袋", slug="bags", parent_id=None, image_url=None, sort_order=3), StoreCategoryNode(id=-5, name="运动户外", slug="sports", parent_id=None, image_url=None, sort_order=4), StoreCategoryNode(id=-6, name="新品上市", slug="new-arrivals",parent_id=None,image_url=None,sort_order=5), ] # 构建树形结构 node_map: dict[int, StoreCategoryNode] = {} for row in rows: node_map[row.id] = StoreCategoryNode( id=row.id, name=row.name, slug=row.slug, parent_id=row.parent_id, image_url=row.image_url, sort_order=row.sort_order, name_en=row.name_en, ) roots: list[StoreCategoryNode] = [] for node in node_map.values(): if node.parent_id and node.parent_id in node_map: node_map[node.parent_id].children.append(node) else: roots.append(node) result = [n.model_dump() for n in roots] await cache_set(cache_key, result, ttl=600) # 10分钟 return roots @router.get("/categories/{slug}", response_model=StoreCategoryDetail, summary="分类详情（公开）") async def get_category_detail( slug: str, db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), ): r = await db.execute( select(Category).where( Category.slug == slug, Category.tenant_id == tid, Category.is_active == 1, ) ) cat = r.scalar_one_or_none() if not cat: raise HTTPException(status_code=404, detail="分类不存在") return cat @router.get("/shop-info", summary="店铺公开信息（公开）") async def get_shop_info( db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), ): cache_key = f"shop_info:{tid}" cached = await cache_get(cache_key) if cached is not None: return cached r = await db.execute( select(TenantSettings).where(TenantSettings.tenant_id == tid) ) s = r.scalar_one_or_none() if not s: return {"store_name": "SME Store", "store_logo": None, "store_subtitle": None, "show_store_name": True, "show_store_subtitle": True, "store_description": None, "site_meta_title": None, "site_meta_description": None, "site_keywords": None, "og_image": None, "favicon_url": None, "shipping_policy": None, "show_real_stock": True, "out_of_stock_display": { "name": "缺货", "name_en": "Out of Stock", "badge_color": "#ef4444", "badge_text": "已售罄", "badge_text_en": "Sold Out", }, "default_locale": "zh", "service_badges": DEFAULT_SERVICE_BADGES, "store_name_en": None, "store_subtitle_en": None, "site_meta_title_en": None, "site_meta_description_en": None, "site_keywords_en": None, "shipping_policy_en": None, "ai_chat_name_en": None, "ai_chat_welcome_en": None} extra = s.extra or {} result = { "store_name": s.store_name or "SME Store", "store_description": s.store_description or None, "store_logo": extra.get("logo_url") or None, "store_subtitle": extra.get("store_subtitle") or None, "show_store_name": bool(extra.get("show_store_name", True)), "show_store_subtitle": bool(extra.get("show_store_subtitle", True)), "address_custom_fields": extra.get("address_custom_fields") or [], "customer_profile_fields": _normalize_cpf(extra.get("customer_profile_fields") or []), "site_meta_title": extra.get("site_meta_title") or None, "site_meta_description": extra.get("site_meta_description") or None, "site_keywords": extra.get("site_keywords") or None, "og_image": extra.get("og_image") or None, "favicon_url": extra.get("favicon_url") or None, "shipping_policy": extra.get("shipping_policy") or None, "show_real_stock": bool(extra.get("show_real_stock", True)), "service_badges": extra.get("service_badges") or DEFAULT_SERVICE_BADGES, "default_locale": extra.get("default_locale") or "zh", "store_name_en": extra.get("store_name_en") or None, "store_subtitle_en": extra.get("store_subtitle_en") or None, "site_meta_title_en": extra.get("site_meta_title_en") or None, "site_meta_description_en": extra.get("site_meta_description_en") or None, "site_keywords_en": extra.get("site_keywords_en") or None, "shipping_policy_en": extra.get("shipping_policy_en") or None, "ai_chat_name_en": extra.get("ai_chat_name_en") or None, "ai_chat_welcome_en": extra.get("ai_chat_welcome_en") or None, } try: from app.plugins.tax.models import TaxSettings as TaxSettingsModel tr = await db.execute( select(TaxSettingsModel).where(TaxSettingsModel.tenant_id == tid) ) ts = tr.scalar_one_or_none() if ts: result["prices_include_tax"] = ts.prices_include_tax result["display_prices_in_shop"] = ts.display_prices_in_shop result["tax_registration_no"] = ts.tax_registration_no or "" else: result["prices_include_tax"] = False result["display_prices_in_shop"] = "excl" result["tax_registration_no"] = "" except Exception: result["prices_include_tax"] = False result["display_prices_in_shop"] = "excl" result["tax_registration_no"] = "" # 缺货默认显示状态（从 stock_statuses 取 out_of_stock 系统状态） oos_r = await db.execute( select(StockStatus).where( StockStatus.tenant_id == tid, StockStatus.slug == "out_of_stock", ) ) oos = oos_r.scalar_one_or_none() if oos: result["out_of_stock_display"] = { "name": oos.name, "name_en": oos.name_en, "badge_color": oos.badge_color, "badge_text": oos.badge_text, "badge_text_en": oos.badge_text_en, } else: result["out_of_stock_display"] = { "name": "缺货", "name_en": "Out of Stock", "badge_color": "#ef4444", "badge_text": "已售罄", "badge_text_en": "Sold Out", } await cache_set(cache_key, result, ttl=300) # 5分钟 return result @router.get("/address-custom-fields", summary="地址自定义字段定义（公开）") async def get_address_custom_fields( db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_by_domain), ): r = await db.execute( select(TenantSettings).where(TenantSettings.tenant_id == tid) ) s = r.scalar_one_or_none() extra = s.extra or {} if s else {} return {"fields": extra.get("address_custom_fields") or []} @router.get("/banners", summary="首页 Banner（公开）") async def list_banners(tid: int = Depends(get_tenant_by_domain)): # 暂时返回静态数据，后续可接数据库 return [ {"id": 1, "title": "夏季新品上市", "subtitle": "探索最新时尚单品", "url": "/products?category=新品上市", "cta": "立即选购", "image": "https://picsum.photos/seed/banner1/1200/400"}, {"id": 2, "title": "品牌特惠专场", "subtitle": "限时折扣，低至 5 折", "url": "/products", "cta": "查看活动", "image": "https://picsum.photos/seed/banner2/1200/400"}, {"id": 3, "title": "精选配饰系列", "subtitle": "提升穿搭质感", "url": "/products?category=配饰", "cta": "浏览配饰", "image": "https://picsum.photos/seed/banner3/1200/400"}, ]