50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
|
from sqlalchemy.exc import SQLAlchemyError
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
from app.models import EventSource, TaggingEvents, VerdictStatus
|
|||
|
|
|
|||
|
|
def create_tagging_event(db, **kwargs):
|
|||
|
|
"""
|
|||
|
|
创建并提交TaggingEvents记录
|
|||
|
|
参数:
|
|||
|
|
db: SQLAlchemy数据库实例
|
|||
|
|
kwargs: 字段键值对，支持所有非主键字段
|
|||
|
|
|
|||
|
|
返回: (success: bool, message: str, event: TaggingEvents)
|
|||
|
|
"""
|
|||
|
|
# 1. 创建事件对象并设置属性
|
|||
|
|
event = TaggingEvents()
|
|||
|
|
|
|||
|
|
# 2. 动态设置字段值
|
|||
|
|
valid_fields = {
|
|||
|
|
'bag_id', 'source', 'rule_version_id', 'reviewer_id', 'verdict',
|
|||
|
|
'level1_tag_id', 'level2_tag_id', 'level3_tag_id', 'level4_tag_id'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for field, value in kwargs.items():
|
|||
|
|
if field in valid_fields:
|
|||
|
|
setattr(event, field, value)
|
|||
|
|
# 自动处理特殊字段类型[1,4](@ref)
|
|||
|
|
elif field == 'ts_event' and isinstance(value, datetime):
|
|||
|
|
event.ts_event = value
|
|||
|
|
|
|||
|
|
# 3. 枚举字段验证[1](@ref)
|
|||
|
|
if 'source' in kwargs and not isinstance(kwargs['source'], EventSource):
|
|||
|
|
return False, "source必须是EventSource枚举类型", None
|
|||
|
|
|
|||
|
|
if 'verdict' in kwargs and not isinstance(kwargs['verdict'], VerdictStatus):
|
|||
|
|
return False, "verdict必须是VerdictStatus枚举类型", None
|
|||
|
|
|
|||
|
|
# 4. 提交到数据库[2,4](@ref)
|
|||
|
|
try:
|
|||
|
|
db.session.add(event)
|
|||
|
|
db.session.commit()
|
|||
|
|
return True, "记录创建成功", event
|
|||
|
|
except SQLAlchemyError as e:
|
|||
|
|
db.session.rollback()
|
|||
|
|
# app.logger.error(f"数据库写入失败: {str(e)}")
|
|||
|
|
return False, f"数据库错误: {str(e)}", None
|
|||
|
|
except Exception as e:
|
|||
|
|
db.session.rollback()
|
|||
|
|
# app.logger.error(f"未知错误: {str(e)}")
|
|||
|
|
return False, f"系统错误: {str(e)}", None
|