小鱼开发
af8c483910
当三级分类（level=3）精确匹配失败时，回退到上级（level=2） 分类随机选取一个子分类，避免 AI 生成无效 scene（如 '电路施工-电路施工'）导致素材匹配完全失败。 - CRUD: 新增 get_children_by_parent_id 方法 - match_material: 新增 _try_fallback_to_parent 辅助函数 - batch_match: 同步增加 fallback 逻辑 - 顺手修复 zip() 缺少 strict 参数的 lint 问题
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
"""
|
|
素材分类 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())
|
|
|
|
|
|
# 导出实例
|
|
broll_category = BrollCategoryCRUD()
|