Files
meijiaka-zy/python-api/app/crud/broll_category.py
T
小鱼开发 3587559a87 fix: 素材回退逻辑支持模糊匹配二级分类
- 新增 broll_category.get_by_name_like_and_level() 模糊匹配方法
- _try_fallback_to_parent 增加三级降级策略:
  1. 精确匹配
  2. 模糊匹配 LIKE %parent_name%(兼容'电路施工'→'电路施工镜')
  3. 自动补后缀'镜'/'阶段'再精确匹配
- 解决 scene 中 parent_name 与数据库二级分类 name 不一致导致回退失败的问题
2026-06-02 15:51:43 +08:00

76 lines
2.4 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
素材分类 CRUD
==============
"""
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.crud.base import CRUDBase
from app.models.broll_category import BrollCategory
class BrollCategoryCRUD(CRUDBase[BrollCategory]):
"""素材分类数据访问"""
def __init__(self) -> None:
super().__init__(BrollCategory)
async def get_by_name_and_level(
self, db: AsyncSession, *, name: str, level: int
) -> BrollCategory | None:
"""根据名称和层级获取启用的分类"""
result = await db.execute(
select(BrollCategory).where(
BrollCategory.name == name,
BrollCategory.level == level,
BrollCategory.status == "active",
)
)
return result.scalar_one_or_none()
async def get_by_names_and_level(
self, db: AsyncSession, *, names: list[str], level: int
) -> list[BrollCategory]:
"""批量根据名称和层级获取启用的分类"""
if not names:
return []
result = await db.execute(
select(BrollCategory).where(
BrollCategory.name.in_(names),
BrollCategory.level == level,
BrollCategory.status == "active",
)
)
return list(result.scalars().all())
async def get_children_by_parent_id(
self, db: AsyncSession, *, parent_id: int, level: int
) -> list[BrollCategory]:
"""根据父分类 ID 和层级获取启用的子分类"""
result = await db.execute(
select(BrollCategory).where(
BrollCategory.parent_id == parent_id,
BrollCategory.level == level,
BrollCategory.status == "active",
)
)
return list(result.scalars().all())
async def get_by_name_like_and_level(
self, db: AsyncSession, *, name: str, level: int
) -> BrollCategory | None:
"""根据名称模糊匹配和层级获取启用的分类(LIKE %name%"""
result = await db.execute(
select(BrollCategory).where(
BrollCategory.name.like(f"%{name}%"),
BrollCategory.level == level,
BrollCategory.status == "active",
)
)
return result.scalar_one_or_none()
# 导出实例
broll_category = BrollCategoryCRUD()