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

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


class AIMeetingStatus(enum.Enum):
    UNKNOW = 0
    CREATED = 1
    RECRODING = 2
    PROCESSING = 3
    FINISHED = 4
    FAILED = 5


class AIMeeting(Base):
    __tablename__ = "ai_meeting"

    id = Column("id", String(36), primary_key=True)
    name = Column("name", String(255), nullable=False, default="")
    description = Column(
        "description", String(255), nullable=False, default=""
    )
    status = Column(
        "status", Enum(AIMeetingStatus), nullable=False, default=AIMeetingStatus.UNKNOW
    )
    duration = Column("duration", Integer, nullable=False, default=0)
    file_url = Column("file_url", String(255), nullable=False, default="")
    content = Column("content", TEXT, default="")
    view_content = Column("view_content", TEXT, default=json.dumps([]))
    asr_task_id = Column("asr_task_id", String(36), nullable=False, default="")
    created_time = Column("created_time", Integer, default=int(time.time()))
    created_at = Column(
        "created_at",
        DateTime(timezone=True),
        nullable=False,
        default=datetime.now,
        server_default=text("CURRENT_TIMESTAMP"),
        index=True
    )
    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"),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'status': self.status,
            'duration': self.duration,
            'file_url': self.file_url,
            'view_content': json.loads(self.view_content) if self.view_content else [],
            'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
            'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S'),
        }

    def to_list_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'status': self.status,
            'duration': self.duration,
            'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
            'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S'),
        }

class AIMeetingModel(BaseModel):
    id: str = ""
    name: str = ""
    description: str = ""
    status: AIMeetingStatus = 0
    duration: float = 0
    file_url: str = ""
    asr_task_id: str = ""
    content: dict = {}
    view_content: str= ""
    replace_word: list = []

    def get_create(self):
        if not self.name:
            self.name = datetime.now().strftime("%Y年%m月%d日会议")
            return self
        else:
            return self

    def get_create_offline(self):
        if not self.file_url or not self.duration:
            raise ParamsError(message="Required parameters are missing: file_url")
        if not self.name:
            self.name = datetime.now().strftime("%Y年%m月%d日会议")
            return self
        else:
            return self

    def get_update(self):
        if self.id and (self.name or self.description or self.status or self.duration or self.file_url or self.asr_task_id):
            return self
        else:
            raise ParamsError(message="ID and at least one of name, description, status, duration, file URL, or ASR task ID can't be empty")

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


class AIMeetingRecord(Base):
    __tablename__ = "ai_meeting_record"

    id = Column("id", Integer, primary_key=True, autoincrement=True)
    meeting_id = Column("meeting_id", Integer, nullable=False, default=0, index=True)
    name = Column("name", String(255), nullable=False, default="")
    content = Column("content", TEXT, default="")
    view_content = Column("view_content", TEXT, 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 AIMeetingRecordModel(BaseModel):
    id: int = 0
    meeting_id: int = 0
    name: str = ""
    content: str = ""
    view_content: str = ""
    order: int = 0

    def get_create(self):
        if self.meeting_id and self.name and self.content and self.view_content and self.order:
            return self
        else:
            raise ParamsError(message="Meeting ID, name, content, view content, and order can't be empty")

    def get_update(self):
        if self.id and (self.name or self.content or self.view_content or self.order):
            return self
        else:
            raise ParamsError(message="ID and at least one of name, content, view content, or order can't be empty")

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


class AIMeetingMinutesTemplate(Base):
    __tablename__ = "ai_meeting_minutes_template"

    id = Column("id", Integer, primary_key=True, autoincrement=True)
    name = Column("name", String(255), nullable=False, default="")
    content = Column("content", String(4000), nullable=False, default="")
    description = Column("description", String(255), nullable=False, default="")
    icon_url = Column("icon_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 AIMeetingMinutesTemplateModel(BaseModel):
    id: int = 0
    name: str = ""
    content: str = ""
    description: str = ""
    icon_url: str = ""

    def get_create(self):
        if self.name and self.content:
            return self
        else:
            raise ParamsError(message="Name and content can't be empty")

    def get_update(self):
        if self.id and (self.name or self.content or self.description or self.icon_url):
            return self
        else:
            raise ParamsError(message="ID and at least one of name, content, description, or icon URL can't be empty")

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


class AIMeetingSpeakModel(BaseModel):
    content: str = ""

    def get_create(self):
        if self.content:
            return self
        else:
            raise ParamsError(message="Content can't be empty")

class AIMeetingUser(Base):
    __tablename__ = "ai_meeting_user"

    id = Column("id",  String(32), primary_key=True)
    name = Column("name", String(255), default="")
    created_at = Column(
        "created_at",
        DateTime(timezone=True),
        nullable=False,
        default=datetime.now,
        server_default=text("CURRENT_TIMESTAMP"),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
        }


class AIMeetingUserModel(BaseModel):
    id: str = ""
    name: str = ""

    def get_create(self):
        if not self.id:
            raise ParamsError(message="Content can't be empty")
        elif not self.name:
            self.name = self.id
            return self
        else:
            return self