Skip to content

H
hh_ccs
提交 b1610f34 authored 作者: 刘擎阳's avatar 刘擎阳
浏览文件
操作

1.优化bug

上级 2127e306
显示空白字符变更
内嵌 并排
正在显示 包含 340 行增加 和 120 行删除
ccs_connect_tiktok/data/timer.xml
浏览文件 @ b1610f34
......@@ -30,11 +30,12 @@
<field name="model_id" ref="ccs_connect_tiktok.model_warning_config"/>
<field name="state">code</field>
<field name="code">model.cron_warn_cc_order()</field>
<field name='interval_number'>1</field>
<field name='interval_type'>days</field>
<field name='interval_number'>30</field>
<field name='interval_type'>minutes</field>
<field name="numbercall">-1</field>
<field name="active" eval="True"/>
<field name="nextcall" eval="(DateTime.now().replace(hour=0, minute=0)).strftime('%Y-%m-%d %H:%M:%S')" />
<field name="nextcall"
eval="(datetime.now() + timedelta(minutes=5)).strftime('%Y-%m-%d %H:%M:%S')"/>
<field name="doall" eval="False"/>
</record>
......
ccs_connect_tiktok/models/warn_config.py
浏览文件 @ b1610f34
......@@ -281,123 +281,123 @@ class WarningConfig(models.Model):
# # print(bl_arr)
# self.send_warn_email(self.format_email_content_grouped(bl_arr))
def cron_warn_cc_order(self):
# 1. 初始化基础参数
utc_time = datetime.utcnow()
warn_days = int(self.env['ir.config_parameter'].sudo().get_param('warn_order_days', 7))
# c_time = utc_time - timedelta(days=warn_days)
end_date = fields.Date.today()
start_date = end_date - timedelta(days=warn_days)
config_objs = self.env['warning.config'].sudo().search([])
bl_objs = self.env['cc.bl'].sudo().search([
('bl_date', '>=', start_date),
('state', '!=', 'done')
])
# def cron_warn_cc_order(self):
# # 1. 初始化基础参数
# utc_time = datetime.utcnow()
# warn_days = int(self.env['ir.config_parameter'].sudo().get_param('warn_order_days', 7))
# # c_time = utc_time - timedelta(days=warn_days)
# end_date = fields.Date.today()
# start_date = end_date - timedelta(days=warn_days)
#
# config_objs = self.env['warning.config'].sudo().search([])
# bl_objs = self.env['cc.bl'].sudo().search([
# ('id', '=', 71)
# ('bl_date', '>=', start_date),
# ('state', '!=', 'done')
# ])
if not config_objs or not bl_objs:
return
# 2. 获取包裹及预计算提单的 ETA (避免重复转换时间)
all_packages = self.env['cc.ship.package'].sudo().search([('bl_id', 'in', bl_objs.ids)])
packages_by_bl = defaultdict(list)
for pkg in all_packages:
packages_by_bl[pkg.bl_id.id].append(pkg)
bl_eta_utc_dict = {bl.id: self.convert_to_utc(bl.eta) if bl.eta else None for bl in bl_objs}
# 3. 核心优化：提前收集所有配置中用到的 process_code，一次性查询！
bl_codes_needed = set()
pkg_codes_needed = set()
for config in config_objs:
is_clearance = (config.time_type == 'clearance_node')
node_type = config.time_point_id.node_type if is_clearance else config.unsynced_node_id.node_type
if node_type == 'bl':
if is_clearance: bl_codes_needed.add(config.time_point_id.tk_code)
bl_codes_needed.add(config.unsynced_node_id.tk_code)
elif node_type == 'package':
if is_clearance: pkg_codes_needed.add(config.time_point_id.tk_code)
pkg_codes_needed.add(config.unsynced_node_id.tk_code)
# 4. 超级工具函数：使用 search_read 避开 ORM 实例化开销
def get_all_logs_dict(model_name, rel_field, record_ids, process_codes):
if not record_ids or not process_codes:
return {}
# search_read 直接返回字典列表，比 search 返回对象快 10 倍以上
logs_data = self.env[model_name].sudo().search_read(
[(rel_field, 'in', tuple(record_ids)), ('process_code', 'in', tuple(process_codes))],
[rel_field, 'process_code', 'operate_time'],
order='operate_time desc'
)
log_dict = defaultdict(dict)
for data in logs_data:
# search_read 中的 Many2one 字段会返回 (id, name) 元组
rec_id = data[rel_field][0] if isinstance(data[rel_field], tuple) else data[rel_field]
code = data['process_code']
if code not in log_dict[rec_id]: # 依靠 order='desc'，最先遍历到的一定是最新的
log_dict[rec_id][code] = data['operate_time']
return log_dict
# 无论有多少条配置规则，查提单日志和包裹日志永远只各查 1 次数据库！
bl_logs_dict = get_all_logs_dict('cc.bl.sync.log', 'bl_id', bl_objs.ids, bl_codes_needed)
pkg_logs_dict = get_all_logs_dict('cc.ship.package.sync.log', 'package_id', all_packages.ids, pkg_codes_needed)
ship_package_arr = defaultdict(list)
bl_arr = defaultdict(list)
# 5. 遍历配置规则进行校验 (此时全是纯内存/字典操作，耗时接近 0)
for config in config_objs:
is_clearance = (config.time_type == 'clearance_node')
node_type = config.time_point_id.node_type if is_clearance else config.unsynced_node_id.node_type
sync_code = config.time_point_id.tk_code if is_clearance else None
unsync_code = config.unsynced_node_id.tk_code
time_offset = timedelta(hours=-int(config.remaining_time or 0))
node_name = config.unsynced_node_id.name or ''
warning_reason = config.warning_reason or ''
if node_type == 'bl':
for bl in bl_objs:
logs = bl_logs_dict.get(bl.id, {})
if unsync_code in logs: # 已同步，跳过
continue
base_time = logs.get(sync_code) if is_clearance else bl_eta_utc_dict.get(bl.id)
if base_time and utc_time > (base_time + time_offset):
bl_arr[bl.bl_no].append({
'name': node_name,
'warning_reason': warning_reason
})
elif node_type == 'package':
for bl in bl_objs:
alert_pkgs = []
# 获取该提单预计算的基准时间 (非clearance模式下)
bl_eta = bl_eta_utc_dict.get(bl.id) if not is_clearance else None
for pkg in packages_by_bl.get(bl.id, []):
logs = pkg_logs_dict.get(pkg.id, {})
if unsync_code in logs:
continue
base_time = logs.get(sync_code) if is_clearance else bl_eta
if base_time and utc_time > (base_time + time_offset):
alert_pkgs.append(pkg)
if alert_pkgs:
ship_package_arr[bl.bl_no].append({
'name': node_name,
'arr': alert_pkgs,
'warning_reason': warning_reason
})
# 6. 发送邮件汇总
if ship_package_arr:
self.send_warn_email(self.format_package_warning_email(ship_package_arr))
if bl_arr:
self.send_warn_email(self.format_email_content_grouped(bl_arr))
# # bl_objs = self.env['cc.bl'].sudo().search([
# # ('id', '=', 71)
# # ])
# if not config_objs or not bl_objs:
# return
# # 2. 获取包裹及预计算提单的 ETA (避免重复转换时间)
# all_packages = self.env['cc.ship.package'].sudo().search([('bl_id', 'in', bl_objs.ids)])
# packages_by_bl = defaultdict(list)
# for pkg in all_packages:
# packages_by_bl[pkg.bl_id.id].append(pkg)
# bl_eta_utc_dict = {bl.id: self.convert_to_utc(bl.eta) if bl.eta else None for bl in bl_objs}
# # 3. 核心优化：提前收集所有配置中用到的 process_code，一次性查询！
# bl_codes_needed = set()
# pkg_codes_needed = set()
# for config in config_objs:
# is_clearance = (config.time_type == 'clearance_node')
# node_type = config.time_point_id.node_type if is_clearance else config.unsynced_node_id.node_type
# if node_type == 'bl':
# if is_clearance: bl_codes_needed.add(config.time_point_id.tk_code)
# bl_codes_needed.add(config.unsynced_node_id.tk_code)
# elif node_type == 'package':
# if is_clearance: pkg_codes_needed.add(config.time_point_id.tk_code)
# pkg_codes_needed.add(config.unsynced_node_id.tk_code)
#
# # 4. 超级工具函数：使用 search_read 避开 ORM 实例化开销
# def get_all_logs_dict(model_name, rel_field, record_ids, process_codes):
# if not record_ids or not process_codes:
# return {}
# # search_read 直接返回字典列表，比 search 返回对象快 10 倍以上
# logs_data = self.env[model_name].sudo().search_read(
# [(rel_field, 'in', tuple(record_ids)), ('process_code', 'in', tuple(process_codes))],
# [rel_field, 'process_code', 'operate_time'],
# order='operate_time desc'
# )
# log_dict = defaultdict(dict)
# for data in logs_data:
# # search_read 中的 Many2one 字段会返回 (id, name) 元组
# rec_id = data[rel_field][0] if isinstance(data[rel_field], tuple) else data[rel_field]
# code = data['process_code']
# if code not in log_dict[rec_id]: # 依靠 order='desc'，最先遍历到的一定是最新的
# log_dict[rec_id][code] = data['operate_time']
# return log_dict
#
# # 无论有多少条配置规则，查提单日志和包裹日志永远只各查 1 次数据库！
# bl_logs_dict = get_all_logs_dict('cc.bl.sync.log', 'bl_id', bl_objs.ids, bl_codes_needed)
# pkg_logs_dict = get_all_logs_dict('cc.ship.package.sync.log', 'package_id', all_packages.ids, pkg_codes_needed)
#
# ship_package_arr = defaultdict(list)
# bl_arr = defaultdict(list)
#
# # 5. 遍历配置规则进行校验 (此时全是纯内存/字典操作，耗时接近 0)
# for config in config_objs:
# is_clearance = (config.time_type == 'clearance_node')
# node_type = config.time_point_id.node_type if is_clearance else config.unsynced_node_id.node_type
#
# sync_code = config.time_point_id.tk_code if is_clearance else None
# unsync_code = config.unsynced_node_id.tk_code
#
# time_offset = timedelta(hours=-int(config.remaining_time or 0))
# node_name = config.unsynced_node_id.name or ''
# warning_reason = config.warning_reason or ''
#
# if node_type == 'bl':
# for bl in bl_objs:
# logs = bl_logs_dict.get(bl.id, {})
# if unsync_code in logs: # 已同步，跳过
# continue
#
# base_time = logs.get(sync_code) if is_clearance else bl_eta_utc_dict.get(bl.id)
#
# if base_time and utc_time > (base_time + time_offset):
# bl_arr[bl.bl_no].append({
# 'name': node_name,
# 'warning_reason': warning_reason
# })
#
# elif node_type == 'package':
# for bl in bl_objs:
# alert_pkgs = []
# # 获取该提单预计算的基准时间 (非clearance模式下)
# bl_eta = bl_eta_utc_dict.get(bl.id) if not is_clearance else None
#
# for pkg in packages_by_bl.get(bl.id, []):
# logs = pkg_logs_dict.get(pkg.id, {})
# if unsync_code in logs:
# continue
#
# base_time = logs.get(sync_code) if is_clearance else bl_eta
#
# if base_time and utc_time > (base_time + time_offset):
# alert_pkgs.append(pkg)
#
# if alert_pkgs:
# ship_package_arr[bl.bl_no].append({
# 'name': node_name,
# 'arr': alert_pkgs,
# 'warning_reason': warning_reason
# })
#
# # 6. 发送邮件汇总
# if ship_package_arr:
# self.send_warn_email(self.format_package_warning_email(ship_package_arr))
# if bl_arr:
# self.send_warn_email(self.format_email_content_grouped(bl_arr))
def format_package_warning_email(self, data):
"""
......@@ -518,6 +518,225 @@ class WarningConfig(models.Model):
# except Exception as e:
# _logger.error(f"发送预警邮件失败: {str(e)}")
def cron_warn_cc_order(self):
# 1. 初始化基础参数
utc_time = datetime.utcnow()
warn_days = int(self.env['ir.config_parameter'].sudo().get_param('warn_order_days', 7))
end_date = fields.Date.today()
start_date = end_date - timedelta(days=warn_days)
config_objs = self.env['warning.config'].sudo().search([])
bl_objs = self.env['cc.bl'].sudo().search([
('bl_date', '>=', start_date),
('state', '!=', 'done')
])
bl_objs = self.env['cc.bl'].sudo().search([
('id', '=', 71)
])
if not config_objs or not bl_objs:
return
# 2. 获取包裹及预计算提单的 ETA
all_packages = self.env['cc.ship.package'].sudo().search([('bl_id', 'in', bl_objs.ids)])
packages_by_bl = defaultdict(list)
for pkg in all_packages:
packages_by_bl[pkg.bl_id.id].append(pkg)
bl_eta_utc_dict = {bl.id: self.convert_to_utc(bl.eta) if bl.eta else None for bl in bl_objs}
# 3. 核心优化：彻底解耦时间节点和未同步节点的类型，分别收集 process_code！
bl_codes_needed = set()
pkg_codes_needed = set()
for config in config_objs:
is_clearance = (config.time_type == 'clearance_node')
# A. 收集未同步节点 (决定了我们要检查谁)
if config.unsynced_node_id.node_type == 'bl':
bl_codes_needed.add(config.unsynced_node_id.tk_code)
elif config.unsynced_node_id.node_type == 'package':
pkg_codes_needed.add(config.unsynced_node_id.tk_code)
# B. 收集预警时间点 (决定了基准时间来自谁)
if is_clearance and config.time_point_id:
if config.time_point_id.node_type == 'bl':
bl_codes_needed.add(config.time_point_id.tk_code)
elif config.time_point_id.node_type == 'package':
pkg_codes_needed.add(config.time_point_id.tk_code)
# 4. 超级工具函数：使用 search_read 避开 ORM 实例化开销
def get_all_logs_dict(model_name, rel_field, record_ids, process_codes):
if not record_ids or not process_codes:
return {}
logs_data = self.env[model_name].sudo().search_read(
[(rel_field, 'in', tuple(record_ids)), ('process_code', 'in', tuple(process_codes))],
[rel_field, 'process_code', 'operate_time'],
order='operate_time desc'
)
log_dict = defaultdict(dict)
for data in logs_data:
rec_id = data[rel_field][0] if isinstance(data[rel_field], tuple) else data[rel_field]
code = data['process_code']
if code not in log_dict[rec_id]:
log_dict[rec_id][code] = data['operate_time']
return log_dict
bl_logs_dict = get_all_logs_dict('cc.bl.sync.log', 'bl_id', bl_objs.ids, bl_codes_needed)
pkg_logs_dict = get_all_logs_dict('cc.ship.package.sync.log', 'package_id', all_packages.ids, pkg_codes_needed)
# 5. 统一数据结构：按提单维度合并所有预警
# 数据格式: { bl_id: {'bl': bl_record, 'bl_warnings': [...], 'pkg_warnings': [...]} }
warnings_by_bl = defaultdict(lambda: {
'bl_record': None,
'bl_warnings': [],
'pkg_warnings': []
})
# 6. 遍历配置规则进行校验 (纯内存/字典操作)
for config in config_objs:
is_clearance = (config.time_type == 'clearance_node')
target_node_type = config.unsynced_node_id.node_type # 预警主体(查提单还是查包裹)
time_node_type = config.time_point_id.node_type if is_clearance else None # 基准时间来源
unsync_code = config.unsynced_node_id.tk_code
sync_code = config.time_point_id.tk_code if is_clearance else None
time_offset = timedelta(hours=-int(config.remaining_time or 0))
node_name = config.unsynced_node_id.name or ''
warning_reason = config.warning_reason or ''
for bl in bl_objs:
warnings_by_bl[bl.id]['bl_record'] = bl
bl_log = bl_logs_dict.get(bl.id, {})
# 统一获取来自提单层级的基准时间 (航班落地 或 提单节点的清关时间)
if not is_clearance:
bl_base_time = bl_eta_utc_dict.get(bl.id)
elif time_node_type == 'bl':
bl_base_time = bl_log.get(sync_code)
else:
bl_base_time = None
# --- 情景 A: 检查提单自身预警 ---
if target_node_type == 'bl':
if unsync_code in bl_log: # 该提单已同步目标节点，跳过
continue
if bl_base_time and utc_time > (bl_base_time + time_offset):
warnings_by_bl[bl.id]['bl_warnings'].append({
'name': node_name,
'warning_reason': warning_reason
})
# --- 情景 B: 检查小包预警 (重头戏) ---
elif target_node_type == 'package':
alert_pkgs = []
for pkg in packages_by_bl.get(bl.id, []):
pkg_log = pkg_logs_dict.get(pkg.id, {})
if unsync_code in pkg_log: # 该小包已同步目标节点，跳过
continue
# 决定该小包的基准时间：如果是以小包节点计算，取小包日志；否则取上方的提单基准时间
if is_clearance and time_node_type == 'package':
base_time = pkg_log.get(sync_code)
else:
base_time = bl_base_time
if base_time and utc_time > (base_time + time_offset):
alert_pkgs.append(pkg)
if alert_pkgs:
warnings_by_bl[bl.id]['pkg_warnings'].append({
'name': node_name,
'arr': alert_pkgs,
'warning_reason': warning_reason
})
# 7. 清理掉没有产生任何预警的提单空壳数据
final_warnings = {
bl_id: data for bl_id, data in warnings_by_bl.items()
if data['bl_warnings'] or data['pkg_warnings']
}
# 8. 发送邮件汇总 (统一入口)
if final_warnings:
# 你需要根据这个新的字典结构，重写/新增一个统一的格式化邮件函数
email_content = self.format_combined_warning_email(final_warnings)
self.send_warn_email(email_content)
# def format_combined_warning_email(self, combined_warnings):
# """
# 接收 combined_warnings 字典，格式化为HTML邮件正文
# """
# html = "<h3>清关异常预警汇总</h3>"
#
# for bl_id, data in combined_warnings.items():
# bl = data['bl_record']
# html += f"<hr/><h4>提单号: {bl.bl_no}</h4>"
#
# # 1. 拼接提单自身预警
# if data['bl_warnings']:
# html += "<b>【提单维度预警】:</b><ul>"
# for w in data['bl_warnings']:
# html += f"<li>节点: {w['name']}, 原因: {w['warning_reason']}</li>"
# html += "</ul>"
#
# # 2. 拼接该提单下的小包预警
# if data['pkg_warnings']:
# html += "<b>【小包维度预警】:</b><ul>"
# for w in data['pkg_warnings']:
# pkg_nos = ", ".join([p.logistic_order_no for p in w['arr']]) # 假设包裹号字段是 name
# html += f"<li>节点: {w['name']}, 原因: {w['warning_reason']}<br/>"
# html += f"异常包裹: {pkg_nos}</li>"
# html += "</ul>"
#
# return html
def format_combined_warning_email(self, combined_warnings):
"""
按照最新文本格式要求格式化合并后的预警邮件
"""
html_content = ""
for bl_id, data in combined_warnings.items():
bl = data['bl_record']
# 提单号头部
html_content += f"提单号{bl.bl_no}<br/>"
# 统一序号计数器
counter = 1
# 1. 拼接提单自身预警
for w in data['bl_warnings']:
node_name = w['name'] or '未知节点'
reason = w['warning_reason'] or '无'
# 格式: 1.关务节点xxxx未同步，xxxxx
html_content += f"{counter}.关务节点{node_name}未同步，{reason};<br/>"
counter += 1
# 2. 拼接该提单下的小包预警
for w in data['pkg_warnings']:
node_name = w['name'] or '未知节点'
reason = w['warning_reason'] or '无'
pkgs = w.get('arr', [])
total_count = len(pkgs)
if total_count > 0:
# 截取前10个小包对象，并提取它们的名字(单号)
display_pkg_names = [p.logistic_order_no for p in pkgs[:10]]
# 用 '/' 将单号拼接起来
pkg_str = "/".join(display_pkg_names)
# 格式: 3.小包xxxx/xxx/xxxx等xxx个小包，节点xxxx未同步，xxxxx
html_content += f"{counter}.小包{pkg_str}等{total_count}个小包，节点{node_name}未同步，{reason};<br/>"
counter += 1
# 如果有多个提单，在下一个提单前额外加一个空行，保持排版清爽
html_content += "<br/>"
# 3. 邮件最末尾统一加上提示语（无需加粗，纯文本风格）
html_content += "请及时操作！"
return html_content
ccs_connect_tiktok/views/warn_config_views.xml
浏览文件 @ b1610f34
......@@ -19,14 +19,14 @@
attrs="{
'invisible': [('time_type', '=', 'flight_landing')],
'required': [('time_type', '=', 'clearance_node')]
}" context="{'show_code_in_name': True}"/>
}" context="{'show_code_in_name': True}" options='{"always_reload": True}'/>
<field name="flight_landing_time"
attrs="{'invisible': [('time_type', '=', 'clearance_node')]}"/>
<field name="remaining_time"/>
</group>
<group>
<field name="unsynced_node_id" context="{'show_code_in_name': True}"/>
<field name="unsynced_node_id" context="{'show_code_in_name': True}" options='{"always_reload": True}'/>
<field name="active" invisible="1"/>
</group>
</group>
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论