import uuid from datetime import datetime from sqlalchemy import String, Boolean, Float, ForeignKey, Index, func, Text, DateTime from sqlalchemy import Uuid, JSON from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.types import TypeDecorator from app.database import Base class JSONType(TypeDecorator): """A JSON type that uses JSONB on PostgreSQL and JSON on other databases.""" impl = JSON cache_ok = True def load_dialect_impl(self, dialect): if dialect.name == "postgresql": return dialect.type_descriptor(JSONB()) return dialect.type_descriptor(JSON()) class Alert(Base): """告警记录表""" __tablename__ = "alerts" id: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4, ) brand_id: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), ForeignKey("brands.id", ondelete="CASCADE"), nullable=False, ) user_id: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), nullable=False, index=True, ) alert_type: Mapped[str] = mapped_column( String(50), nullable=False, comment="告警类型: score_drop / score_rise / negative_sentiment / competitor_overtake / new_platform_mention", ) severity: Mapped[str] = mapped_column( String(20), nullable=False, comment="严重程度: critical / warning / info", ) title: Mapped[str] = mapped_column(String(200), nullable=False) message: Mapped[str] = mapped_column(Text, nullable=False) data: Mapped[dict | None] = mapped_column( JSONType, nullable=True, comment="告警相关数据(JSON)", ) is_read: Mapped[bool] = mapped_column( Boolean, default=False, nullable=False, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) brand: Mapped["Brand"] = relationship("Brand") # noqa: F821 __table_args__ = ( Index("idx_alerts_user_id", "user_id"), Index("idx_alerts_brand_id", "brand_id"), Index("idx_alerts_alert_type", "alert_type"), Index("idx_alerts_is_read", "is_read"), Index("idx_alerts_created_at", "created_at"), Index("idx_alerts_user_read", "user_id", "is_read"), )