ai-meeting-server/app/models/asr.py

import enum
from datetime import datetime
from sqlalchemy.orm import Mapped
from app.models.db import Base
from sqlalchemy import Column, Integer, String, DateTime, Enum, text
from pydantic import BaseModel
from app.errors.base import ParamsError


class AsrRecordStatus(enum.Enum):
    UNKNOW = 0
    CREATED = 1
    PROCESSING = 2
    FINISHED = 3

class AsrRecordTypes(enum.Enum):
    UNKNOW = 0
    OFFLINE = 1
    REALTIME = 2

class AsrRecord(Base):
    __tablename__ = "asr_record"

    id = Column("id",  String(36), primary_key=True)
    task_id = Column("task_id", String(36), nullable=False, default="", index=True)
    meeting_id = Column("meeting_id", String(36), nullable=False, default="", index=True)
    file_url = Column("file_url", String(255), nullable=False, default="")
    type = Column("type", Enum(AsrRecordTypes), nullable=False, default=AsrRecordTypes.UNKNOW)
    status = Column("status", Enum(AsrRecordStatus), nullable=False, default=AsrRecordStatus.UNKNOW)
    result_url = Column("result_url", String(255), nullable=False, default="")
    created_at = Column(
        "created_at",
        DateTime(timezone=True),
        nullable=False,
        default=datetime.now,
        server_default=text("CURRENT_TIMESTAMP"),
    )
    updated_at = Column(
        "updated_at",
        DateTime(timezone=True),
        nullable=False,
        default=datetime.now,
        server_default=text("CURRENT_TIMESTAMP"),
        onupdate=datetime.now,
        server_onupdate=text("CURRENT_TIMESTAMP"),
    )

class AsrRecordModel(BaseModel):
    id: int = 0
    task_id: str = ""
    file_url: str = ""
    meeting_id: str = ""
    type: AsrRecordTypes = 0
    status: AsrRecordStatus = 0
    result_url: str = ""
    hotwords: str = ""

    def get_create(self):
        if self.task_id and self.file_url:
            return self
        else:
            raise ParamsError(message="Task ID and file URL can't be empty")

    def get_update(self):
        if self.id and (self.task_id or self.file_url or self.type or self.status or self.result_url):
            return self
        else:
            raise ParamsError(message="ID and at least one of task ID, file URL, type, status, or result URL can't be empty")

    def get_delete(self):
        if self.id:
            return self
        else:
            raise ParamsError(message="ID can't be empty")