"""前台公开商品 API — 无需登录""" from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, or_, join from sqlalchemy.orm import selectinload from app.api.deps import get_db from app.core.models.product import Product, ProductImage from app.core.models.category import Category from app.schemas.common import PageResult from app.schemas.product import ProductListOut, ProductOut router = APIRouter(prefix="/store", tags=["前台商城"]) # 从请求头或查询参数获取 tenant_id（多租户；单租户项目默认取 id=1） async def get_tenant_id(tenant_id: int = Query(2)) -> int: return tenant_id @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, sort: Optional[str] = Query("popular"), # popular / newest / price_asc / price_desc db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_id), ): # Join Product with Category to get category name as "category" field j = join(Product, Category, Product.category_id == Category.id, isouter=True) q = select(Product, Category.name.label("category_name")).select_from(j).where( Product.tenant_id == tid, Product.status == "active", ) if keyword: kw = keyword.strip() # 绕过 aiomysql 多字节 UTF-8 bind param 编码 bug： # 直接用 text() + 格式化字符串将搜索词嵌入 SQL 字面量（无 bind param preparation） # 搜索词中的 % 和 _ 已在 Python 层转义为 \% 和 \_，MySQL LIKE 不把它们当作通配符 # 无需 ESCAPE 子句，因为转义在 Python 端完成 from sqlalchemy import text safe_kw = kw.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') kw_where = text( "(lower(products.name) LIKE lower('%" + safe_kw + "%') " "OR lower(products.sku) LIKE lower('%" + safe_kw + "%'))" ) q = q.where(kw_where) if category: q = q.where(Category.name == category) # 排序 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) rows = result.all() items = [] for row in rows: product = row[0] # Product object category_name = row[1] or "未分类" # Fetch cover image via raw query (relationship uses lazy="noload") img_r = await db.execute( select(ProductImage.url) .where(ProductImage.product_id == product.id, ProductImage.is_primary == True) .limit(1) ) img_row = img_r.fetchone() cover_url = img_row[0] if img_row else None items.append(ProductListOut( id=product.id, name=product.name, sku=product.sku, category=category_name, base_price=product.base_price, stock_qty=product.stock_qty, status=product.status, cover_url=cover_url, slug=product.slug, sales_count=getattr(product, 'sales_count', 0) or 0, created_at=product.created_at, )) 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_id), ): j = join(Product, Category, Product.category_id == Category.id, isouter=True) r = await db.execute( select(Product, Category.name.label("category_name")) .select_from(j) .where(Product.tenant_id == tid, Product.status == "active") .order_by(Product.sales_count.desc()) .limit(limit) ) rows = r.all() return [ ProductListOut( id=p.id, name=p.name, sku=p.sku, category=cname or "未分类", base_price=p.base_price, stock_qty=p.stock_qty, status=p.status, cover_url=getattr(p, 'cover_url', None), slug=p.slug, sales_count=getattr(p, 'sales_count', 0) or 0, created_at=p.created_at, ) for p, cname in rows ] @router.get("/products/{slug}", response_model=ProductOut, summary="商品详情（公开）") async def get_product( slug: str, db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_id), ): # 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 name for response category_name = "未分类" if p.category_id: cr = await db.execute( select(Category.name).where(Category.id == p.category_id) ) cname = cr.scalar_one_or_none() if cname: category_name = cname # 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 return ProductOut( id=p.id, name=p.name, sku=p.sku, category=category_name, 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=0, slug=p.slug, meta_title=p.meta_title, meta_description=p.meta_description, 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, ) @router.get("/categories", response_model=list[str], summary="分类列表（公开）") async def list_categories( db: AsyncSession = Depends(get_db), tid: int = Depends(get_tenant_id), ): r = await db.execute( select(Category.name) .where(Category.tenant_id == tid, Category.is_active == True) .order_by(Category.sort_order) ) names = [row[0] for row in r.all()] # 如果没有分类数据，返回默认值 return names or ["服装", "鞋靴", "配饰", "包袋", "运动户外", "新品上市"] @router.get("/banners", summary="首页 Banner（公开）") async def list_banners(tid: int = Depends(get_tenant_id)): # 暂时返回静态数据，后续可接数据库 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"}, ]