fe85197e68
Add full backend support for a demand request page, including: - New database models for demand hero, feature cards, form configuration, and product recommendations - Alembic migration for the new tables - Default seed data for demand page components - Public site config endpoint exposing demand data - Admin CRUD operations and configuration for demand modules - Extended lead query filters for source page, keyword, and date range - Updated Pydantic schemas for lead query parameters - Comprehensive test coverage for all new endpoints
67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from .models import DemandRecommendation, DemandRecommendationProduct, Product
|
|
from .serializers import model_dict
|
|
|
|
|
|
def demand_recommendation_error(status_code: int, message: str, code: str, details: dict | None = None) -> None:
|
|
raise HTTPException(status_code=status_code, detail={"message": message, "code": code, "details": details or {}})
|
|
|
|
|
|
def demand_recommendation_dict(item: DemandRecommendation, *, public: bool = False) -> dict:
|
|
links = sorted(item.products or [], key=lambda link: (link.sortOrder, link.productId))
|
|
if public:
|
|
links = [link for link in links if link.product and link.product.status == "published"]
|
|
|
|
data = {
|
|
"id": item.id,
|
|
"title": item.title,
|
|
"subtitle": item.subtitle,
|
|
"productIds": [link.productId for link in links],
|
|
"isActive": item.isActive,
|
|
}
|
|
if not public:
|
|
base = model_dict(item)
|
|
data.update(
|
|
{
|
|
"sortOrder": item.sortOrder,
|
|
"createdAt": base["createdAt"],
|
|
"updatedAt": base["updatedAt"],
|
|
}
|
|
)
|
|
return data
|
|
|
|
|
|
def demand_recommendation_query():
|
|
return (
|
|
select(DemandRecommendation)
|
|
.options(selectinload(DemandRecommendation.products).selectinload(DemandRecommendationProduct.product))
|
|
.order_by(DemandRecommendation.sortOrder.asc())
|
|
)
|
|
|
|
|
|
def validate_demand_recommendation_product_ids(db: Session, product_ids: list[str]) -> None:
|
|
if len(set(product_ids)) != len(product_ids):
|
|
demand_recommendation_error(422, "推荐线路商品不能重复", "DEMAND_RECOMMENDATION_PRODUCT_DUPLICATE")
|
|
if not product_ids:
|
|
return
|
|
|
|
products = db.scalars(select(Product).where(Product.id.in_(product_ids))).all()
|
|
found_ids = {product.id for product in products}
|
|
missing_ids = [product_id for product_id in product_ids if product_id not in found_ids]
|
|
if missing_ids:
|
|
demand_recommendation_error(422, "推荐线路商品不存在", "DEMAND_RECOMMENDATION_PRODUCT_NOT_FOUND", {"productIds": missing_ids})
|
|
|
|
|
|
def replace_demand_recommendation_products(db: Session, recommendation: DemandRecommendation, product_ids: list[str]) -> None:
|
|
validate_demand_recommendation_product_ids(db, product_ids)
|
|
db.execute(delete(DemandRecommendationProduct).where(DemandRecommendationProduct.recommendationId == recommendation.id))
|
|
db.flush()
|
|
recommendation.products = []
|
|
for index, product_id in enumerate(product_ids):
|
|
link = DemandRecommendationProduct(recommendationId=recommendation.id, productId=product_id, sortOrder=index)
|
|
recommendation.products.append(link)
|
|
db.add(link)
|