zwf
601 lines
29 KiB
Python
601 lines
29 KiB
Python
# coding: utf-8
|
||
import datetime
|
||
import random
|
||
from typing import Union
|
||
|
||
import pandas as pd
|
||
from sqlalchemy import select, delete
|
||
from tornado_swagger.model import register_swagger_model
|
||
from wtforms import StringField, TextAreaField, IntegerField, DateTimeField
|
||
from wtforms.validators import Length
|
||
|
||
import models
|
||
from models.common_model import CommonModel
|
||
from models.db_models import TD3iGovsOrderDetail
|
||
from paste.core.logging import echo_log
|
||
from paste.rbac.rbac_user import RbacUser
|
||
from paste.util.pagination import Pagination
|
||
from paste.web.form import ModelForm
|
||
|
||
|
||
class GovsOrderDetailForm(ModelForm):
|
||
"""省12345工单详情表单验证类"""
|
||
|
||
id = IntegerField('详情记录唯一ID')
|
||
master_id = IntegerField('关联工单主表ID')
|
||
order_id = StringField('工单编号', validators=[Length(max=50)])
|
||
order_no = StringField('工单号', validators=[Length(max=50)])
|
||
master_id_origin = StringField('原始主表ID', validators=[Length(max=50)])
|
||
tenant_id = StringField('租户ID', validators=[Length(max=50)])
|
||
tenant_name = StringField('租户名称', validators=[Length(max=50)])
|
||
order_status = StringField('工单状态码', validators=[Length(max=10)])
|
||
order_status_for_view = StringField('工单状态显示值', validators=[Length(max=50)])
|
||
claim_status = StringField('签收状态', validators=[Length(max=10)])
|
||
over_due = StringField('是否超期', validators=[Length(max=10)])
|
||
is_supervise = StringField('是否督办', validators=[Length(max=10)])
|
||
first_order_status = StringField('一级状态编码', validators=[Length(max=10)])
|
||
secord_order_status = StringField('二级状态编码', validators=[Length(max=10)])
|
||
atomic_order_status = StringField('原子状态编码', validators=[Length(max=10)])
|
||
order_invalid_type = TextAreaField('工单作废原因')
|
||
order_finish_time = DateTimeField('工单完成时间')
|
||
case_content = TextAreaField('诉求内容')
|
||
case_goal = TextAreaField('诉求目的')
|
||
title = StringField('工单标题', validators=[Length(max=500)])
|
||
case_labels = TextAreaField('工单标签列表')
|
||
case_public = StringField('是否公开', validators=[Length(max=10)])
|
||
hotspot = StringField('是否热点事件', validators=[Length(max=10)])
|
||
case_is_urgent = StringField('紧急程度', validators=[Length(max=10)])
|
||
case_is_visit = StringField('是否回访', validators=[Length(max=10)])
|
||
info_protect = StringField('信息保护', validators=[Length(max=10)])
|
||
case_accord_type_one_name = StringField('诉求归口一级', validators=[Length(max=50)])
|
||
case_accord_type_two_name = StringField('诉求归口二级', validators=[Length(max=50)])
|
||
case_accord_type_three_name = StringField('诉求归口三级', validators=[Length(max=50)])
|
||
case_accord_type_four_name = StringField('诉求归口四级', validators=[Length(max=50)])
|
||
case_accord_type_five_name = StringField('诉求归口五级', validators=[Length(max=50)])
|
||
case_accord_code = StringField('事项编码', validators=[Length(max=50)])
|
||
first_level_affiliation = TextAreaField('一级归属单位')
|
||
second_level_affiliation = TextAreaField('二级归属单位')
|
||
third_level_affiliation = TextAreaField('三级归属单位')
|
||
fourth_level_affiliation = TextAreaField('四级归属单位')
|
||
fifth_level_affiliation = TextAreaField('五级归属单位')
|
||
sixth_level_affiliation = TextAreaField('六级归属单位')
|
||
seventh_level_affiliation = TextAreaField('七级归属单位')
|
||
appeal_dept = StringField('诉求部门', validators=[Length(max=100)])
|
||
order_source = StringField('诉求来源', validators=[Length(max=50)])
|
||
order_source_detail = StringField('诉求来源详情', validators=[Length(max=50)])
|
||
order_source_for_view = StringField('诉求来源显示值', validators=[Length(max=50)])
|
||
belong_platform = StringField('所属平台代码', validators=[Length(max=50)])
|
||
belong_platform_name = StringField('受理平台名称', validators=[Length(max=50)])
|
||
current_processing_platform = TextAreaField('当前处理平台')
|
||
service_object_type = StringField('服务对象类型', validators=[Length(max=50)])
|
||
order_type = StringField('表单类型', validators=[Length(max=50)])
|
||
form_type = StringField('表单类型代码', validators=[Length(max=50)])
|
||
area_code = StringField('区域代码', validators=[Length(max=10)])
|
||
area_code_city = StringField('市区域代码', validators=[Length(max=50)])
|
||
area_code_area = StringField('区区域代码', validators=[Length(max=50)])
|
||
area_code_street = StringField('街道区域代码', validators=[Length(max=50)])
|
||
address_detail = StringField('详细地址', validators=[Length(max=500)])
|
||
case_lnglat = StringField('地理坐标', validators=[Length(max=100)])
|
||
call_number = StringField('来电号码', validators=[Length(max=20)])
|
||
call_number_for_dh = StringField('来电号码（脱敏）', validators=[Length(max=20)])
|
||
raw_call_numer = StringField('原始来电号码', validators=[Length(max=20)])
|
||
contact_number = StringField('联系电话', validators=[Length(max=20)])
|
||
raw_contact_number = StringField('原始联系电话', validators=[Length(max=20)])
|
||
contact_number_for_dh = StringField('联系电话（脱敏）', validators=[Length(max=20)])
|
||
call_time = DateTimeField('来电时间')
|
||
order_sound_record_id = StringField('通话记录ID', validators=[Length(max=50)])
|
||
create_date = DateTimeField('创建日期')
|
||
update_date = DateTimeField('更新日期')
|
||
plan_finish_time = DateTimeField('计划完成时间')
|
||
plan_sign_time = DateTimeField('计划签收时间')
|
||
judgment_flag = StringField('判定标志', validators=[Length(max=10)])
|
||
is_coordination = StringField('是否协调', validators=[Length(max=10)])
|
||
coordination_time = DateTimeField('协调时间')
|
||
thrid_order_id = TextAreaField('第三方工单ID')
|
||
relate_order_ids = TextAreaField('关联工单ID列表')
|
||
relate_order_count = IntegerField('关联工单数量')
|
||
order_user_id = StringField('用户ID', validators=[Length(max=50)])
|
||
user_word = TextAreaField('用户反馈')
|
||
show_flag = StringField('显示标志', validators=[Length(max=10)])
|
||
origin_show = IntegerField('原始显示标志')
|
||
order_user = TextAreaField('诉求人信息（JSON对象）')
|
||
order_phone_dto = TextAreaField('电话号码信息（JSON对象）')
|
||
order_attachment_list = TextAreaField('附件列表（JSON数组）')
|
||
pre_process_list = TextAreaField('预处理流程列表（JSON数组）')
|
||
tripartite_call_records = TextAreaField('三方通话记录（JSON对象）')
|
||
tripartite_call_records_list = TextAreaField('三方通话记录列表（JSON数组）')
|
||
order_custom_form_fields = TextAreaField('自定义表单字段（JSON数组）')
|
||
knowledge_references = TextAreaField('知识参考（JSON对象）')
|
||
sound_recording_address_list = TextAreaField('录音文件路径列表（JSON数组）')
|
||
active_dept_ids = TextAreaField('当前处理部门ID列表')
|
||
attachment_ids = TextAreaField('附件ID列表')
|
||
attachment_list = TextAreaField('附件列表JSON')
|
||
contactor_list = TextAreaField('联系人列表（JSON数组）')
|
||
tsjb_entry_info = TextAreaField('投诉举报入口信息（JSON对象）')
|
||
order_erge_revoke_plug_dto_list = TextAreaField('撤销插件信息（JSON数组）')
|
||
order_environmental = TextAreaField('环境信息（JSON对象）')
|
||
order_demands_dto = TextAreaField('诉求DTO（JSON对象）')
|
||
order_appeal_list = TextAreaField('申诉列表（JSON数组）')
|
||
torder_process_list = TextAreaField('流程列表（JSON数组）')
|
||
pre_process = TextAreaField('预处理信息（JSON对象）')
|
||
extension = TextAreaField('扩展字段')
|
||
remark = TextAreaField('备注')
|
||
file_exist = IntegerField('是否存在附件')
|
||
exist_quoto_info = TextAreaField('是否存在引用信息')
|
||
residue_date = TextAreaField('剩余天数')
|
||
whether_approval = StringField('是否审批', validators=[Length(max=10)])
|
||
over_time_warning_flag = StringField('超时预警标志', validators=[Length(max=10)])
|
||
create_no = StringField('创建编号', validators=[Length(max=20)])
|
||
return_visit_reason = TextAreaField('回访原因')
|
||
back_count = StringField('回退次数', validators=[Length(max=100)])
|
||
visit_adv_content = TextAreaField('走访建议内容')
|
||
is_dispatch_accurate = StringField('是否精准分派', validators=[Length(max=10)])
|
||
process_instance_id = StringField('流程实例ID', validators=[Length(max=100)])
|
||
knowledge_quote = TextAreaField('知识引用')
|
||
special_type = TextAreaField('特殊类型')
|
||
supervise_type = TextAreaField('监督类型')
|
||
leader_indicate = TextAreaField('领导批示')
|
||
case_solve = TextAreaField('处理结果')
|
||
result_satisfied = TextAreaField('结果满意度')
|
||
first_vist_satisfied = TextAreaField('首次走访满意度')
|
||
contact_timely = StringField('是否及时联系', validators=[Length(max=50)])
|
||
distribute_type = StringField('分派类型', validators=[Length(max=50)])
|
||
dept_type = TextAreaField('部门类型')
|
||
dept_name = TextAreaField('部门名称')
|
||
active_dept_name = StringField('当前处理部门名称', validators=[Length(max=50)])
|
||
org_id = StringField('组织ID', validators=[Length(max=50)])
|
||
org_name = TextAreaField('组织名称')
|
||
snapshot_time = DateTimeField('快照抓取时间')
|
||
|
||
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 GovsOrderDetailBase(TD3iGovsOrderDetail, CommonModel):
|
||
"""省12345工单详情业务基类"""
|
||
|
||
FieldMapping = {
|
||
# ==================== 主键与关联 ====================
|
||
'master_id': 'masterId', # 关联工单主表ID（从外部传入）
|
||
'order_id': 'orderId', # 工单编号
|
||
'order_no': 'orderNo', # 工单号
|
||
'tenant_id': 'tenantId', # 租户ID
|
||
|
||
# ==================== 工单状态 ====================
|
||
'order_status': 'orderStatus', # 工单状态码
|
||
'order_status_for_view': 'orderStatusForView', # 工单状态显示值
|
||
'first_order_status': 'firstOrderStatus', # 一级状态编码
|
||
'secord_order_status': 'secordOrderStatus', # 二级状态编码
|
||
'atomic_order_status': 'atomicOrderStatus', # 原子状态编码
|
||
'order_invalid_type': 'orderInvalidType', # 工单作废原因
|
||
'order_finish_time': 'orderFinishTime', # 工单完成时间
|
||
|
||
# ==================== 诉求内容 ====================
|
||
'case_content': 'caseContent', # 诉求内容
|
||
'case_goal': 'caseGoal', # 诉求目的
|
||
'title': 'title', # 工单标题
|
||
'case_labels': 'caseLabels', # 工单标签列表
|
||
'case_public': 'casePublic', # 是否公开
|
||
'hotspot': 'hotspot', # 是否热点事件
|
||
|
||
# ==================== 紧急与保护 ====================
|
||
'case_is_urgent': 'caseIsUrgent', # 紧急程度（一般/紧急/特急）
|
||
'case_is_visit': 'caseIsVisit', # 是否回访（是/否）
|
||
'info_protect': 'infoProtect', # 信息保护（是/否）
|
||
|
||
# ==================== 诉求归口分类 ====================
|
||
'case_accord_type_one_name': 'caseAccordTypeOneName', # 诉求归口一级
|
||
'case_accord_type_two_name': 'caseAccordTypeTwoName', # 诉求归口二级
|
||
'case_accord_type_three_name': 'caseAccordTypeThreeName', # 诉求归口三级
|
||
'case_accord_type_four_name': 'caseAccordTypeFourName', # 诉求归口四级
|
||
'case_accord_type_five_name': 'caseAccordTypeFiveName', # 诉求归口五级
|
||
'case_accord_code': 'caseAccordCode', # 事项编码
|
||
|
||
# ==================== 归属单位 ====================
|
||
'first_level_affiliation': 'firstLevelAffiliation', # 一级归属单位
|
||
'second_level_affiliation': 'secondLevelAffiliation', # 二级归属单位
|
||
'third_level_affiliation': 'thirdLevelAffiliation', # 三级归属单位
|
||
'fourth_level_affiliation': 'fourthLevelAffiliation', # 四级归属单位
|
||
'fifth_level_affiliation': 'fifthLevelAffiliation', # 五级归属单位
|
||
'sixth_level_affiliation': 'sixthLevelAffiliation', # 六级归属单位
|
||
'seventh_level_affiliation': 'seventhLevelAffiliation', # 七级归属单位
|
||
'appeal_dept': 'appealDept', # 诉求部门
|
||
|
||
# ==================== 来源与平台 ====================
|
||
'order_source': 'orderSource', # 诉求来源（电话/互联网）
|
||
'order_source_detail': 'orderSourceDetail', # 诉求来源详情（12345/随手拍）
|
||
'order_source_for_view': 'orderSourceForView', # 诉求来源显示值
|
||
'belong_platform': 'belongPlatform', # 所属平台代码
|
||
'belong_platform_name': 'belongPlatformName', # 受理平台名称
|
||
'current_processing_platform': 'currentProcessingPlatform', # 当前处理平台
|
||
|
||
# ==================== 服务对象与类型 ====================
|
||
'service_object_type': 'serviceObjectType', # 服务对象类型（投诉举报/咨询/建议等）
|
||
'order_type': 'orderType', # 表单类型（个人/企业/其他）
|
||
'form_type': 'formType', # 表单类型代码
|
||
|
||
# ==================== 区域信息 ====================
|
||
'area_code_city': 'areaCodeCity', # 市区域代码
|
||
'area_code_area': 'areaCodeArea', # 区区域代码
|
||
'area_code_street': 'areaCodeStreet', # 街道区域代码
|
||
'address_detail': 'addressDetail', # 详细地址
|
||
'case_lnglat': 'caseLnglat', # 地理坐标
|
||
|
||
# ==================== 联系方式 ====================
|
||
'call_number': 'callNumber', # 来电号码
|
||
'call_number_for_dh': 'callNumberForDH', # 来电号码（脱敏）
|
||
'raw_call_numer': 'rawCallNumer', # 原始来电号码
|
||
'contact_number': 'contactNumber', # 联系电话
|
||
'raw_contact_number': 'rawContactNumber', # 原始联系电话
|
||
'contact_number_for_dh': 'contactNumberForDH', # 联系电话（脱敏）
|
||
'call_time': 'callTime', # 来电时间
|
||
'order_sound_record_id': 'orderSoundRecordId', # 通话记录ID
|
||
|
||
# ==================== 时间节点 ====================
|
||
'create_date': 'createDate', # 创建日期
|
||
'update_date': 'updateDate', # 更新日期
|
||
'plan_finish_time': 'planFinishTime', # 计划完成时间
|
||
'plan_sign_time': 'planSignTime', # 计划签收时间
|
||
|
||
# ==================== 判定与协调 ====================
|
||
'judgment_flag': 'judgmentFlag', # 判定标志
|
||
'is_coordination': 'isCoordination', # 是否协调
|
||
'coordination_time': 'coordinationTime', # 协调时间
|
||
|
||
# ==================== 第三方关联 ====================
|
||
'thrid_order_id': 'thridOrderId', # 第三方工单ID
|
||
'relate_order_ids': 'relateOrderIds', # 关联工单ID列表
|
||
'relate_order_count': 'relateOrderCount', # 关联工单数量
|
||
|
||
# ==================== 用户相关 ====================
|
||
'order_user_id': 'orderUserId', # 用户ID（身份证号）
|
||
'user_word': 'userWord', # 用户反馈
|
||
'show_flag': 'showFlag', # 显示标志
|
||
'origin_show': 'originShow', # 原始显示标志
|
||
|
||
# ==================== JSON 对象字段（需序列化存储） ====================
|
||
'order_user': 'orderUser', # 诉求人信息（JSON对象）
|
||
'order_phone_dto': 'orderPhoneDTO', # 电话号码信息（JSON对象）
|
||
'order_attachment_list': 'orderAttachmentList', # 附件列表（JSON数组）
|
||
'pre_process_list': 'preProcessList', # 预处理流程列表（JSON数组）
|
||
'tripartite_call_records': 'tripartiteCallRecords', # 三方通话记录（JSON对象）
|
||
'tripartite_call_records_list': 'tripartiteCallRecordsList', # 三方通话记录列表（JSON数组）
|
||
'order_custom_form_fields': 'orderCustomFormFields', # 自定义表单字段（JSON数组）
|
||
'knowledge_references': 'knowledgeReferences', # 知识参考（JSON对象）
|
||
'sound_recording_address_list': 'soundRecordingAddressList', # 录音文件路径列表（JSON数组）
|
||
'active_dept_ids': 'activeDeptIds', # 当前处理部门ID列表
|
||
'attachment_ids': 'attachmentIds', # 附件ID列表
|
||
'attachment_list': 'attachmentList', # 附件列表JSON
|
||
'contactor_list': 'contactorList', # 联系人列表（JSON数组）
|
||
'tsjb_entry_info': 'tsjbEntryInfo', # 投诉举报入口信息（JSON对象）
|
||
'order_erge_revoke_plug_dto_list': 'orderErgeRevokePlugDTOList', # 撤销插件信息（JSON数组）
|
||
'order_environmental': 'orderEnvironmental', # 环境信息（JSON对象）
|
||
'order_demands_dto': 'orderDemandsDTO', # 诉求DTO（JSON对象）
|
||
'order_appeal_list': 'orderAppealList', # 申诉列表（JSON数组）
|
||
'torder_process_list': 'torderProcessList', # 流程列表（JSON数组）
|
||
'pre_process': 'preProcess', # 预处理信息（JSON对象）
|
||
'extension': 'extension', # 扩展字段
|
||
'remark': 'remark', # 备注
|
||
|
||
# ==================== 其他字段 ====================
|
||
'file_exist': None, # 是否存在附件（根据 orderAttachmentList 计算）
|
||
'exist_quoto_info': 'existQuotoInfo', # 是否存在引用信息
|
||
'residue_date': 'residueDate', # 剩余天数
|
||
'whether_approval': 'whetherApproval', # 是否审批
|
||
'over_time_warning_flag': 'overTimeWarningFlag', # 超时预警标志
|
||
'create_no': 'createNo', # 创建编号
|
||
'return_visit_reason': 'returnVisitReason', # 回访原因
|
||
'back_count': 'backCount', # 回退次数
|
||
'visit_adv_content': 'visitAdvContent', # 走访建议内容
|
||
'is_dispatch_accurate': 'isDispatchAccurate', # 是否精准分派
|
||
'process_instance_id': 'processInstanceId', # 流程实例ID
|
||
'knowledge_quote': 'knowledgeQuote', # 知识引用
|
||
'special_type': 'specialType', # 特殊类型
|
||
'supervise_type': 'superviseType', # 监督类型
|
||
'leader_indicate': 'leaderIndicate', # 领导批示
|
||
'case_solve': 'caseSolve', # 处理结果
|
||
'result_satisfied': 'resultSatisfied', # 结果满意度
|
||
'first_vist_satisfied': 'firstVistSatisfied', # 首次走访满意度
|
||
'contact_timely': 'contactTimely', # 是否及时联系
|
||
'distribute_type': 'distributeType', # 分派类型
|
||
'dept_type': 'deptType', # 部门类型
|
||
'dept_name': 'deptName', # 部门名称
|
||
'active_dept_name': 'activeDeptName', # 当前处理部门名称
|
||
'org_id': 'orgId', # 组织ID
|
||
'org_name': 'orgName', # 组织名称
|
||
}
|
||
|
||
@classmethod
|
||
async def exist_other(cls, id: Union[str, int], order_id: str = None, order_no: str = None):
|
||
"""检查是否存在除当前详情外的其他同编号工单详情"""
|
||
_query = select(cls).where(cls.id != id)
|
||
if order_id:
|
||
_query = _query.where(cls.order_id == order_id)
|
||
if order_no:
|
||
_query = _query.where(cls.order_no == order_no)
|
||
_task: cls = await cls.query_first(_query)
|
||
return _task
|
||
|
||
@classmethod
|
||
async def find_by_ids(cls, ids: list[Union[str, int]]):
|
||
"""根据ID列表批量查找工单详情"""
|
||
_query = select(cls).where(cls.id.in_(ids))
|
||
_list: list[cls] = (await cls.orm_execute_scalars(_query)).all()
|
||
return _list
|
||
|
||
@classmethod
|
||
async def is_exist(cls, order_id: str = None, order_no: str = None):
|
||
"""检查工单详情是否已经存在"""
|
||
_query = select(cls)
|
||
if order_id:
|
||
_query = _query.where(cls.order_id == order_id)
|
||
if order_no:
|
||
_query = _query.where(cls.order_no == order_no)
|
||
_task: cls = await cls.query_first(_query)
|
||
return _task
|
||
|
||
@classmethod
|
||
async def search_base(cls, is_paging=True, **kwargs):
|
||
"""按参数搜索工单详情的基础方法"""
|
||
page_number = kwargs.get('page_number', random.randint(1, 100))
|
||
page_size = kwargs.get('page_size', 20)
|
||
kwargs.update({'page_number': page_number, 'page_size': page_size})
|
||
|
||
_name_likes = {
|
||
cls.order_status.key: '%{}%',
|
||
cls.order_source.key: '%{}%',
|
||
cls.case_accord_type_one_name.key: '%{}%',
|
||
cls.belong_platform_name.key: '%{}%',
|
||
}
|
||
|
||
_query = select(cls).where(
|
||
*cls.search_wheres(likes=_name_likes, **kwargs)
|
||
).group_by(cls.id)
|
||
|
||
_paging = None
|
||
if is_paging:
|
||
_row_count = await cls.query_count(_query)
|
||
_paging = Pagination(_row_count).paging(page_number, page_size)
|
||
_data_query = _query.limit(page_size).offset(_paging.offset_size)
|
||
else:
|
||
_data_query = _query.where()
|
||
|
||
_sort_clause = cls.sort_clauses(kwargs.get('sort_clause', {}))
|
||
if _sort_clause:
|
||
_data_query = _data_query.order_by(*_sort_clause)
|
||
else:
|
||
_data_query = _data_query.order_by(cls.create_date.desc())
|
||
|
||
_df = await cls.query_as_df(_data_query)
|
||
if not _df.empty:
|
||
_df.replace(models.EmptyInDF + models.EmptyDatetimeInDF, '', inplace=True)
|
||
_df[cls.id.key] = _df[cls.id.key].astype(str)
|
||
|
||
return _df, _paging
|
||
|
||
@classmethod
|
||
async def search(cls, **kwargs):
|
||
"""按参数搜索工单详情，返回分页格式数据"""
|
||
_df, _paging = await cls.search_base(**kwargs)
|
||
return {
|
||
'total': _paging.row_count,
|
||
'rows': _df.to_dict('records'),
|
||
'pagination': {
|
||
'page_number': _paging.page_number,
|
||
'page_count': _paging.page_count,
|
||
'page_size': _paging.page_size,
|
||
},
|
||
}
|
||
|
||
@classmethod
|
||
async def exists_order_id(cls, data_df: pd.DataFrame):
|
||
"""根据 order_id 判断数据是否存在"""
|
||
if data_df.empty:
|
||
return pd.DataFrame(), pd.DataFrame()
|
||
|
||
order_ids = data_df[cls.order_id.key].unique().tolist()
|
||
if not order_ids:
|
||
return pd.DataFrame(), data_df.copy()
|
||
|
||
_query = select(cls.id, cls.order_id).where(cls.order_id.in_(order_ids))
|
||
existing_df = await cls.query_as_df(_query)
|
||
|
||
if existing_df.empty:
|
||
return pd.DataFrame(), data_df.copy()
|
||
|
||
order_id_to_id_map = dict(zip(existing_df[cls.order_id.key], existing_df[cls.id.key]))
|
||
|
||
mask_exists = data_df[cls.order_id.key].isin(existing_df[cls.order_id.key])
|
||
exists_df = data_df[mask_exists].copy()
|
||
exists_df[cls.id.key] = exists_df[cls.order_id.key].map(order_id_to_id_map)
|
||
latest_df = data_df[~mask_exists].copy()
|
||
return exists_df, latest_df
|
||
|
||
@classmethod
|
||
async def find_by_master_id(cls, master_id: Union[str, int]):
|
||
"""根据主表ID查找工单详情"""
|
||
_query = select(cls).where(cls.master_id == master_id)
|
||
return await cls.query_first(_query)
|
||
|
||
@classmethod
|
||
async def find_by_order_id(cls, order_id: str):
|
||
"""根据工单编号查找详情"""
|
||
_query = select(cls).where(cls.order_id == order_id)
|
||
return await cls.query_first(_query)
|
||
|
||
@classmethod
|
||
async def find_latest_snapshot(cls, order_id: str):
|
||
"""查找最新的快照"""
|
||
_query = select(cls).where(cls.order_id == order_id).order_by(cls.snapshot_time.desc())
|
||
return await cls.query_first(_query)
|
||
|
||
@classmethod
|
||
async def find_by_master_ids(cls, master_ids: list[Union[str, int]]):
|
||
"""根据主表ID列表批量查找工单详情"""
|
||
_query = select(cls).where(cls.master_id.in_(master_ids))
|
||
_list: list[cls] = (await cls.orm_execute_scalars(_query)).all()
|
||
return _list
|
||
|
||
|
||
@register_swagger_model
|
||
class GovsOrderDetail(GovsOrderDetailBase):
|
||
"""省12345工单详情业务类"""
|
||
|
||
@classmethod
|
||
async def create(cls, user: RbacUser = None, **kwargs):
|
||
"""创建新工单详情"""
|
||
for _k, _v in kwargs.items():
|
||
if isinstance(_v, str):
|
||
kwargs[_k] = _v.strip()
|
||
|
||
_form = GovsOrderDetailForm(formdata=kwargs)
|
||
_form.validate_form()
|
||
|
||
_existing = await cls.is_exist(
|
||
order_id=_form.order_id.data,
|
||
order_no=_form.order_no.data
|
||
)
|
||
assert _existing is None, "工单编号或工单号已存在，不能重复创建。"
|
||
|
||
_detail = cls().copy_from_dict(_form.data, skip_none=True).before_save()
|
||
_detail.snapshot_time = datetime.datetime.now()
|
||
if user:
|
||
_detail.created_by = user.username
|
||
_detail.updated_by = user.username
|
||
await _detail.async_save()
|
||
return _detail
|
||
|
||
@classmethod
|
||
async def delete(cls, detail_id: Union[str, int]):
|
||
"""删除工单详情"""
|
||
_detail: cls = await cls.async_find_by_id(detail_id)
|
||
assert _detail, f"根据 ID {detail_id} 未找到工单详情。"
|
||
|
||
_del_query = delete(cls).where(cls.id == _detail.id)
|
||
await cls.raw_execute(_del_query)
|
||
echo_log(f'已删除工单详情（工单号：{_detail.order_no}，ID：{_detail.id}）.')
|
||
return _detail
|
||
|
||
@classmethod
|
||
async def modify(cls, detail_id: Union[str, int], user: RbacUser = None, **kwargs):
|
||
"""修改工单详情信息"""
|
||
for _k, _v in kwargs.items():
|
||
if isinstance(_v, str):
|
||
kwargs[_k] = _v.strip()
|
||
|
||
_form = GovsOrderDetailForm(formdata=kwargs)
|
||
_form.validate_form()
|
||
|
||
_other = await cls.exist_other(
|
||
detail_id,
|
||
order_id=_form.order_id.data,
|
||
order_no=_form.order_no.data
|
||
)
|
||
assert _other is None, "工单编号或工单号已存在，不能重复修改。"
|
||
|
||
_detail: cls = await cls.async_find_by_id(detail_id)
|
||
assert _detail, f'查无此工单详情信息。'
|
||
|
||
_detail.copy_from_dict(_form.data, skip_none=True).before_save()
|
||
if user:
|
||
_detail.updated_by = user.username
|
||
await _detail.async_save()
|
||
return _detail
|
||
|
||
@classmethod
|
||
async def create_batch(cls, data_df: pd.DataFrame, user: RbacUser = None):
|
||
"""批量创建工单详情"""
|
||
if data_df.empty:
|
||
return 0
|
||
|
||
if user:
|
||
data_df['created_by'] = user.username
|
||
data_df['updated_by'] = user.username
|
||
|
||
records = data_df.to_dict('records')
|
||
details = [cls().copy_from_dict(record, skip_none=True).before_save() for record in records]
|
||
|
||
session = cls.get_aio_session()
|
||
try:
|
||
session.add_all(details)
|
||
await session.commit()
|
||
except Exception as e:
|
||
await session.rollback()
|
||
raise e
|
||
finally:
|
||
await session.close()
|
||
echo_log(f"批量创建成功：创建 {len(details)} 条工单详情。")
|
||
return len(details)
|
||
|
||
@classmethod
|
||
async def modify_batch(cls, data_df: pd.DataFrame, user: RbacUser = None):
|
||
"""批量修改工单详情"""
|
||
if data_df.empty:
|
||
return 0
|
||
|
||
if 'id' not in data_df.columns:
|
||
echo_log(f"错误：modify_batch 要求输入数据必须包含 '{cls.id.key}' 列")
|
||
return 0
|
||
|
||
data_df['updated_at'] = datetime.datetime.now()
|
||
if user:
|
||
data_df['updated_by'] = user.username
|
||
|
||
update_data = data_df.to_dict('records')
|
||
|
||
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, user: RbacUser = None):
|
||
"""批量保存数据，自动处理新建和更新"""
|
||
_exists_df, _latest_df = await cls.exists_order_id(data_df)
|
||
_created_count = await cls.create_batch(_latest_df, user)
|
||
_updated_count = await cls.modify_batch(_exists_df, user)
|
||
return _created_count, _updated_count
|
||
|
||
@classmethod
|
||
async def create_or_update(cls, user: RbacUser = None, **kwargs):
|
||
"""创建或更新工单详情（单条）"""
|
||
for _k, _v in kwargs.items():
|
||
if isinstance(_v, str):
|
||
kwargs[_k] = _v.strip()
|
||
|
||
_form = GovsOrderDetailForm(formdata=kwargs)
|
||
_form.validate_form()
|
||
|
||
_existing = await cls.find_by_order_id(_form.order_id.data)
|
||
if _existing:
|
||
_existing.copy_from_dict(_form.data, skip_none=True).before_save()
|
||
_existing.snapshot_time = datetime.datetime.now()
|
||
if user:
|
||
_existing.updated_by = user.username
|
||
await _existing.async_save()
|
||
return _existing
|
||
|
||
_detail = cls().copy_from_dict(_form.data, skip_none=True).before_save()
|
||
_detail.snapshot_time = datetime.datetime.now()
|
||
if user:
|
||
_detail.created_by = user.username
|
||
_detail.updated_by = user.username
|
||
await _detail.async_save()
|
||
return _detail |