小鱼开发
47bb987e06
- material_service: 精确查询失败后全量内存标准化匹配，兼容数据库 name 含不可见字符 - material_service: 素材时长过滤放宽到 70% 兜底，避免打轴合并导致匹配失败 - material_service: 增加详细 warn 日志，便于诊断未匹配原因 - broll_category: 新增 get_by_level 方法供全量查询使用 - VoiceMaterialLibrary: 上传弹窗文案换行显示 - ScriptCreation: 主题卡片 min-height 64px 修复文字截断
88 lines
2.8 KiB
Python
88 lines
2.8 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())
|
||
|
||
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()
|
||
|
||
async def get_by_level(
|
||
self, db: AsyncSession, *, level: int
|
||
) -> list[BrollCategory]:
|
||
"""根据层级获取所有启用的分类"""
|
||
result = await db.execute(
|
||
select(BrollCategory).where(
|
||
BrollCategory.level == level,
|
||
BrollCategory.status == "active",
|
||
)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
# 导出实例
|
||
broll_category = BrollCategoryCRUD()
|