emcp/backend/app/repositories/alarm_repo.py

from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional

from app.db.sqlite import get_connection
from app.schemas.platform import AlarmEvent


class AlarmRepository:
    def save_alarm(self, alarm: AlarmEvent) -> int:
        with get_connection() as connection:
            cursor = connection.execute(
                """
                INSERT INTO alarm_event (event_time, line_code, event_type, event_code, event_value, content)
                VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    alarm.event_time.isoformat(sep=" ", timespec="seconds"),
                    alarm.line_code,
                    alarm.event_type,
                    alarm.event_code,
                    alarm.event_value,
                    alarm.content,
                ),
            )
            connection.commit()
            return int(cursor.lastrowid)

    def list_alarms(
        self,
        page: int = 1,
        size: int = 20,
        line_code: Optional[int] = None,
        event_type: Optional[int] = None,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
    ) -> List[Dict[str, Any]]:
        offset = (page - 1) * size
        conditions = []
        params: List[Any] = []

        if line_code is not None:
            conditions.append("line_code = ?")
            params.append(line_code)
        if event_type is not None:
            conditions.append("event_type = ?")
            params.append(event_type)
        if start_time is not None:
            conditions.append("event_time >= ?")
            params.append(start_time.isoformat(sep=" ", timespec="seconds"))
        if end_time is not None:
            conditions.append("event_time <= ?")
            params.append(end_time.isoformat(sep=" ", timespec="seconds"))

        where_clause = ""
        if conditions:
            where_clause = "WHERE " + " AND ".join(conditions)

        with get_connection() as connection:
            cursor = connection.execute(
                f"""
                SELECT id, event_time, line_code, event_type, event_code, event_value, content
                FROM alarm_event
                {where_clause}
                ORDER BY id DESC
                LIMIT ? OFFSET ?
                """,
                tuple(params + [size, offset]),
            )
            return [dict(row) for row in cursor.fetchall()]