flask_rulebase_serve/app/models.py

from datetime import datetime
import enum

from sqlalchemy import JSON, DateTime, Enum, ForeignKey, Index, func
from app import db


class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password = db.Column(db.String(255), nullable=False)
    status = db.Column(db.Integer, default=1, comment="用户状态：1-正常，0-禁用")
    created_at = db.Column(db.DateTime, default=datetime.now, comment="创建时间")
    updated_at = db.Column(
        db.DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间"
    )

    # 自引用外键
    created_by = db.Column(db.Integer, db.ForeignKey("users.id"))
    updated_by = db.Column(db.Integer, db.ForeignKey("users.id"))

    # 关系定义
    creator = db.relationship(
        "User", foreign_keys=[created_by], remote_side=[id], backref="created_users"
    )
    updater = db.relationship(
        "User", foreign_keys=[updated_by], remote_side=[id], backref="updated_users"
    )

    # 关联关系
    roles = db.relationship(
        "Role",
        secondary="user_roles",
        back_populates="users",
        foreign_keys="[UserRole.user_id, UserRole.role_id]",
    )  # 明确指定外键)
    created_roles = db.relationship(
        "Role", backref="creator", foreign_keys="Role.created_by"
    )
    updated_roles = db.relationship(
        "Role", backref="updater", foreign_keys="Role.updated_by"
    )

    def __repr__(self):
        return f"<User {self.username}>"


class Role(db.Model):
    __tablename__ = "roles"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(50), unique=True, nullable=False, comment="角色名称")
    description = db.Column(db.String(255), comment="角色描述")
    status = db.Column(db.Integer, default=1, comment="状态：1-启用，0-禁用")
    created_at = db.Column(db.DateTime, default=datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)

    # 外键字段
    created_by = db.Column(db.Integer, db.ForeignKey("users.id"))
    updated_by = db.Column(db.Integer, db.ForeignKey("users.id"))

    # 关系定义
    users = db.relationship(
        "User",
        secondary="user_roles",
        back_populates="roles",
        foreign_keys="[UserRole.role_id, UserRole.user_id]",
    )
    permissions = db.relationship(
        "Permission", secondary="role_permissions", back_populates="roles"
    )

    def __repr__(self):
        return f"<Role {self.name}>"


class UserRole(db.Model):
    __tablename__ = "user_roles"

    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), primary_key=True)
    role_id = db.Column(db.Integer, db.ForeignKey("roles.id"), primary_key=True)
    created_at = db.Column(db.DateTime, default=datetime.now)
    created_by = db.Column(db.Integer, db.ForeignKey("users.id"))

    # 关系定义
    user = db.relationship("User", foreign_keys=[user_id], backref="role_assignments")
    role = db.relationship("Role", foreign_keys=[role_id], backref="user_assignments")
    creator_rel = db.relationship(
        "User", foreign_keys=[created_by], backref="created_assignments"
    )

    def __repr__(self):
        return f"<UserRole user={self.user_id} role={self.role_id}>"


class Permission(db.Model):
    __tablename__ = "permissions"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    code = db.Column(
        db.String(100), unique=True, nullable=False, comment="权限码（资源:操作）"
    )
    description = db.Column(db.String(255), comment="权限描述")
    category = db.Column(db.String(50), comment="权限分类（如用户管理、系统设置）")
    created_at = db.Column(db.DateTime, default=datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)

    # 外键字段
    created_by = db.Column(db.Integer, db.ForeignKey("users.id"))
    updated_by = db.Column(db.Integer, db.ForeignKey("users.id"))

    # 关系定义
    roles = db.relationship(
        "Role", secondary="role_permissions", back_populates="permissions"
    )
    creator = db.relationship(
        "User", foreign_keys=[created_by], backref="created_permissions"
    )
    updater = db.relationship(
        "User", foreign_keys=[updated_by], backref="updated_permissions"
    )

    def __repr__(self):
        return f"<Permission {self.code}>"


class RolePermission(db.Model):
    __tablename__ = "role_permissions"

    role_id = db.Column(db.Integer, db.ForeignKey("roles.id"), primary_key=True)
    permission_id = db.Column(
        db.Integer, db.ForeignKey("permissions.id"), primary_key=True
    )
    created_at = db.Column(db.DateTime, default=datetime.now)
    created_by = db.Column(db.Integer, db.ForeignKey("users.id"))

    # 关系定义
    role = db.relationship(
        "Role", foreign_keys=[role_id], backref="permission_assignments"
    )
    permission = db.relationship(
        "Permission", foreign_keys=[permission_id], backref="role_assignments"
    )
    creator = db.relationship(
        "User", foreign_keys=[created_by], backref="created_role_permissions"
    )

    def __repr__(self):
        return f"<RolePermission role={self.role_id} perm={self.permission_id}>"


class Fst(db.Model):
    __tablename__ = "fst"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255, collation="utf8mb4_unicode_ci"), nullable=False)
    parent_id = db.Column(db.Integer, db.ForeignKey("fst.id"))
    update_time = db.Column(
        db.DateTime, default=func.current_timestamp(), nullable=False
    )
    reserved_json = db.Column(JSON)
    bag_sum = db.Column(db.Integer)
    level = db.Column(
        db.SmallInteger, comment="标签层级：0最高，1=一级, 2=二级, 3=三级, 4=四级"
    )
    name_cn = db.Column(db.String(255, collation="utf8mb4_unicode_ci"), comment="中文")
    annotation = db.Column(
        db.String(500, collation="utf8mb4_unicode_ci"), comment="标签注释"
    )

    # 自引用关系（树形结构）
    children = db.relationship(
        "Fst", backref=db.backref("parent", remote_side=[id]), lazy="dynamic"
    )

    def __repr__(self):
        return f"<Fst {self.name}>"



class OperationHistory(db.Model):
    __tablename__ = "operation_history"

    id = db.Column(
        db.BigInteger, primary_key=True, autoincrement=True, comment="主键ID"
    )
    user_id = db.Column(db.BigInteger, nullable=False, comment="用户ID")
    username = db.Column(db.String(50), nullable=False, comment="用户名")
    api_path = db.Column(db.String(255), nullable=False, comment="接口路径")
    http_method = db.Column(db.String(10), nullable=False, comment="HTTP方法")
    operation_time = db.Column(db.DateTime, default=datetime.now, comment="操作时间")
    request_params = db.Column(db.Text, comment="请求参数")
    response_code = db.Column(db.Integer, comment="响应状态码")
    ip_address = db.Column(db.String(45), comment="客户端IP地址")
    user_agent = db.Column(db.String(255), comment="用户代理信息")
    operation_result = db.Column(db.Integer, default=1, comment="1=成功,0=失败")
    error_message = db.Column(db.String(255), comment="错误信息")
    trace_id = db.Column(db.String(50), comment="请求追踪ID")

    # 索引（在SQLAlchemy中通过__table_args__定义）
    __table_args__ = (
        db.Index("idx_user_id", "user_id"),
        db.Index("idx_operation_time", "operation_time"),
        db.Index("idx_api_path", "api_path"),
        db.Index("idx_user_operation", "user_id", "operation_time"),
    )

    def __repr__(self):
        return f"<OperationHistory {self.api_path} by {self.username}>"


# ===== 枚举类型定义 =====
class BagStatus(enum.Enum):
    UNPROCESSED = "UNPROCESSED"
    PROCESSED_NOT_A_CANDIDATE = "PROCESSED_NOT_A_CANDIDATE"
    PROCESSED_NOT_REVIEWED = "PROCESSED_NOT_REVIEWED"
    REVIEW_IN_PROGRESS = "REVIEW_IN_PROGRESS"
    RULE_BASED_ACCEPTED = "RULE_BASED_ACCEPTED"
    MANUAL_OVERRIDE_ACCEPTED = "MANUAL_OVERRIDE_ACCEPTED"
    REVIEWED_BUT_INVALID = "REVIEWED_BUT_INVALID"


class EventSource(enum.Enum):
    RULE = "RULE"
    MANUAL = "MANUAL"


class VerdictStatus(enum.Enum):
    ACCEPTED = "ACCEPTED"
    OVERRIDDEN = "OVERRIDDEN"


class OperationResult(enum.Enum):
    SUCCESS = 1
    FAILURE = 0


class QaStatus(enum.Enum):
    QA_NOT_REVIEWED = "QA_NOT_REVIEWED"
    QA_PASSED = "QA_PASSED"
    QA_MODIFY = "QA_MODIFY"
    QA_INVALID = "QA_INVALID"


class SyncStatus(enum.Enum):
    SYNC_NOT_READY = "SYNC_NOT_READY"
    SYNCED = "SYNCED"


# ===== 核心数据模型 =====


class RuleVersions(db.Model):
    """规则版本表"""

    __tablename__ = "rule_versions"

    rule_version_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    ruleset_name = db.Column(db.String(128), nullable=False)
    version_str = db.Column(db.String(32), nullable=False)
    description = db.Column(db.Text)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())

    __table_args__ = (
        db.UniqueConstraint("ruleset_name", "version_str", name="uk_ruleset_version"),
    )


class Reviewers(db.Model):
    """审核人员表"""

    __tablename__ = "reviewers"

    reviewer_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    login_name = db.Column(db.String(64), unique=True, nullable=False)
    display_name = db.Column(db.String(128))
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())


# ===== 业务数据模型 =====
class BagFile(db.Model):
    """BAG文件主表"""

    __tablename__ = "bag_file"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    file_name = db.Column(db.String(255), nullable=False, comment="bag名称")
    event = db.Column(db.String(255), comment="event")
    file_path = db.Column(db.String(500))
    capture_datetime = db.Column(
        db.DateTime, nullable=False, comment="采集时间（含时分秒）"
    )
    vehicle_status = db.Column(JSON, comment="车辆状态JSON")
    user_id = db.Column(db.Integer, ForeignKey("users.id"), comment="操作用户id")
    create_time = db.Column(
        db.DateTime, default=db.func.current_timestamp(), comment="标签的创建时间"
    )
    update_time = db.Column(
        db.DateTime,
        default=db.func.current_timestamp(),
        onupdate=db.func.current_timestamp(),
        comment="标签的更新时间",
    )
    car_state = db.Column(db.SmallInteger, comment="车辆状态: 0=静止, 1=运动")
    frame_url = db.Column(db.String(500), comment="图片url")
    video_url = db.Column(db.String(500), comment="视频url")
    level1_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="一级标签id")
    level2_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="二级标签id")
    level3_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="三级标签id")
    level4_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="四级标签id")
    batch_name = db.Column(db.String(255), comment="批次名称")
    status = db.Column(
        Enum(BagStatus),
        nullable=False,
        default=BagStatus.PROCESSED_NOT_REVIEWED,
        comment="处理的状态",
    )
    last_time = db.Column(db.DateTime, comment="最后处理的时间")
    last_rule_version_id = db.Column(
        db.Integer,
        ForeignKey("rule_versions.rule_version_id"),
        comment="rule_versions的外键",
    )
    front_starttime = db.Column(db.DateTime, comment="前摄像头的开始时间")
    front_endtime = db.Column(db.DateTime, comment="前摄像头的结束时间")
    case_type = db.Column(db.Integer, comment="1简单场景;2复杂场景")
    comment1 = db.Column(db.String(255), comment="注释1")
    comment2 = db.Column(db.String(255), comment="注释2")
    comment3 = db.Column(db.String(255), comment="注释3")
    bag_status = db.Column(db.Integer, comment="标签状态，0开始，1更新")

    # 新增字段
    qa_status = db.Column(
        Enum(QaStatus),
        nullable=False,
        default=QaStatus.QA_NOT_REVIEWED,
        comment="质检的状态",
    )
    sync_status = db.Column(
        Enum(SyncStatus),
        nullable=False,
        default=SyncStatus.SYNC_NOT_READY,
        comment="同步的状态",
    )
    fst_version = db.Column(db.String(255), comment="fst版本")
    front_start_sec = db.Column(db.Integer, comment="前摄的开始秒")
    front_end_sec = db.Column(db.Integer, comment="前摄的结束秒")

    bag_update_time = db.Column(
        db.DateTime, default=db.func.current_timestamp(), comment="bag的更新时间"
    )

    qa_confirm_time = db.Column(
        db.DateTime, default=db.func.current_timestamp(), comment="qa的确认时间"
    )

    db_time = db.Column(
        db.DateTime,
        default=db.func.current_timestamp(),
        comment="更新到远程数据库的时间",
    )

    qa_id = db.Column(db.Integer, ForeignKey("users.id"),comment="操作用户id")

    high_speed = db.Column(db.Integer, comment="高速,1是,0否")
    urban = db.Column(db.Integer, comment="城区,1是,0否")
    parking = db.Column(db.Integer, comment="停车,1是,0否")
    # 关系定义
    user = db.relationship("User", foreign_keys=[user_id],backref="bag_files")
    rule_version = db.relationship("RuleVersions", backref="bag_files")
    level1_tag = db.relationship("Fst", foreign_keys=[level1_tag_id])
    level2_tag = db.relationship("Fst", foreign_keys=[level2_tag_id])
    level3_tag = db.relationship("Fst", foreign_keys=[level3_tag_id])
    level4_tag = db.relationship("Fst", foreign_keys=[level4_tag_id])
    qa_user = db.relationship('User', foreign_keys=[qa_id], backref='qa_bag_files')
    
    
    __table_args__ = (
        db.Index("idx_capture_time", "capture_datetime"),
        db.Index("status", "status"),
        db.Index("idx_bag_file_name", "file_name"),
    )


class TaggingEvents(db.Model):
    """标签事件记录表"""

    __tablename__ = "tagging_events"

    event_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    bag_id = db.Column(
        db.Integer, ForeignKey("bag_file.id"), nullable=False, comment="bag的id"
    )
    source = db.Column(Enum(EventSource), nullable=False, comment="来源")
    rule_version_id = db.Column(
        db.Integer, ForeignKey("rule_versions.rule_version_id"), comment="rule版本id"
    )
    # reviewer_id = db.Column(db.Integer, ForeignKey('reviewers.reviewer_id'),
    #                        comment='审核人id')
    reviewer_id = db.Column(db.Integer, ForeignKey("users.id"), comment="审核人id")
    verdict = db.Column(Enum(VerdictStatus), comment="结论")
    ts_event = db.Column(
        db.DateTime, default=db.func.current_timestamp(), comment="当前时间"
    )
    level1_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="一级标签id")
    level2_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="二级标签id")
    level3_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="三级标签id")
    level4_tag_id = db.Column(db.Integer, ForeignKey("fst.id"), comment="四级标签id")
    case_type = db.Column(db.Integer, comment="1简单场景;2复杂场景")
    front_starttime = db.Column(db.DateTime, comment="前摄像头开始时间")
    front_endtime = db.Column(db.DateTime, comment="前摄像头结束时间")
    front_start_sec = db.Column(db.Integer, comment="前摄的开始秒")
    front_end_sec = db.Column(db.Integer, comment="前摄的结束秒")
    high_speed = db.Column(db.Integer, comment="高速,1是,0否")
    urban = db.Column(db.Integer, comment="城区,1是,0否")
    parking = db.Column(db.Integer, comment="停车,1是,0否")
    qa_status = db.Column(
        Enum(QaStatus),
        nullable=False,
        default=QaStatus.QA_NOT_REVIEWED,
        comment="质检状态",
    )
    is_deleted = db.Column(
        db.Boolean,
        nullable=False,
        default=False,
        comment="soft delete flag",
    )
    note = db.Column(db.Text, comment="备注")

    # 关系定义
    bag = db.relationship("BagFile", backref="tagging_events")
    rule_version = db.relationship("RuleVersions")
    reviewer = db.relationship("User")
    level1_tag = db.relationship("Fst", foreign_keys=[level1_tag_id])
    level2_tag = db.relationship("Fst", foreign_keys=[level2_tag_id])
    level3_tag = db.relationship("Fst", foreign_keys=[level3_tag_id])
    level4_tag = db.relationship("Fst", foreign_keys=[level4_tag_id])

    __table_args__ = (db.Index("idx_evt_bag_ts", "bag_id", "ts_event"),)


class VlmFilter(db.Model):
    __tablename__ = 'vlm_filter'
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment='主键')
    bag_name = db.Column(db.String(255), nullable=True, comment='bag的名称')
    level1_tag_id = db.Column(db.Integer, db.ForeignKey('fst.id', ondelete='CASCADE', onupdate='CASCADE'), 
                             nullable=True, index=True, comment='一级标签id')
    level2_tag_id = db.Column(db.Integer, db.ForeignKey('fst.id', ondelete='CASCADE', onupdate='CASCADE'), 
                             nullable=True, index=True, comment='二级标签id')
    level3_tag_id = db.Column(db.Integer, db.ForeignKey('fst.id', ondelete='CASCADE', onupdate='CASCADE'), 
                             nullable=True, index=True, comment='三级标签id')
    level4_tag_id = db.Column(db.Integer, db.ForeignKey('fst.id', ondelete='CASCADE', onupdate='CASCADE'), 
                             nullable=True, index=True, comment='四级标签id')
    video_url = db.Column(db.String(255), nullable=True, comment='视频路径')
    comment = db.Column(db.String(255), nullable=True, comment='备注')
    user_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE', onupdate='CASCADE'), 
                       nullable=True, index=True, comment='写入的用户')
    collection_time = db.Column(db.DateTime, nullable=True, comment='标签采集时间')
    create_time = db.Column(db.DateTime, nullable=True, comment='写入的时间')
    init_label = db.Column(db.Integer, nullable=True, comment='入库的状态，0是标注，1是质检')
    status = db.Column(db.Integer, nullable=True, comment='入库的状态，0是标注，1是质检')
    
    # 可以根据需要添加关系引用
    level1_tag = db.relationship('Fst', foreign_keys=[level1_tag_id])
    level2_tag = db.relationship('Fst', foreign_keys=[level2_tag_id])
    level3_tag = db.relationship('Fst', foreign_keys=[level3_tag_id])
    level4_tag = db.relationship('Fst', foreign_keys=[level4_tag_id])
    user = db.relationship('User', foreign_keys=[user_id])

class BagMergeRecord(db.Model):
    __tablename__ = "bag_merge_record"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment='主键')
    joined_name = db.Column(
        db.String(255),
        nullable=False,
        index=True,
        comment="合并后 bag 名称（joined_name）"
    )
    src_name = db.Column(
        db.String(255),
        nullable=False,
        index=True,
        comment="参与合并的原始 bag 名称"
    )
    is_initiator = db.Column(
        db.Boolean,
        nullable=False,
        default=False,
        comment="是否为本次合并操作的主 bag（发起合并的那个 src）",
    )
    create_time = db.Column(
        db.DateTime,
        default=db.func.current_timestamp(),
        comment="记录创建时间",
    )
    created_by = db.Column(
        db.Integer,
        db.ForeignKey("users.id"),
        index=True,
        comment="操作用户 id",
    )

    # 可选：反向关系，如果有需要
    creator = db.relationship("User", backref="bag_merge_records")

    __table_args__ = (
        # 如果希望避免“同一个 joined_name + src_name 被插入多次”，可以加一个联合唯一约束：
        # db.UniqueConstraint("joined_name", "src_name", name="uk_joined_src"),
    )