backend/models.py - master - NoWay1co/AudioAnalytics

backend/models.py
296 строк12 KB
Ivan
Сделан рефакторинг модели RiskTag, чтобы использовать список по умолчанию для столбца фраз. Обновлена логика обновления тегов риска, чтобы явно помечать фразы как измененные. Улучшена ScriptsSettingsPage для обработки ожидающих фраз перед сохранением.
07 фев 2026, 09:38
07 фев 2026, 09:3835eaeeb
"""
SQLAlchemy ORM models.
Maps Python classes to PostgreSQL tables.
Based on PostgreSQL/new_script.sql schema.
"""

from sqlalchemy import (
    Column, String, Text, Integer, BigInteger, Boolean,
    DateTime, Date, Numeric, ForeignKey, DECIMAL
)
from sqlalchemy.dialects.postgresql import UUID, JSONB, INET
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
import uuid

from database import Base


# ============== Core Models ==============

class User(Base):
    """User model - system users (owners, managers, admins)."""
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(100), unique=True, nullable=False)
    password_hash = Column(String(255))
    first_name = Column(String(50))
    last_name = Column(String(50))
    phone = Column(String(20))
    role = Column(String(30))  # owner, manager, admin
    organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id", ondelete="CASCADE"))
    salon_id = Column(UUID(as_uuid=True), ForeignKey("salons.id", ondelete="CASCADE"))
    is_active = Column(Boolean, default=True)
    is_verified = Column(Boolean, default=False)
    created_at = Column(DateTime, server_default=func.now())
    last_login_at = Column(DateTime, server_default=func.now())

    # Relationships
    organization = relationship("Organization", back_populates="users", foreign_keys=[organization_id])
    salon = relationship("Salon", back_populates="users", foreign_keys=[salon_id])

    @property
    def full_name(self) -> str:
        """Get full name of user."""
        parts = [self.first_name, self.last_name]
        return " ".join(filter(None, parts)) or self.email


class Organization(Base):
    """Organization model - companies that own salons."""
    __tablename__ = "organizations"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String(255), nullable=False)
    legal_name = Column(String(255), nullable=False)
    inn = Column(String(20), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    phone = Column(String(20))
    owner_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"))
    settings = Column(JSONB, default={})
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    users = relationship("User", back_populates="organization", foreign_keys=[User.organization_id])
    salons = relationship("Salon", back_populates="organization")
    recordings = relationship("Recording", back_populates="organization")


class Salon(Base):
    """Salon model - physical locations/branches."""
    __tablename__ = "salons"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id", ondelete="CASCADE"))
    name = Column(String(255), nullable=False)
    address = Column(Text, nullable=False)
    city = Column(String(50), nullable=False)
    phone = Column(String(20))
    manager_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"))
    working_hours = Column(JSONB, default={})
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    organization = relationship("Organization", back_populates="salons")
    users = relationship("User", back_populates="salon", foreign_keys=[User.salon_id])
    employees = relationship("Employee", back_populates="salon")
    recordings = relationship("Recording", back_populates="salon")


class Employee(Base):
    """Employee model - salon staff members."""
    __tablename__ = "employees"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    salon_id = Column(UUID(as_uuid=True), ForeignKey("salons.id", ondelete="CASCADE"))
    email = Column(String(100), unique=True, nullable=False)
    first_name = Column(String(50))
    last_name = Column(String(50))
    phone = Column(String(20))
    telegram_id = Column(String(20))
    position = Column(String(30))
    is_active = Column(Boolean, default=True)
    avg_quality_score = Column(DECIMAL(5, 2))
    avg_sentiment = Column(DECIMAL(3, 2))
    total_recordings = Column(Integer, default=0)
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    salon = relationship("Salon", back_populates="employees")
    recordings = relationship("Recording", back_populates="employee")

    @property
    def full_name(self) -> str:
        """Get full name of employee."""
        parts = [self.first_name, self.last_name]
        return " ".join(filter(None, parts)) or self.email


# ============== Recording & Analysis Models ==============

class Recording(Base):
    """Recording model - audio recordings of conversations."""
    __tablename__ = "recordings"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id", ondelete="CASCADE"))
    salon_id = Column(UUID(as_uuid=True), ForeignKey("salons.id", ondelete="CASCADE"))
    employee_id = Column(UUID(as_uuid=True), ForeignKey("employees.id", ondelete="CASCADE"))
    file_url = Column(String(255))
    file_size_bytes = Column(BigInteger)
    duration_seconds = Column(Integer)
    channel = Column(String(50))
    source = Column(String(20))
    recorded_at = Column(DateTime)
    processing_status = Column(String(20), default="pending")  # pending, transcribing, analyzing, done, error
    processing_error = Column(Text)
    is_analyzed = Column(Boolean, default=False)
    has_risk = Column(Boolean, default=False)
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    organization = relationship("Organization", back_populates="recordings")
    salon = relationship("Salon", back_populates="recordings")
    employee = relationship("Employee", back_populates="recordings")
    transcript = relationship("Transcript", back_populates="recording", uselist=False)
    analysis = relationship("Analysis", back_populates="recording", uselist=False)


class Transcript(Base):
    """Transcript model - text transcription of recordings."""
    __tablename__ = "transcripts"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    recording_id = Column(UUID(as_uuid=True), ForeignKey("recordings.id", ondelete="CASCADE"), unique=True)
    full_text = Column(Text)
    segments = Column(JSONB, default=[])
    language = Column(String(30))
    confidence = Column(DECIMAL(3, 2))
    word_count = Column(Integer)
    model_version = Column(String(10))
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    recording = relationship("Recording", back_populates="transcript")


class Analysis(Base):
    """Analysis model - AI analysis results."""
    __tablename__ = "analyses"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    transcript_id = Column(UUID(as_uuid=True), ForeignKey("transcripts.id", ondelete="CASCADE"))
    script_id = Column(UUID(as_uuid=True), ForeignKey("scripts.id", ondelete="CASCADE"))
    script_version = Column(Integer)
    quality_score = Column(Integer)  # 0-100
    compliance_score = Column(Integer)
    sentiment_score = Column(DECIMAL(3, 2))
    sentiment_class = Column(String(30))  # positive, neutral, negative
    summary = Column(Text)
    key_moments = Column(JSONB, default=[])
    risk_type = Column(String(30))
    risk_severity = Column(String(30))
    checklist = Column(JSONB, default=[])
    recommendations = Column(Text)                   # LLM recommendations text
    topics = Column(JSONB, default=[])               # Detected conversation topics
    conversation_count = Column(Integer, default=1)  # Number of conversations in recording
    model_version = Column(String(10))
    tokens_used = Column(Integer)
    created_at = Column(DateTime, server_default=func.now())

    # Add recording relationship through transcript
    recording_id = Column(UUID(as_uuid=True), ForeignKey("recordings.id", ondelete="CASCADE"), unique=True)
    recording = relationship("Recording", back_populates="analysis")


# ============== Script Models ==============

class Script(Base):
    """Script model - conversation scripts/templates."""
    __tablename__ = "scripts"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id", ondelete="CASCADE"))
    salon_id = Column(UUID(as_uuid=True), ForeignKey("salons.id", ondelete="CASCADE"))
    name = Column(String(50))
    description = Column(Text)
    type = Column(String(30))
    current_version = Column(Integer, default=1)
    is_active = Column(Boolean, default=True)
    is_default = Column(Boolean, default=False)
    created_by = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"))
    created_at = Column(DateTime, server_default=func.now())

    # Script analysis fields
    requirements_text = Column(Text)        # Human-readable requirements written by manager/admin
    generated_prompt = Column(Text)         # Auto-generated LLM system prompt from requirements
    prompt_generated_at = Column(DateTime)  # Timestamp of last prompt generation

    # Relationships
    steps = relationship("ScriptStep", back_populates="script")


class ScriptStep(Base):
    """Script step model - individual steps in a script."""
    __tablename__ = "script_steps"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    script_id = Column(UUID(as_uuid=True), ForeignKey("scripts.id", ondelete="CASCADE"))
    order_index = Column(Integer)
    label = Column(String(50))
    description = Column(Text)
    weight = Column(Integer)
    is_required = Column(Boolean, default=True)
    keywords = Column(JSONB, default=[])  # Keywords for step detection by LLM
    created_at = Column(DateTime, server_default=func.now())

    # Relationships
    script = relationship("Script", back_populates="steps")


# ============== Risk Tag Models ==============

class RiskTag(Base):
    """Configurable risk tags for conversation analysis alerts."""
    __tablename__ = "risk_tags"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    organization_id = Column(
        UUID(as_uuid=True),
        ForeignKey("organizations.id", ondelete="CASCADE"),
        nullable=True,
    )
    name = Column(String(100), nullable=False)
    description = Column(Text, nullable=True)
    phrases = Column(JSONB, default=list)
    category = Column(String(50), nullable=True)
    color = Column(String(20), default="slate")
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())


# ============== Subscription & Billing Models ==============

class SubscriptionPlan(Base):
    """Subscription plan model."""
    __tablename__ = "subscription_plans"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String(255))
    description = Column(Text)
    max_salons = Column(Integer)
    max_employees = Column(Integer)
    max_recordings_per_month = Column(Integer)
    price_monthly = Column(DECIMAL(10, 2))
    price_yearly = Column(DECIMAL(10, 2))
    currency = Column(String(30))
    features = Column(JSONB, default=[])
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())


# ============== Session Model ==============

class Session(Base):
    """Session model - user sessions."""
    __tablename__ = "sessions"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"))
    token_hash = Column(String(255))
    ip_address = Column(INET)
    expires_at = Column(DateTime)
    created_at = Column(DateTime, server_default=func.now())