修改模型更新脚本

ad/ad_monitor_util.py

@@ -69,7 +69,7 @@ def _monitor(level, msg: str, start, elapsed, top10):
                f"\n- 任务开始时间: {timestamp_format(start)}" \
                f"\n- 任务状态: {level_task_status_map[level]}" \
                f"\n- 任务耗时: {seconds_convert(elapsed)}" \
-               f"\n- 任务描述: {msg}"
+               f"\n- 任务描述:\n{msg}"
     card_json = {
         "schema": "2.0",
         "header": {

ad/pai_flow_operator_v3.py

@@ -367,17 +367,41 @@ class PAIClient:
     @staticmethod
     def get_flow_out_put(pipeline_run_id: str, node_id: str, depth: int):
         client = PAIClient.create_flow_client()
-        list_pipeline_run_node_outputs_request = paiflow_20210202_models.ListPipelineRunNodeOutputsRequest(
-            depth=depth
-        )
         runtime = util_models.RuntimeOptions()
         headers = {}
+        page_number = 1
+        page_size = 100
+        all_outputs = []
+        total_count = None
+        request_id = None
         try:
-            # 复制代码运行请自行打印 API 的返回值
-            resp = client.list_pipeline_run_node_outputs_with_options(pipeline_run_id, node_id,
-                                                                      list_pipeline_run_node_outputs_request, headers,
-                                                                      runtime)
-            return resp.body.to_map()
+            while True:
+                list_pipeline_run_node_outputs_request = paiflow_20210202_models.ListPipelineRunNodeOutputsRequest(
+                    depth=depth,
+                    page_number=page_number,
+                    page_size=page_size,
+                )
+                resp = client.list_pipeline_run_node_outputs_with_options(
+                    pipeline_run_id, node_id,
+                    list_pipeline_run_node_outputs_request, headers, runtime
+                )
+                body = resp.body.to_map()
+                request_id = body.get('RequestId', request_id)
+                total_count = body.get('TotalCount', total_count)
+                outputs = body.get('Outputs') or []
+                all_outputs.extend(outputs)
+                if not outputs:
+                    break
+                if total_count is not None and len(all_outputs) >= int(total_count):
+                    break
+                if len(outputs) < page_size:
+                    break
+                page_number += 1
+            return {
+                'Outputs': all_outputs,
+                'TotalCount': total_count if total_count is not None else len(all_outputs),
+                'RequestId': request_id,
+            }
         except Exception as error:
             # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
             # 错误 message
@@ -636,43 +660,81 @@ def validate_model_data_accuracy():
         table_dict = {}
         node_dict = get_node_dict()
         job_dict = get_job_dict()
+        if '虚拟起始节点' not in job_dict:
+            raise Exception(f"未找到今日成功的虚拟起始节点任务，当前 job_dict={job_dict}")
         job_id = job_dict['虚拟起始节点']
         validate_job_detail = wait_job_end(job_id)
-        if validate_job_detail['Status'] == 'Succeeded':
-            pipeline_run_id = validate_job_detail['RunId']
-            node_id = validate_job_detail['PaiflowNodeId']
-            flow_out_put_detail = PAIClient.get_flow_out_put(pipeline_run_id, node_id, 3)
-            print(flow_out_put_detail)
-            outputs = flow_out_put_detail['Outputs']
-            for output in outputs:
-                if output["Producer"] == node_dict['二分类评估-扫码1'] and output["Name"] == "outputMetricTable":
-                    value1 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-扫码1'] = value1['location']['table']
-                if output["Producer"] == node_dict['二分类评估-加微1'] and output["Name"] == "outputMetricTable":
-                    value2 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-加微1'] = value2['location']['table']
-                if output["Producer"] == node_dict['二分类评估-转化1'] and output["Name"] == "outputMetricTable":
-                    value3 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-转化1'] = value3['location']['table']
-                if output["Producer"] == node_dict['二分类评估-扫码2'] and output["Name"] == "outputMetricTable":
-                    value4 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-扫码2'] = value4['location']['table']
-                if output["Producer"] == node_dict['二分类评估-加微2'] and output["Name"] == "outputMetricTable":
-                    value5 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-加微2'] = value5['location']['table']
-                if output["Producer"] == node_dict['二分类评估-转化2'] and output["Name"] == "outputMetricTable":
-                    value6 = json.loads(output["Info"]['value'])
-                    table_dict['二分类评估-转化2'] = value6['location']['table']
-
-                if output["Producer"] == node_dict['预测结果对比-扫码'] and output["Name"] == "outputTable":
-                    value7 = json.loads(output["Info"]['value'])
-                    table_dict['预测结果对比-扫码'] = value7['location']['table']
-                if output["Producer"] == node_dict['预测结果对比-加微'] and output["Name"] == "outputTable":
-                    value8 = json.loads(output["Info"]['value'])
-                    table_dict['预测结果对比-加微'] = value8['location']['table']
-                if output["Producer"] == node_dict['预测结果对比-转化'] and output["Name"] == "outputTable":
-                    value9 = json.loads(output["Info"]['value'])
-                    table_dict['预测结果对比-转化'] = value9['location']['table']
+        if validate_job_detail['Status'] != 'Succeeded':
+            raise Exception(f"虚拟起始节点任务未成功，status={validate_job_detail['Status']}")
+
+        pipeline_run_id = validate_job_detail['RunId']
+        node_id = validate_job_detail['PaiflowNodeId']
+        # 扫码/加微/转化三路评估与对比节点共 5 层
+        flow_out_put_detail = PAIClient.get_flow_out_put(pipeline_run_id, node_id, 5)
+        if not flow_out_put_detail:
+            raise Exception('获取工作流输出失败，flow_out_put_detail 为空')
+        print(flow_out_put_detail)
+        outputs = flow_out_put_detail.get('Outputs') or []
+        metric_names = {
+            '二分类评估-扫码1', '二分类评估-加微1', '二分类评估-转化1',
+            '二分类评估-扫码2', '二分类评估-加微2', '二分类评估-转化2',
+        }
+        compare_names = {
+            '预测结果对比-扫码', '预测结果对比-加微', '预测结果对比-转化',
+        }
+        required_names = metric_names | compare_names
+        producer_to_name = {node_dict[name]: name for name in required_names if name in node_dict}
+        print(f'node_dict keys: {list(node_dict.keys())}')
+        print(f'producer_to_name: {producer_to_name}')
+        for output in outputs:
+            producer = output.get('Producer')
+            name = producer_to_name.get(producer)
+            if not name:
+                continue
+            output_name = output.get('Name')
+            if name in metric_names and output_name == 'outputMetricTable':
+                value = json.loads(output['Info']['value'])
+                table_dict[name] = value['location']['table']
+            elif name in compare_names and output_name == 'outputTable':
+                value = json.loads(output['Info']['value'])
+                table_dict[name] = value['location']['table']
+        print(f'table_dict: {table_dict}')
+
+        # 兜底：起始节点 depth 仍可能漏表，按各节点自身 Job 再取一次输出
+        for name in required_names:
+            if name in table_dict or name not in job_dict or name not in node_dict:
+                continue
+            node_job_detail = PAIClient.get_job_detail(job_dict[name])
+            if node_job_detail.get('Status') != 'Succeeded':
+                continue
+            node_outputs = PAIClient.get_flow_out_put(
+                node_job_detail['RunId'], node_job_detail['PaiflowNodeId'], 1
+            ).get('Outputs') or []
+            for output in node_outputs:
+                if output.get('Producer') != node_dict[name]:
+                    continue
+                output_name = output.get('Name')
+                if name in metric_names and output_name == 'outputMetricTable':
+                    value = json.loads(output['Info']['value'])
+                    table_dict[name] = value['location']['table']
+                elif name in compare_names and output_name == 'outputTable':
+                    value = json.loads(output['Info']['value'])
+                    table_dict[name] = value['location']['table']
+        print(f'table_dict after fallback: {table_dict}')
+
+        required_tables = [
+            '预测结果对比-扫码', '预测结果对比-加微', '预测结果对比-转化',
+            '二分类评估-扫码1', '二分类评估-加微1', '二分类评估-转化1',
+            '二分类评估-扫码2', '二分类评估-加微2', '二分类评估-转化2',
+        ]
+        missing_tables = [k for k in required_tables if k not in table_dict]
+        if missing_tables:
+            missing_nodes = [k for k in required_tables if k not in node_dict]
+            raise Exception(
+                f"未获取到评估产物表: {missing_tables}；"
+                f"工作流中缺失节点: {missing_nodes}；"
+                f"当前 table_dict={table_dict}"
+            )
 
         num = 10
         bizdate = get_previous_days_date(1)
@@ -688,24 +750,36 @@ def validate_model_data_accuracy():
 
         for name, compare_key, new_eval_key, old_eval_key in categories:
             df = get_data_from_odps('pai_algo', table_dict[compare_key], num)
+            if df is None:
+                raise Exception(f'【{name}】预测结果对比表数据不足 {num} 条: {table_dict[compare_key]}')
             old_abs_avg = df['old_error'].abs().sum() / num
             new_abs_avg = df['new_error'].abs().sum() / num
-            new_auc = get_dict_from_odps('pai_algo', table_dict[new_eval_key])['AUC']
-            old_auc = get_dict_from_odps('pai_algo', table_dict[old_eval_key])['AUC']
+            new_auc_dict = get_dict_from_odps('pai_algo', table_dict[new_eval_key])
+            old_auc_dict = get_dict_from_odps('pai_algo', table_dict[old_eval_key])
+            if not new_auc_dict or not old_auc_dict:
+                raise Exception(f'【{name}】评估指标表读取失败: new={table_dict[new_eval_key]}, old={table_dict[old_eval_key]}')
+            new_auc = float(new_auc_dict['AUC'])
+            old_auc = float(old_auc_dict['AUC'])
 
             if name in ('扫码', '加微') and old_auc - new_auc > auc_decline_threshold:
                 auc_check_passed = False
 
-            msg += f'\n【{name}】'
-            msg += f'\n\t - 老模型AUC: {old_auc}'
-            msg += f'\n\t - 新模型AUC: {new_auc}'
-            msg += f'\n\t - 老模型Top10差异平均值: {old_abs_avg}'
-            msg += f'\n\t - 新模型Top10差异平均值: {new_abs_avg}'
+            msg += (
+                f'\n\n**{name}**'
+                f'\n- 老模型AUC: {old_auc:.6f}'
+                f'\n- 新模型AUC: {new_auc:.6f}'
+                f'\n- 老模型Top10差异平均值: {old_abs_avg:.6f}'
+                f'\n- 新模型Top10差异平均值: {new_abs_avg:.6f}'
+            )
 
-            section_msg = f'\n### {name}\n| CID  | 老模型相对真实CTCVR的变化 | 新模型相对真实CTCVR的变化 |'
-            section_msg += '\n| ---- | --------- | -------- |'
+            section_msg = f'\n### {name}\n| CID | 老模型相对真实CTCVR的变化 | 新模型相对真实CTCVR的变化 |'
+            section_msg += '\n| --- | --- | --- |'
             for _, row in df.iterrows():
-                section_msg += f"\n| {int(row['cid'])} | {row['old_error']} | {row['new_error']} |"
+                section_msg += (
+                    f"\n| {int(row['cid'])} "
+                    f"| {float(row['old_error']):.6f} "
+                    f"| {float(row['new_error']):.6f} |"
+                )
             top10_msg += section_msg
             print(section_msg)