letteka
f19ccf349b
Reviewed-on: #29 Co-authored-by: Andrew Kettel <andrew.kettel@gmail.com> Co-committed-by: Andrew Kettel <andrew.kettel@gmail.com>
118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from sqlalchemy import Boolean, DateTime, Integer, String
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
db: SQLAlchemy = SQLAlchemy(model_class=Base)
|
|
|
|
|
|
class CameraStatus(Base):
|
|
__tablename__ = "camera_status"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
running: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
default=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
@staticmethod
|
|
def get() -> CameraStatus:
|
|
"""Get the single status row, creating it if it doesn't exist."""
|
|
status = db.session.get(CameraStatus, 1)
|
|
if status is None:
|
|
status = CameraStatus(id=1, running=False)
|
|
db.session.add(status)
|
|
db.session.commit()
|
|
return status
|
|
|
|
@staticmethod
|
|
def set_running(running: bool) -> CameraStatus:
|
|
status = CameraStatus.get()
|
|
status.running = running
|
|
status.updated_at = datetime.now(UTC)
|
|
db.session.commit()
|
|
return status
|
|
|
|
|
|
class CameraEvent(Base):
|
|
__tablename__ = "camera_events"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
action: Mapped[str] = mapped_column(String(10), nullable=False) # 'start' | 'stop'
|
|
ip_address: Mapped[str] = mapped_column(String(45), nullable=False)
|
|
timestamp: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
default=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
@staticmethod
|
|
def log(action: str, ip_address: str) -> CameraEvent:
|
|
event = CameraEvent(action=action, ip_address=ip_address)
|
|
db.session.add(event)
|
|
db.session.commit()
|
|
return event
|
|
|
|
@staticmethod
|
|
def recent(limit: int = 50) -> list[CameraEvent]:
|
|
return list(
|
|
db.session.execute(
|
|
db.select(CameraEvent)
|
|
.order_by(CameraEvent.timestamp.desc())
|
|
.limit(limit)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
|
|
class CameraRecordingEvent(Base):
|
|
"""Audit log for recording start/stop actions."""
|
|
|
|
__tablename__ = "camera_recording_events"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
action: Mapped[str] = mapped_column(
|
|
String(20), nullable=False
|
|
) # 'record_start' | 'record_stop'
|
|
ip_address: Mapped[str] = mapped_column(String(45), nullable=False)
|
|
file_path: Mapped[str] = mapped_column(String(512), nullable=False, default="")
|
|
timestamp: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
default=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
@staticmethod
|
|
def log(action: str, ip_address: str, file_path: str = "") -> CameraRecordingEvent:
|
|
event = CameraRecordingEvent(
|
|
action=action,
|
|
ip_address=ip_address,
|
|
file_path=file_path,
|
|
)
|
|
db.session.add(event)
|
|
db.session.commit()
|
|
return event
|
|
|
|
@staticmethod
|
|
def recent(limit: int = 50) -> list[CameraRecordingEvent]:
|
|
return list(
|
|
db.session.execute(
|
|
db.select(CameraRecordingEvent)
|
|
.order_by(CameraRecordingEvent.timestamp.desc())
|
|
.limit(limit)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|