zwf
236 lines
9.9 KiB
Python
236 lines
9.9 KiB
Python
from typing import Optional, Callable
|
|
from paste.web.form import ModelForm
|
|
from paste.core.logging import echo_log
|
|
from wtforms import StringField, IntegerField,TextAreaField
|
|
from wtforms.validators import Length
|
|
from tornado_swagger.model import register_swagger_model
|
|
import models
|
|
from models.common_model import CommonModel
|
|
from models.db_models import TD3iDcmTaskExtendedInfo
|
|
import pandas as pd
|
|
from sqlalchemy import select
|
|
|
|
|
|
class DcmTaskExtendedInfoForm(ModelForm):
|
|
"""
|
|
更多信息表单验证类（完全映射 TD3iDcmTaskExtendedInfo 字段）。
|
|
|
|
用于验证和处理数字城管-部门待办任务扩展信息数据。
|
|
字段完全映射数据库表 t_d3i_dcm_task_extend_info 的字段结构。
|
|
"""
|
|
rec_id=IntegerField('记录ID')
|
|
subtype_id=StringField('子类型ID',validators=[Length(max=50,message='子类型ID长度不能超过50个字符')])
|
|
content_range=StringField('内容范围',validators=[Length(max=255,message='内容范围长度不能超过255个字符')])
|
|
control_type=StringField('控件类型',validators=[Length(max=50,message='控件类型长度不能超过50个字符')])
|
|
data_type_id=StringField('数据类型ID',validators=[Length(max=50,message='数据类型ID长度不能超过50个字符')])
|
|
display_name=StringField('显示名称',validators=[Length(max=100,message='显示名称长度不能超过100个字符')])
|
|
field_id=StringField('字段ID',validators=[Length(max=50,message='字段ID长度不能超过50个字符')])
|
|
field_value=StringField('字段值',validators=[Length(max=255,message='字段值长度不能超过255个字符')])
|
|
list_content=TextAreaField('下拉框选项内容')
|
|
null_flag=StringField('是否可空标识(0:不可空,1:可空)',validators=[Length(max=20,message='标识长度不能超过20个字符')])
|
|
subtype_field_name=StringField('子类型字段名称',validators=[Length(max=100,message='子类型字段名称长度不能超过100个字符')])
|
|
|
|
def process(self, formdata=None, obj=None, **kwargs):
|
|
if formdata:
|
|
for name, values in formdata.items():
|
|
if isinstance(values, list) and values:
|
|
formdata[name] = [v.strip() if isinstance(v, str) else v for v in values]
|
|
elif isinstance(values, str):
|
|
formdata[name] = values.strip()
|
|
super().process(formdata, obj, **kwargs)
|
|
|
|
|
|
class DcmTaskExtendedInfoBase(TD3iDcmTaskExtendedInfo, CommonModel):
|
|
"""
|
|
扩展信息基础类（完全映射 TD3iDcmTaskExtendedInfo 字段）。
|
|
|
|
封装所有与扩展信息相关的通用操作方法。
|
|
"""
|
|
FieldMapping = {
|
|
'rec_id': 'recID',
|
|
'subtype_id': 'subtypeID',
|
|
'content_range': 'contentRange',
|
|
'control_type': 'controlType',
|
|
'data_type_id': 'dataTypeID',
|
|
'display_name': 'displayName',
|
|
'field_id': 'fieldID',
|
|
'field_value': 'fieldValue',
|
|
'list_content': 'listContent',
|
|
'null_flag': 'nullFlag',
|
|
'subtype_field_name': 'subtypeFieldName'
|
|
}
|
|
|
|
@classmethod
|
|
async def exists_rec_id(cls, data_df: pd.DataFrame):
|
|
"""
|
|
查找 data_df 中在数据库中已存在和不存在的记录。仅根据 rec_id 字段判断。
|
|
|
|
:param data_df: 输入的数据框架，必须包含 raw_id（rec_id）列
|
|
:return: (exists_df: pd.DataFrame, latest_df: pd.DataFrame)
|
|
- exists_df: 在数据库中存在的记录（已匹配数据库id）
|
|
- latest_df: 在数据库中不存在的记录
|
|
"""
|
|
if data_df.empty:
|
|
return pd.DataFrame(), pd.DataFrame()
|
|
|
|
# 获取待查询的 rec_id（去重）
|
|
rec_ids = data_df[cls.rec_id.key].drop_duplicates().tolist()
|
|
if not rec_ids:
|
|
return pd.DataFrame(), data_df.copy()
|
|
|
|
# 查询数据库仅根据 rec_id 匹配
|
|
_query = select(cls.id, cls.rec_id).where(
|
|
cls.rec_id.in_(rec_ids)
|
|
)
|
|
exists_df = await cls.query_as_df(_query)
|
|
|
|
if exists_df.empty:
|
|
return pd.DataFrame(), data_df.copy()
|
|
|
|
# 构建 rec_id -> 数据库id 的映射（单字段）
|
|
key_to_id_map = dict(zip(exists_df[cls.rec_id.key], exists_df[cls.id.key]))
|
|
|
|
# 根据 rec_id 判断是否存在
|
|
mask_exists = data_df.apply(lambda row: row[cls.rec_id.key] in key_to_id_map, axis=1)
|
|
|
|
# 拆分存在/不存在的数据
|
|
exists_df = data_df[mask_exists].copy()
|
|
# 通过 rec_id 匹配数据库主键
|
|
exists_df[cls.id.key] = exists_df.apply(lambda row: key_to_id_map[row[cls.rec_id.key]], axis=1)
|
|
latest_df = data_df[~mask_exists].copy()
|
|
|
|
return exists_df, latest_df
|
|
|
|
@classmethod
|
|
async def fill_extend_info(cls, data_df: pd.DataFrame, index_field: str = 'id',
|
|
column_name: str = 'extend_infos',
|
|
preprocessing: Optional[Callable] = None):
|
|
"""
|
|
填充扩展信息数据到数据框架。
|
|
|
|
用于在查询结果中添加关联的扩展信息。
|
|
|
|
:param pandas.DataFrame data_df: 待填充的数据框架
|
|
:param str index_field: 索引字段，一般是任务ID
|
|
:param str column_name: 填充时，新增加的列名称，默认为`extend_info`
|
|
:param preprocessing: 预处理，注意预处理必须要返回处理后的结果
|
|
:return: 扩展信息数据框架（已填充）
|
|
:rtype: pandas.DataFrame
|
|
"""
|
|
if data_df.empty:
|
|
return pd.DataFrame()
|
|
|
|
_task_ids = list(set(data_df[index_field].unique().tolist()))
|
|
if not _task_ids:
|
|
return pd.DataFrame()
|
|
|
|
_query = select(cls).where(cls.dcm_task_id.in_(_task_ids))
|
|
_extend_info_df: pd.DataFrame = await cls.query_as_df(_query)
|
|
if not _extend_info_df.empty:
|
|
_extend_info_df.replace(models.EmptyInDF+models.EmptyDatetimeInDF, '', inplace=True)
|
|
# 整理输出数据类型
|
|
_extend_info_df[cls.id.key] = _extend_info_df[cls.id.key].astype(str)
|
|
_extend_info_df[cls.dcm_task_id.key] = _extend_info_df[cls.dcm_task_id.key].astype(str)
|
|
# 设置索引
|
|
_extend_info_df['index_id'] = _extend_info_df[cls.dcm_task_id.key]
|
|
_extend_info_df.set_index(['index_id'], inplace=True)
|
|
# 对数据进行预处理
|
|
if isinstance(preprocessing, Callable):
|
|
_extend_info_df = preprocessing(_extend_info_df)
|
|
# 增加数据填充列
|
|
data_df[column_name] = data_df[index_field].apply(
|
|
lambda x: _extend_info_df.query(f"{cls.dcm_task_id.key}=='{x}'").to_dict('records')
|
|
)
|
|
else:
|
|
data_df[column_name] = [[] for _ in range(len(data_df))]
|
|
return _extend_info_df
|
|
|
|
|
|
@register_swagger_model
|
|
class DcmTaskExtendedInfo(DcmTaskExtendedInfoBase):
|
|
"""
|
|
扩展信息模型类（主业务类，完全继承 TD3iDcmTaskExtendedInfo 字段）。
|
|
"""
|
|
|
|
@classmethod
|
|
async def create_batch(cls, data_df: pd.DataFrame):
|
|
"""
|
|
批量创建新扩展信息（传入数据应为全新记录，无需校验是否存在）。
|
|
|
|
:param data_df: 包含扩展信息数据的 DataFrame，字段需与模型属性匹配
|
|
:return: 成功创建的记录数量
|
|
:rtype: int
|
|
"""
|
|
if data_df.empty:
|
|
return 0
|
|
|
|
# 一次性转为字典列表（C 层高效）
|
|
records = data_df.to_dict('records')
|
|
|
|
# 用列表推导式构造对象
|
|
records = [cls().copy_from_dict(record, skip_none=True).before_save() for record in records]
|
|
|
|
# 批量插入
|
|
session = cls.get_aio_session()
|
|
try:
|
|
session.add_all(records)
|
|
await session.commit()
|
|
except Exception as e:
|
|
await session.rollback()
|
|
raise e
|
|
finally:
|
|
await session.close()
|
|
echo_log(f"批量创建成功：创建 {len(records)} 条任务扩展信息。")
|
|
return len(records)
|
|
|
|
@classmethod
|
|
async def modify_batch(cls, data_df: pd.DataFrame):
|
|
"""
|
|
批量修改已有扩展信息。
|
|
|
|
:param data_df: 包含扩展信息数据的 DataFrame
|
|
:return: 成功更新的记录数量
|
|
:rtype: int
|
|
"""
|
|
if data_df.empty:
|
|
return 0
|
|
|
|
# 必须包含 id 列
|
|
if 'id' not in data_df.columns:
|
|
echo_log(f"错误：modify_batch 要求输入数据必须包含 '{cls.id.key}' 列（主键）")
|
|
return 0
|
|
|
|
# 转换为字典列表
|
|
update_data = data_df.to_dict('records')
|
|
|
|
# 使用 bulk_update_mappings
|
|
session = cls.get_aio_session()
|
|
try:
|
|
await session.run_sync(
|
|
lambda sync_session: sync_session.bulk_update_mappings(cls, update_data)
|
|
)
|
|
await session.commit()
|
|
updated_count = len(update_data)
|
|
except Exception as e:
|
|
await session.rollback()
|
|
raise e
|
|
finally:
|
|
await session.close()
|
|
|
|
echo_log(f"批量修改成功：更新 {updated_count} 条任务扩展信息。")
|
|
return updated_count
|
|
|
|
@classmethod
|
|
async def save_batch(cls, data_df: pd.DataFrame):
|
|
"""
|
|
批量保存数据，自动处理新建和更新。
|
|
|
|
:param data_df: 要保存的数据框架
|
|
:return: 新建和更新的数量
|
|
"""
|
|
# 筛选数据状态
|
|
_exists_df, _latest_df = await DcmTaskExtendedInfo.exists_rec_id(data_df)
|
|
# 保存到数据库
|
|
_created_count = await DcmTaskExtendedInfo.create_batch(_latest_df)
|
|
_updated_count = await DcmTaskExtendedInfo.modify_batch(_exists_df)
|
|
return _created_count, _updated_count |