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