590 lines
21 KiB
Python
590 lines
21 KiB
Python
from datetime import datetime
|
||
import enum
|
||
|
||
from sqlalchemy import (
|
||
BIGINT,
|
||
INTEGER,
|
||
JSON,
|
||
Column,
|
||
DateTime,
|
||
Enum,
|
||
ForeignKey,
|
||
Index,
|
||
String,
|
||
Text,
|
||
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="更新时间"
|
||
)
|
||
email = db.Column(db.String(255), nullable=True, comment="邮箱")
|
||
avatar = db.Column(db.String(255), nullable=True, comment="头像（通常存储图片URL）")
|
||
region = db.Column(db.String(255), nullable=True, comment="国家地区")
|
||
phone = db.Column(db.String(255), nullable=True, comment="电话")
|
||
brief = db.Column(db.String(255), nullable=True, comment="个人简介")
|
||
third_party_account = db.Column(
|
||
db.String(255), nullable=True, comment="第三方账号（如微信、QQ等）"
|
||
)
|
||
position = db.Column(db.String(255), nullable=True, comment="职务")
|
||
department = db.Column(db.String(255), nullable=True, comment="部门")
|
||
label = db.Column(
|
||
db.String(255), nullable=True, 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]",
|
||
)
|
||
|
||
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 AccidList(db.Model):
|
||
"""事故工单列表模型类（对应 accid_list 表）"""
|
||
|
||
__tablename__ = "accid_list" # 对应 MySQL 表名
|
||
__table_args__ = (
|
||
# 修正：使用 db.Index（和 OperationHistory 模型一致），移除注释避免兼容问题
|
||
db.Index("idx_ticket_id", "ticket_id"),
|
||
db.Index("idx_creator", "creator"),
|
||
db.Index("idx_create_time", "create_time"),
|
||
db.Index("idx_carline", "carline"),
|
||
db.Index("idx_affects_versions", "affects_versions"),
|
||
db.Index("idx_components", "components"),
|
||
db.Index("idx_labels", "labels"),
|
||
db.Index("idx_fix_versions", "fix_versions"),
|
||
db.Index("idx_accident_level", "accident_level"),
|
||
db.Index("idx_assignee_name", "assignee_name"),
|
||
# 表级属性（参考 Fst 模型的 collation 写法）
|
||
{
|
||
"mysql_charset": "utf8mb4",
|
||
"mysql_collate": "utf8mb4_unicode_ci",
|
||
"mysql_engine": "InnoDB",
|
||
"mysql_row_format": "DYNAMIC",
|
||
},
|
||
)
|
||
|
||
# 核心修正：所有字段使用 db. 前缀（和 User 模型一致）
|
||
id = db.Column(
|
||
db.Integer, primary_key=True, autoincrement=True, comment="主键ID，自增"
|
||
)
|
||
ticket_id = db.Column(db.String(50), nullable=False, comment="工单ID")
|
||
model = db.Column(db.String(100), nullable=True, default=None, comment="型号")
|
||
vin = db.Column(db.String(50), nullable=False, comment="车辆识别码(VIN)")
|
||
creator = db.Column(
|
||
db.String(50), nullable=False, default="", comment="创建人（用户名，如ZHANMEN）"
|
||
)
|
||
create_time = db.Column(db.DateTime, nullable=False, comment="工单新建时间")
|
||
occur_datetime = db.Column(db.DateTime, nullable=False, comment="问题发生时间")
|
||
case_description = db.Column(db.Text, nullable=False, comment="Case描述")
|
||
fo_solution = db.Column(db.Text, nullable=False, comment="Fo解答")
|
||
category = db.Column(db.String(100), nullable=False, default="", comment="分类")
|
||
keywords = db.Column(
|
||
db.String(200), nullable=False, default="", comment="关键词（多个用逗号分隔）"
|
||
)
|
||
handle_status = db.Column(
|
||
db.String(20),
|
||
nullable=True,
|
||
default="已完成",
|
||
comment="处理状态：已完成/未完成",
|
||
)
|
||
# 修正：时间字段默认值用 db.func.current_timestamp()（和 BagFile 模型一致）
|
||
create_at = db.Column(
|
||
db.DateTime,
|
||
nullable=True,
|
||
default=func.current_timestamp(),
|
||
comment="记录插入数据库时间",
|
||
)
|
||
update_at = db.Column(
|
||
db.DateTime,
|
||
nullable=True,
|
||
default=func.current_timestamp(),
|
||
onupdate=func.current_timestamp(),
|
||
comment="记录最后更新时间",
|
||
)
|
||
bmbs_name = db.Column(
|
||
db.String(50), nullable=True, default="", comment="BMBS联系人"
|
||
)
|
||
# 修正：Enum 默认值错误（原 default='' 改为 'P1'），使用 db.Enum
|
||
accident_level = db.Column(
|
||
db.Enum("P1", "P2", "P3"), nullable=True, default="P1", comment="事故评级"
|
||
)
|
||
rdca_reporter = db.Column(
|
||
db.String(50), nullable=True, default="", comment="RDCA报告人"
|
||
)
|
||
carline = db.Column(
|
||
db.String(100),
|
||
nullable=True,
|
||
default=None,
|
||
comment="车系（关联字典表dict_code=carline）",
|
||
)
|
||
software_version = db.Column(
|
||
db.String(50), nullable=True, default="", comment="软件版本"
|
||
)
|
||
labels = db.Column(
|
||
db.String(100),
|
||
nullable=True,
|
||
default=None,
|
||
comment="事故类型标签（关联字典表dict_code=labels）",
|
||
)
|
||
attachment_url = db.Column(
|
||
db.String(255), nullable=True, default="", comment="附件地址（多个用逗号分隔）"
|
||
)
|
||
assignee_name = db.Column(
|
||
db.String(50), nullable=True, default="", comment="经办人（关联字典表/用户表）"
|
||
)
|
||
affects_versions = db.Column(
|
||
db.String(100),
|
||
nullable=True,
|
||
default=None,
|
||
comment="受影响的版本（关联字典表dict_code=affects_versions）",
|
||
)
|
||
components = db.Column(
|
||
db.String(100),
|
||
nullable=True,
|
||
default=None,
|
||
comment="零部件/组件（关联字典表dict_code=components）",
|
||
)
|
||
fix_versions = db.Column(
|
||
db.String(100),
|
||
nullable=True,
|
||
default=None,
|
||
comment="修复方案对应的版本（关联字典表dict_code=fix_versions）",
|
||
)
|
||
subtasks = db.Column(
|
||
db.String(200), nullable=False, default="", comment="子任务（多个用逗号分隔）"
|
||
)
|
||
linked_work_items = db.Column(
|
||
db.String(200),
|
||
nullable=False,
|
||
default="",
|
||
comment="关联的工作项（格式：类型:ID,类型:ID）",
|
||
)
|
||
descriptions = db.Column(
|
||
db.String(255),
|
||
nullable=False,
|
||
default="",
|
||
comment="补充描述（区别于case_description）",
|
||
)
|
||
mviz_link = db.Column(
|
||
db.String(255), nullable=False, default="", comment="mviz的链接"
|
||
)
|
||
system_function = db.Column(
|
||
db.String(255), nullable=False, default="", comment="system_function字段"
|
||
)
|
||
image_ids = db.Column(
|
||
db.String(255),
|
||
nullable=False,
|
||
default="",
|
||
comment="关联图片ID串，逗号分隔（如：1,2,3）",
|
||
)
|
||
|
||
def __repr__(self):
|
||
return f"<AccidList(ticket_id='{self.ticket_id}', vin='{self.vin}')>"
|
||
|
||
def to_dict(self):
|
||
"""转为字典，适配接口返回（参考 DictType 的 to_dict 方法）"""
|
||
return {
|
||
"id": self.id,
|
||
"ticket_id": self.ticket_id,
|
||
"model": self.model,
|
||
"vin": self.vin,
|
||
"creator": self.creator,
|
||
"create_time": (
|
||
self.create_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.create_time
|
||
else None
|
||
),
|
||
"occur_datetime": (
|
||
self.occur_datetime.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.occur_datetime
|
||
else None
|
||
),
|
||
"case_description": self.case_description,
|
||
"fo_solution": self.fo_solution,
|
||
"category": self.category,
|
||
"keywords": self.keywords,
|
||
"handle_status": self.handle_status,
|
||
"create_at": (
|
||
self.create_at.strftime("%Y-%m-%d %H:%M:%S") if self.create_at else None
|
||
),
|
||
"update_at": (
|
||
self.update_at.strftime("%Y-%m-%d %H:%M:%S") if self.update_at else None
|
||
),
|
||
"bmbs_name": self.bmbs_name,
|
||
"accident_level": self.accident_level,
|
||
"rdca_reporter": self.rdca_reporter,
|
||
"carline": self.carline,
|
||
"software_version": self.software_version,
|
||
"labels": self.labels,
|
||
"attachment_url": self.attachment_url,
|
||
"assignee_name": self.assignee_name,
|
||
"affects_versions": self.affects_versions,
|
||
"components": self.components,
|
||
"fix_versions": self.fix_versions,
|
||
"subtasks": self.subtasks,
|
||
"linked_work_items": self.linked_work_items,
|
||
"descriptions": self.descriptions,
|
||
"mviz_link": self.mviz_link,
|
||
"system_function":self.system_function,
|
||
"image_ids":self.image_ids
|
||
}
|
||
|
||
|
||
class DictType(db.Model):
|
||
"""字典类型表模型类（对应 dict_type 表）"""
|
||
|
||
__tablename__ = "dict_type" # 匹配 MySQL 表名
|
||
__table_args__ = (
|
||
# 修正：使用 db.UniqueConstraint（参考 RuleVersions 模型）
|
||
db.UniqueConstraint(
|
||
"dict_code", name="uk_dict_code", comment="字典编码唯一（对应唯一字段）"
|
||
),
|
||
# 表级属性（对齐项目规范）
|
||
{
|
||
"mysql_charset": "utf8mb4",
|
||
"mysql_collate": "utf8mb4_unicode_ci",
|
||
"mysql_engine": "InnoDB",
|
||
"mysql_row_format": "Dynamic",
|
||
"comment": "字典类型表（管理需要下拉的字段）",
|
||
},
|
||
)
|
||
|
||
# 修正：所有字段使用 db. 前缀，类型改为 db.Integer（和 User 模型一致）
|
||
id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment="主键ID")
|
||
dict_code = db.Column(
|
||
db.String(50),
|
||
nullable=False,
|
||
comment="字典编码（对应accid_list的字段名，如carline、labels）",
|
||
)
|
||
dict_name = db.Column(
|
||
db.String(50),
|
||
nullable=False,
|
||
comment="字典名称（字段中文名称，如车系、事故类型）",
|
||
)
|
||
remark = db.Column(
|
||
db.String(200),
|
||
nullable=True,
|
||
default="",
|
||
comment="备注（如“事故工单-车系下拉选项”）",
|
||
)
|
||
status = db.Column(
|
||
db.Integer, nullable=False, default=1, comment="状态：1-启用，0-禁用"
|
||
)
|
||
create_time = db.Column(
|
||
db.DateTime, nullable=True, default=func.current_timestamp(), comment="创建时间"
|
||
)
|
||
update_time = db.Column(
|
||
db.DateTime,
|
||
nullable=True,
|
||
default=func.current_timestamp(),
|
||
onupdate=func.current_timestamp(),
|
||
comment="更新时间",
|
||
)
|
||
|
||
def __repr__(self):
|
||
return f"<DictType(dict_code='{self.dict_code}', dict_name='{self.dict_name}', status={self.status})>"
|
||
|
||
def to_dict(self):
|
||
"""将模型对象转为字典，便于接口返回数据"""
|
||
return {
|
||
"id": self.id,
|
||
"dict_code": self.dict_code,
|
||
"dict_name": self.dict_name,
|
||
"remark": self.remark,
|
||
"status": self.status,
|
||
"create_time": (
|
||
self.create_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.create_time
|
||
else None
|
||
),
|
||
"update_time": (
|
||
self.update_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.update_time
|
||
else None
|
||
),
|
||
}
|
||
|
||
|
||
class DictItem(db.Model):
|
||
"""字典项表模型类（对应 dict_item 表）"""
|
||
|
||
__tablename__ = "dict_item" # 严格匹配 MySQL 表名
|
||
__table_args__ = (
|
||
# 修正：使用 db.Index（对齐 OperationHistory 模型）
|
||
db.Index("idx_dict_id", "dict_id"),
|
||
db.Index("idx_item_value", "item_value"),
|
||
# 表级属性（对齐项目规范）
|
||
{
|
||
"mysql_charset": "utf8mb4",
|
||
"mysql_collate": "utf8mb4_unicode_ci",
|
||
"mysql_engine": "InnoDB",
|
||
"mysql_row_format": "Dynamic",
|
||
"comment": "字典项表（管理每个字段的具体下拉选项）",
|
||
},
|
||
)
|
||
|
||
# 修正：所有字段使用 db. 前缀，类型改为 db.Integer（和 User 模型一致）
|
||
id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment="主键ID")
|
||
# 修正：外键使用 db.ForeignKey（参考 BagFile 模型）
|
||
dict_id = db.Column(
|
||
db.Integer,
|
||
db.ForeignKey("dict_type.id"),
|
||
nullable=False,
|
||
comment="关联字典类型表的ID",
|
||
)
|
||
item_value = db.Column(
|
||
db.String(100),
|
||
nullable=False,
|
||
comment="选项值（存储到accid_list的实际值，如“大众帕萨特”“P1”）",
|
||
)
|
||
item_label = db.Column(
|
||
db.String(100),
|
||
nullable=False,
|
||
comment="选项标签（前端显示的文本，如“大众帕萨特（车系）”）",
|
||
)
|
||
sort = db.Column(
|
||
db.Integer,
|
||
nullable=True,
|
||
default=0,
|
||
comment="排序号（前端下拉框显示顺序，数字越小越靠前）",
|
||
)
|
||
status = db.Column(
|
||
db.Integer,
|
||
nullable=False,
|
||
default=1,
|
||
comment="状态：1-启用，0-禁用（禁用后前端不显示）",
|
||
)
|
||
remark = db.Column(
|
||
db.String(200),
|
||
nullable=True,
|
||
default="",
|
||
comment="备注（如“2025款帕萨特专属”）",
|
||
)
|
||
create_time = db.Column(
|
||
db.DateTime, nullable=True, default=func.current_timestamp(), comment="创建时间"
|
||
)
|
||
update_time = db.Column(
|
||
db.DateTime,
|
||
nullable=True,
|
||
default=func.current_timestamp(),
|
||
onupdate=func.current_timestamp(),
|
||
comment="更新时间",
|
||
)
|
||
|
||
# 修正：关系使用 db.relationship + db.backref（参考 User 模型）
|
||
dict_type = db.relationship(
|
||
"DictType", backref=db.backref("dict_items", lazy="joined"), lazy="joined"
|
||
)
|
||
|
||
def __repr__(self):
|
||
return f"<DictItem(dict_id={self.dict_id}, item_value='{self.item_value}', item_label='{self.item_label}', status={self.status})>"
|
||
|
||
def to_dict(self):
|
||
"""将模型对象转为字典，适配接口返回数据"""
|
||
return {
|
||
"id": self.id,
|
||
"dict_id": self.dict_id,
|
||
"item_value": self.item_value,
|
||
"item_label": self.item_label,
|
||
"sort": self.sort,
|
||
"status": self.status,
|
||
"remark": self.remark,
|
||
"create_time": (
|
||
self.create_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.create_time
|
||
else None
|
||
),
|
||
"update_time": (
|
||
self.update_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.update_time
|
||
else None
|
||
),
|
||
# 关联返回字典类型信息（可选，提升接口易用性）
|
||
"dict_code": self.dict_type.dict_code if self.dict_type else None,
|
||
"dict_name": self.dict_type.dict_name if self.dict_type else None,
|
||
}
|
||
|
||
|
||
class FileUpload(db.Model):
|
||
"""通用文件上传元信息表模型类（对应 file_uploads 表）"""
|
||
|
||
__tablename__ = "file_uploads" # 对应 MySQL 表名
|
||
__table_args__ = (
|
||
# 索引定义（与建表语句完全一致）
|
||
db.Index("idx_business_type_id", "business_type", "business_id"),
|
||
db.Index("idx_create_time", "create_time"),
|
||
db.Index("idx_status", "status"),
|
||
# 表级属性（与 AccidList 模型格式一致）
|
||
{
|
||
"mysql_charset": "utf8mb4",
|
||
"mysql_collate": "utf8mb4_unicode_ci",
|
||
"mysql_engine": "InnoDB",
|
||
"mysql_row_format": "DYNAMIC",
|
||
},
|
||
)
|
||
|
||
# 核心修正：全用 db.Integer，通过 info 指定 MySQL 底层类型（无参数报错）
|
||
id = db.Column(
|
||
db.Integer, # 模型层用 int 兼容
|
||
primary_key=True,
|
||
autoincrement=True,
|
||
info={'mysql_type': 'BIGINT UNSIGNED'}, # 数据库底层仍是 BIGINT UNSIGNED
|
||
comment="主键（自增数值ID）",
|
||
)
|
||
file_name = db.Column(
|
||
db.String(255), nullable=False, comment="原始文件名（如：事故图片1.jpg）"
|
||
)
|
||
file_alias = db.Column(
|
||
db.String(255),
|
||
nullable=False,
|
||
comment="存储别名（避免重名，如：20260210_153950_89782.jpg）",
|
||
)
|
||
file_path = db.Column(
|
||
db.String(512),
|
||
nullable=False,
|
||
comment="存储路径/访问URL（本地：/uploads/202602/xxx.jpg；OSS：https://xxx.oss-cn-beijing.aliyuncs.com/xxx.jpg）",
|
||
)
|
||
file_size = db.Column(
|
||
db.Integer, # 模型层用 int 兼容
|
||
nullable=False,
|
||
info={'mysql_type': 'BIGINT UNSIGNED'}, # 数据库底层 BIGINT UNSIGNED
|
||
comment="文件大小（字节）"
|
||
)
|
||
file_type = db.Column(
|
||
db.String(50),
|
||
nullable=False,
|
||
comment="文件MIME类型（如：image/jpeg、image/png）",
|
||
)
|
||
business_type = db.Column(
|
||
db.String(50),
|
||
nullable=True,
|
||
default=None,
|
||
comment="关联业务类型（如：case=案例、user=用户头像）",
|
||
)
|
||
business_id = db.Column(
|
||
db.Integer, # 模型层用 int 兼容
|
||
nullable=True,
|
||
default=None,
|
||
info={'mysql_type': 'BIGINT UNSIGNED'}, # 数据库底层 BIGINT UNSIGNED
|
||
comment="关联业务主键（案例ID/用户ID等，初始为NULL）",
|
||
)
|
||
create_time = db.Column(
|
||
db.DateTime,
|
||
nullable=False,
|
||
default=func.current_timestamp(), # 对应 DEFAULT CURRENT_TIMESTAMP
|
||
comment="上传时间",
|
||
)
|
||
create_by = db.Column(
|
||
db.String(50), nullable=True, default=None, comment="上传人（用户ID/用户名）"
|
||
)
|
||
status = db.Column(
|
||
db.Integer, # 模型层用 int 兼容
|
||
nullable=False,
|
||
default=1,
|
||
info={'mysql_type': 'TINYINT(3) UNSIGNED'}, # 数据库底层 TINYINT(3) UNSIGNED
|
||
comment="状态：1=有效，0=逻辑删除",
|
||
)
|
||
|
||
def __repr__(self):
|
||
"""模型字符串表示，便于调试"""
|
||
return f"<FileUpload(id='{self.id}', file_name='{self.file_name}', business_type='{self.business_type}')>"
|
||
|
||
def to_dict(self):
|
||
"""转为字典，适配接口返回（与 AccidList 模型风格一致）"""
|
||
return {
|
||
"id": self.id,
|
||
"file_name": self.file_name,
|
||
"file_alias": self.file_alias,
|
||
"file_path": self.file_path,
|
||
"file_size": self.file_size,
|
||
"file_type": self.file_type,
|
||
"business_type": self.business_type,
|
||
"business_id": self.business_id,
|
||
"create_time": (
|
||
self.create_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if self.create_time
|
||
else None
|
||
),
|
||
"create_by": self.create_by,
|
||
"status": self.status,
|
||
"status_text": "有效" if self.status == 1 else "已删除",
|
||
} |