101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
大屏素材服务
|
|
"""
|
|
import logging
|
|
from typing import List
|
|
|
|
from sqlalchemy import select, func, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.base_service import BaseService
|
|
from online_dev.screen_design.material_model import ScreenMaterialCategory, ScreenMaterial
|
|
from online_dev.screen_design.material_schema import (
|
|
MaterialCategoryCreateIn, MaterialCategoryUpdateIn,
|
|
MaterialCreateIn, MaterialUpdateIn,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MaterialCategoryService(BaseService[ScreenMaterialCategory, MaterialCategoryCreateIn, MaterialCategoryUpdateIn]):
|
|
"""素材分类服务"""
|
|
|
|
model = ScreenMaterialCategory
|
|
|
|
@classmethod
|
|
async def get_all(cls, db: AsyncSession) -> List[ScreenMaterialCategory]:
|
|
"""获取所有素材分类"""
|
|
stmt = select(ScreenMaterialCategory).where(
|
|
ScreenMaterialCategory.is_deleted == False
|
|
).order_by(ScreenMaterialCategory.sort.asc(), ScreenMaterialCategory.sys_create_datetime.desc())
|
|
result = await db.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def check_code_exists(cls, db: AsyncSession, code: str, exclude_id: str = None) -> bool:
|
|
"""检查编码是否存在"""
|
|
stmt = select(ScreenMaterialCategory).where(
|
|
ScreenMaterialCategory.code == code,
|
|
ScreenMaterialCategory.is_deleted == False,
|
|
)
|
|
if exclude_id:
|
|
stmt = stmt.where(ScreenMaterialCategory.id != exclude_id)
|
|
result = await db.execute(stmt)
|
|
return result.scalar_one_or_none() is not None
|
|
|
|
@classmethod
|
|
async def get_material_count(cls, db: AsyncSession, category_id: str) -> int:
|
|
"""获取分类下的素材数量"""
|
|
stmt = select(func.count()).where(
|
|
ScreenMaterial.category_id == category_id,
|
|
ScreenMaterial.is_deleted == False,
|
|
)
|
|
result = await db.execute(stmt)
|
|
return result.scalar() or 0
|
|
|
|
@classmethod
|
|
async def delete_with_materials(cls, db: AsyncSession, category_id: str) -> bool:
|
|
"""删除分类及其下的所有素材"""
|
|
# 删除分类下的所有素材（硬删除）
|
|
await db.execute(
|
|
delete(ScreenMaterial).where(ScreenMaterial.category_id == category_id)
|
|
)
|
|
|
|
# 删除分类（硬删除）
|
|
await db.execute(
|
|
delete(ScreenMaterialCategory).where(ScreenMaterialCategory.id == category_id)
|
|
)
|
|
|
|
return True
|
|
|
|
|
|
class MaterialService(BaseService[ScreenMaterial, MaterialCreateIn, MaterialUpdateIn]):
|
|
"""素材服务"""
|
|
|
|
model = ScreenMaterial
|
|
|
|
@classmethod
|
|
async def get_list_by_category(
|
|
cls,
|
|
db: AsyncSession,
|
|
category_id: str = None,
|
|
) -> List[ScreenMaterial]:
|
|
"""获取素材列表"""
|
|
stmt = select(ScreenMaterial).where(ScreenMaterial.is_deleted == False)
|
|
|
|
if category_id:
|
|
stmt = stmt.where(ScreenMaterial.category_id == category_id)
|
|
|
|
stmt = stmt.order_by(ScreenMaterial.sort.asc(), ScreenMaterial.sys_create_datetime.desc())
|
|
result = await db.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def batch_delete(cls, db: AsyncSession, material_ids: List[str]) -> int:
|
|
"""批量删除素材（硬删除）"""
|
|
stmt = delete(ScreenMaterial).where(ScreenMaterial.id.in_(material_ids))
|
|
result = await db.execute(stmt)
|
|
return result.rowcount
|