72d388a047
Replace hardcoded homepage featured route groups with a fully managed dynamic system: - add database models RouteSection and RouteSectionProduct for storing route groups and their associated products - create Alembic migration 0004_route_sections for the new tables - extend SiteConfigPatchIn schema with subtitle and productIds fields - refactor shared site_config utility to load dynamic route sections instead of fixed groups - implement admin CRUD API with validation for duplicate/conflicting product associations - update public and admin API documentation to reflect the new system - add default route section seed data and comprehensive test coverage
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from .models import Product, RouteSection, RouteSectionProduct
|
|
from .serializers import model_dict
|
|
|
|
|
|
ROUTE_SECTION_DEFAULTS = [
|
|
("routes", "经典人文打卡线路", "黄果树、荔波小七孔、千户苗寨、镇远古城、梵净山一次串联", 0, 8),
|
|
("routes-outdoor", "极限山野户外野咖线路", "溶洞、峡谷、漂流、峰林骑行和山野咖啡组合", 8, 16),
|
|
("routes-mix", "人文+户外综合混搭线路", "非遗村寨、古城夜游、自然轻探险和精品住宿同程安排", 16, 24),
|
|
]
|
|
|
|
|
|
def route_section_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 route_section_dict(item: RouteSection, *, 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 route_section_query():
|
|
return (
|
|
select(RouteSection)
|
|
.options(selectinload(RouteSection.products).selectinload(RouteSectionProduct.product))
|
|
.order_by(RouteSection.sortOrder.asc())
|
|
)
|
|
|
|
|
|
|
|
|
|
def validate_route_section_product_ids(db: Session, section_id: str, product_ids: list[str]) -> None:
|
|
if len(set(product_ids)) != len(product_ids):
|
|
route_section_error(422, "线路商品不能重复", "ROUTE_SECTION_PRODUCT_DUPLICATE", {"sectionId": section_id})
|
|
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:
|
|
route_section_error(422, "线路商品不存在", "ROUTE_SECTION_PRODUCT_NOT_FOUND", {"productIds": missing_ids})
|
|
|
|
conflicts = db.scalars(
|
|
select(RouteSectionProduct).where(
|
|
RouteSectionProduct.productId.in_(product_ids),
|
|
RouteSectionProduct.sectionId != section_id,
|
|
)
|
|
).all()
|
|
if conflicts:
|
|
conflict = conflicts[0]
|
|
route_section_error(
|
|
409,
|
|
"同一线路商品不能同时属于多个精选线路子分组",
|
|
"ROUTE_SECTION_PRODUCT_CONFLICT",
|
|
{"productId": conflict.productId, "sectionId": conflict.sectionId},
|
|
)
|
|
|
|
|
|
def replace_route_section_products(db: Session, section: RouteSection, product_ids: list[str]) -> None:
|
|
validate_route_section_product_ids(db, section.id, product_ids)
|
|
db.execute(delete(RouteSectionProduct).where(RouteSectionProduct.sectionId == section.id))
|
|
db.flush()
|
|
section.products = []
|
|
for index, product_id in enumerate(product_ids):
|
|
link = RouteSectionProduct(sectionId=section.id, productId=product_id, sortOrder=index)
|
|
section.products.append(link)
|
|
db.add(link)
|
|
|