kangda/app/crud/event.py

from typing import List, Optional, Dict, Any
from sqlalchemy import select, and_, or_, update, bindparam, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.crud.base import CRUDBase
from app.models.models import Event, Image, Temperature, Message, Robot
from app.schemas.event import EventUpdate, EventQuery, BackStageEvent, BackStageEventDto, BackStageEventDetail, EditTemperatureDto, OcrAlertMessage, OcrAlertMessageDto, GetRobotDto

from pydantic import VERSION as PYDANTIC_VERSION
from datetime import datetime

# from app.util import kangda

class CRUDEvent(CRUDBase[Event, EventUpdate, EventUpdate]):
    async def get_by_id(self, db: AsyncSession, *, event_id: str) -> Optional[Event]:
        """根据ID获取事件"""
        query = (
            select(Event)
            .options(
                selectinload(Event.images),
                selectinload(Event.temperatures)
            )
            .where(Event.eventId == event_id)
        )
        result = await db.execute(query)
        return result.scalar_one_or_none()
    
    async def get_event_detail(
        self,
        db: AsyncSession,
        *,
        eventId: str 
    ) -> Optional[BackStageEventDetail]:
        stmt = (select(Event.eventId, Event.number, Event.name,Image.imageUrl, Image.localPath, Temperature.temperature,Temperature.confidence, Temperature.status, Temperature.createTime).
                select_from(Event).
                outerjoin(Image, Image.eventId == Event.eventId).
                outerjoin(Temperature, Event.eventId==Temperature.eventId).
                where(Event.eventId==eventId)
                )
        
        result = await db.execute(stmt)
        result = result.mappings().first()
        
        return result
    
    async def get_multi_backstage_events(
        self,
        db: AsyncSession,
        *,
        query: BackStageEventDto
    ) -> List[BackStageEvent]:

        """后台根据条件获取事件列表"""
        
        conditions = []
        
        if query.eventId is not None and len(query.eventId) > 0:
            conditions.append(Event.eventId==query.eventId)
        if query.number is not None and len(query.number) > 0:
            conditions.append(Event.number == query.number)
        if query.name is not None and  len(query.name) > 0:
            conditions.append(Event.name == query.name)
        if query.status is not None and len(query.status) > 0:
            conditions.append(Temperature.status == query.status)
        if query.start_time is not None:
            conditions.append(Temperature.createTime >= query.start_time)
        if query.end_time is not None:
            conditions.append(Temperature.createTime <= query.end_time)
            

        
        stmt = (select(Event.eventId, Event.number, Event.name, Image.imageUrl, Image.localPath, Temperature.temperature, Temperature.confidence, Temperature.createTime, Temperature.status)
                .select_from(Temperature)
                .outerjoin(Event, Event.eventId==Temperature.eventId)
                .outerjoin(Image, Image.imageId == Temperature.imageId)
                .where(*conditions)
                .order_by(Temperature.createTime.desc())
                .offset(query.skip)
                .limit(query.limit))
        
        results = await db.execute(stmt)
        
        results = results.mappings().all()
        
        back = list()
        
        for result in results:
            # if result["temperature"] is not None:
            back.append(BackStageEvent(**result))
                
        return back

                
        

    async def get_multi_with_query(
        self, 
        db: AsyncSession, 
        *, 
        query: EventQuery
    ) -> List[Event]:
        """根据查询条件获取事件列表"""
        conditions = []
        
        if query.start_time:
            conditions.append(Event.insDate >= query.start_time)
        if query.end_time:
            conditions.append(Event.insDate <= query.end_time)
        if query.etypeName:
            conditions.append(Event.etypeName == query.etypeName)
        if query.area:
            conditions.append(Event.area == query.area)
            
        query_stmt = (
            select(Event)
            .options(
                selectinload(Event.images),
                selectinload(Event.temperatures)
            )
        )
        
        if conditions:
            query_stmt = query_stmt.where(and_(*conditions))
            
        query_stmt = query_stmt.offset(query.skip).limit(query.limit)
        
        result = await db.execute(query_stmt)
        return result.scalars().all()

    async def update_event(
        self,
        db: AsyncSession,
        *,
        event_id: str,
        obj_in: EditTemperatureDto
    ) -> Optional[Event]:
        """更新事件信息"""
        
        dict_values = dict()
        
        if PYDANTIC_VERSION.startswith("2."):
            field_names = list(EditTemperatureDto.model_fields.keys())
        else:
            field_names = list(EditTemperatureDto.__fields__.keys())
        
        for key in field_names:
            
            value = getattr(obj_in, key)
            if value is not None:
                dict_values[key] = value
        try:
            update_stmt = (
                update(Temperature).where(Temperature.eventId == event_id).values(**dict_values)
            )
            
            await db.execute(update_stmt)
            await db.commit()
            
            return True
        except:
            return False
        
        
        # event = await self.get_by_id(db, event_id=event_id)
        # if not event:
        #     return None
            
        # update_data = obj_in.model_dump()
        # for field, value in update_data.items():
        #     setattr(event, field, value)
            
        # # db.add(event)
        # await db.commit()
        # await db.refresh(event)
        # return event

    async def delete_event(
        self,
        db: AsyncSession,
        *,
        event_id: str
    ) -> Optional[Event]:
        """删除事件"""
        event = await self.get_by_id(db, event_id=event_id)
        if not event:
            return None
            
        await db.delete(event)
        await db.commit()
        return event
    
    async def get_messages(
        self,
        db: AsyncSession,
        skip:int,
        limit:int,
        condition: list = []
    ) -> Optional[List[OcrAlertMessage]]:
        query_stmt = (
            select(Message.messageId,Message.eventId ,Message.messageType, Message.eventType, Message.handle, Message.remark, Message.createTime,
                   Event.number,Event.name,
                #    Image.imageUrl, Image.localPath,
                   Temperature.temperature)
            .select_from(Message)
            .outerjoin(Event, Message.eventId == Event.eventId)
            # .outerjoin(Image, Message.eventId == Image.eventId)
            .outerjoin(Temperature, Message.eventId == Temperature.eventId)
            .where(*condition)
            .order_by(Message.createTime.desc())
            .offset(skip)
            .limit(limit)
        )
        
        messages = await db.execute(query_stmt)
        messages = messages.mappings().all()
    
        back = list()
        for message in messages:
            # 获取图片URL列表
            query_image_stmt = (
                select(Image.imageUrl)
                .where(Image.eventId == message.get("eventId"))
            )
            images = await db.execute(query_image_stmt)
            images = images.mappings().all()
            
            # 将RowMapping转换为字典
            message_dict = dict(message)
            # 将图片URL列表转换为普通列表
            image_urls = [dict(img)["imageUrl"] for img in images]
            message_dict["imageUrl"] = image_urls
            
            ocrAlertMessage = OcrAlertMessage(**message_dict)
            back.append(ocrAlertMessage)
        
        return back
    
    async def handle_ocr_alerts(
        self,
        db:AsyncSession,
        messageIdList: List[int]
    ):
        """处理OCR告警消息
        
        Args:
            db: 数据库会话
            messageIdList: 要处理的messageId列表
            
        Returns:
            bool: 更新是否成功
        """
        try:
            # 获取所有需要更新的消息ID
            # 构建更新语句
            update_stmt = (
                update(Message)
                .where(Message.messageId.in_(messageIdList))
                .values(
                    handle="1",
                    updateTime=datetime.now()
                )
            )
            
            # 准备更新数据
            
            # 执行更新
            await db.execute(
                update_stmt,
                # update_data,
                execution_options={"synchronize_session": False}
            )
            await db.commit()
            return True
            
        except Exception as e:
            print(f"更新OCR告警消息失败: {str(e)}")
            await db.rollback()
            return False
        
    async def handle_ocr_alerts_get(
        self,
        db: AsyncSession
    ):
        try:
            update_stmt = (
                update(Message)
                .where(Message.handle == "0")
                .values(
                    handle = "1",
                    updateTime=datetime.now()
                )
            )
            
            # 执行更新
            await db.execute(
                update_stmt,
                execution_options={"synchronize_session": False}
            )
            await db.commit()
            return True
            
        except Exception as e:
            print(f"更新OCR告警消息失败: {str(e)}")
            await db.rollback()
            return False
    async def handle_ocr_alert(
        self, 
        db: AsyncSession,
        ocrAlertMessageDto: OcrAlertMessageDto
        ):
        try:
            update_stmt = (
                update(Message)
                .where(Message.messageId == ocrAlertMessageDto.messageId)
                .values(
                    remark = ocrAlertMessageDto.remark,
                    updateTime=datetime.now()
                    )
            )
            await db.execute(update_stmt)
            await db.commit()
            return True
        
        except Exception as e:
            print(f"处理单个消息失败: {e}")
            await db.rollback()
            return False
        
    async def get_alert_detail(
        self, 
        db: AsyncSession,
        messageId: int
        ) ->Optional[OcrAlertMessage]:
        try:
            select_stmt = (
                select(Message.messageId,Message.eventId ,Message.messageType, Message.eventType, Message.handle, Message.remark, Message.createTime,
                   Event.number,Event.name,
                   Image.imageUrl, Image.localPath,
                   Temperature.temperature)
                .select_from(Message)
                .outerjoin(Event, Message.eventId == Event.eventId)
                .outerjoin(Image, Message.eventId == Image.eventId)
                .outerjoin(Temperature, Message.eventId == Temperature.eventId)
                .where(Message.messageId == messageId)
                )
            
            result = await db.execute(select_stmt)
            message = result.mappings().first()
            
            if message:
                # 获取图片URL列表
                query_image_stmt = (
                    select(Image.imageUrl)
                    .where(Image.eventId == message.get("eventId"))
                )
                images = await db.execute(query_image_stmt)
                images = images.mappings().all()
                
                # 将RowMapping转换为字典
                message_dict = dict(message)
                # 添加图片URL列表
                message_dict["imageUrl"] = images
                
                return OcrAlertMessage(**message_dict)
            return None
            
        except Exception as e:
            print(f"获取告警详情失败: {e}")
            return None

    async def get_alert_count(
        self,
        db: AsyncSession 
    ):
        try:
            query_stmt = (
                select(func.count(Message.messageId))
                .where(Message.handle == "0")
            )
            
            result = await db.execute(query_stmt)
            return result.scalar()
        except Exception as e:
            print(f"{e}")
            
    async def get_robot_count(
        self, 
        db: AsyncSession,
        *, # 关键字参数分隔符,代表后面的参数一定要用关键字指定
        status: list 
    ):
        try:
            query_stmt = (
                select(func.count(Robot.robotId))
                .where(*status)
            )
            result = await db.execute(query_stmt)
            return result.scalar()
        
        except Exception as e:
            print(f"{e}")
            
    async def get_messages_st(
        self,
        db: AsyncSession,
        day: int
    ):
        try:
            query_stmt = (
                select(
                    # Message.eventType, 
                       func.count(Message.messageId))
                .group_by(Message.eventType)
            )
            result = await db.execute(query_stmt)
            return result.scalars()
        except Exception as e:
            print(f"{e}")
            
    async def get_robot_list(
        self,
        db: AsyncSession,
        get_robot_dto: GetRobotDto
    ):
        # 前端登录
        d = kangda._login_front()
        duty_list = kangda._get_robot_group(d["tenantInfoId"], d["token"])
            
    
        

event = CRUDEvent(Event)